diff --git a/Emby.Server.Implementations/QuickConnect/QuickConnectManager.cs b/Emby.Server.Implementations/QuickConnect/QuickConnectManager.cs index 855205e4aa..5ca0c7de73 100644 --- a/Emby.Server.Implementations/QuickConnect/QuickConnectManager.cs +++ b/Emby.Server.Implementations/QuickConnect/QuickConnectManager.cs @@ -147,6 +147,12 @@ namespace Emby.Server.Implementations.QuickConnect // Change the time on the request so it expires one minute into the future. It can't expire immediately as otherwise some clients wouldn't ever see that they have been authenticated. result.DateAdded = DateTime.UtcNow.Add(TimeSpan.FromMinutes(1)); + // The guard above is a read on shared state, so it cannot settle a race between instances; the claim can. + if (!await _store.TryClaimAuthorizationAsync(result.Secret, ExpiryOf(result)).ConfigureAwait(false)) + { + throw new InvalidOperationException("Request is already authorized"); + } + var authenticationResult = await _sessionManager.AuthenticateDirect(new AuthenticationRequest { UserId = userId, diff --git a/Emby.Server.Implementations/QuickConnect/RedisQuickConnectStore.cs b/Emby.Server.Implementations/QuickConnect/RedisQuickConnectStore.cs index 715e0a6c6e..e3396be7ba 100644 --- a/Emby.Server.Implementations/QuickConnect/RedisQuickConnectStore.cs +++ b/Emby.Server.Implementations/QuickConnect/RedisQuickConnectStore.cs @@ -13,13 +13,32 @@ namespace Emby.Server.Implementations.QuickConnect; /// /// A Redis-backed that lets the initiate, authorize and exchange legs -/// of a quick connect flow land on different instances. Expiry is the key TTL, and an authorization is -/// consumed with GETDEL so only one instance can ever hand out a given secret's access token. +/// of a quick connect flow land on different instances. Expiry is the key TTL, an authorization is +/// claimed with a Lua check-and-set and consumed with GETDEL, so only one instance can ever mint +/// a given secret's access token and only one can ever hand it out. /// +/// +/// A pending request survives an unreachable Redis through a process-local fallback, because a second +/// copy of it is harmless. An authorization has none: a second copy of it is a second access token, and +/// a write whose response timed out may well have been applied, so a transport failure on that path is +/// surfaced rather than degraded. +/// public sealed class RedisQuickConnectStore : IQuickConnectStore { private const string KeyPrefix = "jellyfin:quickconnect:"; + /// + /// Lua script for the atomic claim of the sole right to authorize a request: the request has to + /// exist and not already be authorized, and the claim marker is taken with SET NX, so of two + /// instances racing on one code exactly one goes on to mint an access token. + /// + private const string ClaimAuthorizationScript = @" +local raw = redis.call('GET', KEYS[1]) +if not raw then return 0 end +if cjson.decode(raw)['Authenticated'] then return 0 end +if redis.call('SET', KEYS[2], '1', 'NX', 'PX', ARGV[1]) then return 1 end +return 0"; + private readonly IDatabase _db; private readonly InMemoryQuickConnectStore _fallback; private readonly ILogger _logger; @@ -41,39 +60,38 @@ public sealed class RedisQuickConnectStore : IQuickConnectStore /// public async Task GetRequestBySecretAsync(string secret, CancellationToken cancellationToken = default) { + RedisValue raw; try { - var raw = await _db.StringGetAsync(RequestKey(secret)).ConfigureAwait(false); - if (raw.HasValue) - { - return JsonSerializer.Deserialize(raw.ToString(), JsonDefaults.Options); - } + raw = await _db.StringGetAsync(RequestKey(secret)).ConfigureAwait(false); } - catch (Exception ex) + catch (Exception ex) when (IsTransportFailure(ex)) { LogDegraded(ex); + return await _fallback.GetRequestBySecretAsync(secret, cancellationToken).ConfigureAwait(false); } - return await _fallback.GetRequestBySecretAsync(secret, cancellationToken).ConfigureAwait(false); + // A miss is an answer rather than a transport failure, so the fallback is not consulted for it. + return raw.HasValue ? JsonSerializer.Deserialize(raw.ToString(), JsonDefaults.Options) : null; } /// public async Task GetRequestByCodeAsync(string code, CancellationToken cancellationToken = default) { + RedisValue secret; try { - var secret = await _db.StringGetAsync(CodeKey(code)).ConfigureAwait(false); - if (secret.HasValue) - { - return await GetRequestBySecretAsync(secret.ToString(), cancellationToken).ConfigureAwait(false); - } + secret = await _db.StringGetAsync(CodeKey(code)).ConfigureAwait(false); } - catch (Exception ex) + catch (Exception ex) when (IsTransportFailure(ex)) { LogDegraded(ex); + return await _fallback.GetRequestByCodeAsync(code, cancellationToken).ConfigureAwait(false); } - return await _fallback.GetRequestByCodeAsync(code, cancellationToken).ConfigureAwait(false); + return secret.HasValue + ? await GetRequestBySecretAsync(secret.ToString(), cancellationToken).ConfigureAwait(false) + : null; } /// @@ -93,13 +111,30 @@ public sealed class RedisQuickConnectStore : IQuickConnectStore await _db.StringSetAsync(RequestKey(request.Secret), json, ttl).ConfigureAwait(false); await _db.StringSetAsync(CodeKey(request.Code), request.Secret, ttl).ConfigureAwait(false); } - catch (Exception ex) + catch (Exception ex) when (IsTransportFailure(ex)) { LogDegraded(ex); await _fallback.SetRequestAsync(request, expiresUtc, cancellationToken).ConfigureAwait(false); } } + /// + public async Task TryClaimAuthorizationAsync(string secret, DateTime expiresUtc, CancellationToken cancellationToken = default) + { + var ttl = expiresUtc - DateTime.UtcNow; + if (ttl <= TimeSpan.Zero) + { + return false; + } + + var claimed = (long?)await _db.ScriptEvaluateAsync( + ClaimAuthorizationScript, + keys: new RedisKey[] { RequestKey(secret), ClaimKey(secret) }, + values: new RedisValue[] { (long)ttl.TotalMilliseconds }).ConfigureAwait(false); + + return claimed == 1; + } + /// public async Task SetAuthorizationAsync(string secret, AuthenticationResult authenticationResult, DateTime expiresUtc, CancellationToken cancellationToken = default) { @@ -109,43 +144,32 @@ public sealed class RedisQuickConnectStore : IQuickConnectStore return; } - try - { - var json = JsonSerializer.Serialize(authenticationResult, JsonDefaults.Options); - await _db.StringSetAsync(AuthorizationKey(secret), json, ttl).ConfigureAwait(false); - } - catch (Exception ex) - { - LogDegraded(ex); - await _fallback.SetAuthorizationAsync(secret, authenticationResult, expiresUtc, cancellationToken).ConfigureAwait(false); - } + var json = JsonSerializer.Serialize(authenticationResult, JsonDefaults.Options); + await _db.StringSetAsync(AuthorizationKey(secret), json, ttl).ConfigureAwait(false); } /// public async Task TryConsumeAuthorizationAsync(string secret, CancellationToken cancellationToken = default) { - try - { - var raw = await _db.StringGetDeleteAsync(AuthorizationKey(secret)).ConfigureAwait(false); - if (raw.HasValue) - { - return JsonSerializer.Deserialize(raw.ToString(), JsonDefaults.Options); - } - } - catch (Exception ex) - { - LogDegraded(ex); - } + var raw = await _db.StringGetDeleteAsync(AuthorizationKey(secret)).ConfigureAwait(false); - return await _fallback.TryConsumeAuthorizationAsync(secret, cancellationToken).ConfigureAwait(false); + return raw.HasValue + ? JsonSerializer.Deserialize(raw.ToString(), JsonDefaults.Options) + : null; } + // Deliberately excludes a malformed stored value, which is a fault of its own rather than a reason + // to answer from this instance. + private static bool IsTransportFailure(Exception exception) => exception is RedisException or TimeoutException; + private static string RequestKey(string secret) => KeyPrefix + "request:" + secret; private static string CodeKey(string code) => KeyPrefix + "code:" + code; + private static string ClaimKey(string secret) => KeyPrefix + "claim:" + secret; + private static string AuthorizationKey(string secret) => KeyPrefix + "auth:" + secret; private void LogDegraded(Exception exception) - => _logger.LogWarning(exception, "Quick connect state could not be shared through Redis; falling back to this instance only."); + => _logger.LogWarning(exception, "Quick connect request state could not be shared through Redis; falling back to this instance only."); } diff --git a/MediaBrowser.Controller/QuickConnect/IQuickConnectStore.cs b/MediaBrowser.Controller/QuickConnect/IQuickConnectStore.cs index 2d47c90507..4bc30d9e3d 100644 --- a/MediaBrowser.Controller/QuickConnect/IQuickConnectStore.cs +++ b/MediaBrowser.Controller/QuickConnect/IQuickConnectStore.cs @@ -38,6 +38,17 @@ public interface IQuickConnectStore /// A representing the asynchronous operation. Task SetRequestAsync(QuickConnectResult request, DateTime expiresUtc, CancellationToken cancellationToken = default); + /// + /// Atomically claims the sole right to authorize the request behind , so + /// that two instances racing on one code cannot both mint an access token. The claim is never + /// released: a mint that failed after writing its token would otherwise be retried into a second one. + /// + /// The request secret. + /// The instant the claim lapses, after which the request can be authorized again. + /// A cancellation token. + /// true when this caller may go on to authorize the request; false when it is unknown, expired, already authorized or being authorized elsewhere. + Task TryClaimAuthorizationAsync(string secret, DateTime expiresUtc, CancellationToken cancellationToken = default); + /// /// Stores the authentication minted for an authorized request until . /// diff --git a/MediaBrowser.Controller/QuickConnect/InMemoryQuickConnectStore.cs b/MediaBrowser.Controller/QuickConnect/InMemoryQuickConnectStore.cs index 0b42bd08d3..2296449406 100644 --- a/MediaBrowser.Controller/QuickConnect/InMemoryQuickConnectStore.cs +++ b/MediaBrowser.Controller/QuickConnect/InMemoryQuickConnectStore.cs @@ -17,6 +17,7 @@ public sealed class InMemoryQuickConnectStore : IQuickConnectStore { private readonly ConcurrentDictionary> _requests = new(StringComparer.Ordinal); private readonly ConcurrentDictionary> _authorizations = new(StringComparer.Ordinal); + private readonly ConcurrentDictionary _authorizationClaims = new(StringComparer.Ordinal); /// public Task GetRequestBySecretAsync(string secret, CancellationToken cancellationToken = default) @@ -44,6 +45,18 @@ public sealed class InMemoryQuickConnectStore : IQuickConnectStore return Task.CompletedTask; } + /// + public Task TryClaimAuthorizationAsync(string secret, DateTime expiresUtc, CancellationToken cancellationToken = default) + { + Expire(); + if (!_requests.TryGetValue(secret, out var entry) || entry.Value.Authenticated) + { + return Task.FromResult(false); + } + + return Task.FromResult(_authorizationClaims.TryAdd(secret, expiresUtc)); + } + /// public Task SetAuthorizationAsync(string secret, AuthenticationResult authenticationResult, DateTime expiresUtc, CancellationToken cancellationToken = default) { @@ -83,6 +96,14 @@ public sealed class InMemoryQuickConnectStore : IQuickConnectStore _authorizations.TryRemove(secret, out _); } } + + foreach (var (secret, expiresUtc) in _authorizationClaims) + { + if (expiresUtc <= now) + { + _authorizationClaims.TryRemove(secret, out _); + } + } } private sealed record Entry(DateTime ExpiresUtc, T Value); diff --git a/tests/Jellyfin.Server.Implementations.Tests/QuickConnect/QuickConnectManagerTests.cs b/tests/Jellyfin.Server.Implementations.Tests/QuickConnect/QuickConnectManagerTests.cs index dffede2851..31eac9bba0 100644 --- a/tests/Jellyfin.Server.Implementations.Tests/QuickConnect/QuickConnectManagerTests.cs +++ b/tests/Jellyfin.Server.Implementations.Tests/QuickConnect/QuickConnectManagerTests.cs @@ -140,6 +140,19 @@ namespace Jellyfin.Server.Implementations.Tests.QuickConnect Assert.True(await _quickConnectManager.AuthorizeRequest(Guid.Empty, res.Code)); } + [Fact] + public async Task AuthorizeRequest_RacedOnOneCode_SucceedsOnce() + { + _config.QuickConnectAvailable = true; + var res = await _quickConnectManager.TryConnect(_quickConnectAuthInfo); + + var outcomes = await Task.WhenAll( + Task.Run(() => AuthorizeAsync(res.Code)), + Task.Run(() => AuthorizeAsync(res.Code))); + + Assert.Single(outcomes, authorized => authorized); + } + [Fact] public async Task GetAuthorizedRequest_SecondExchange_ThrowsResourceNotFoundException() { @@ -150,5 +163,17 @@ namespace Jellyfin.Server.Implementations.Tests.QuickConnect Assert.NotNull(await _quickConnectManager.GetAuthorizedRequest(res.Secret)); await Assert.ThrowsAsync(() => _quickConnectManager.GetAuthorizedRequest(res.Secret)); } + + private async Task AuthorizeAsync(string code) + { + try + { + return await _quickConnectManager.AuthorizeRequest(Guid.Empty, code).ConfigureAwait(false); + } + catch (InvalidOperationException) + { + return false; + } + } } } diff --git a/tests/Jellyfin.Server.Tests/HighAvailability/JellyfinSectionConfigurationTests.cs b/tests/Jellyfin.Server.Tests/HighAvailability/JellyfinSectionConfigurationTests.cs index 4cfc9518ed..45f6d598aa 100644 --- a/tests/Jellyfin.Server.Tests/HighAvailability/JellyfinSectionConfigurationTests.cs +++ b/tests/Jellyfin.Server.Tests/HighAvailability/JellyfinSectionConfigurationTests.cs @@ -13,6 +13,7 @@ namespace Jellyfin.Server.Tests.HighAvailability; /// startup configuration the host reads them from must therefore accept that form; when it does not, /// a correctly set variable is dropped and the feature it configures stays off without any error. /// +[Collection("JellyfinSectionConfiguration")] public sealed class JellyfinSectionConfigurationTests : IDisposable { private const string RedisKey = "Jellyfin:TranscodeStore:RedisConnectionString"; diff --git a/tests/Jellyfin.Server.Tests/HighAvailability/RedisFaultProxy.cs b/tests/Jellyfin.Server.Tests/HighAvailability/RedisFaultProxy.cs new file mode 100644 index 0000000000..14d6c1a22c --- /dev/null +++ b/tests/Jellyfin.Server.Tests/HighAvailability/RedisFaultProxy.cs @@ -0,0 +1,170 @@ +using System; +using System.Collections.Concurrent; +using System.Globalization; +using System.IO; +using System.Net; +using System.Net.Sockets; +using System.Threading; +using System.Threading.Tasks; +using StackExchange.Redis; + +namespace Jellyfin.Server.Tests.HighAvailability; + +/// +/// A loopback TCP proxy in front of a Redis server. Cutting it drops every connection through it and +/// refuses new ones, so a test can take Redis away from one instance mid-flow - and give it back - the +/// way a restarted valkey does, and watch what a real StackExchange.Redis client makes of it. +/// +public sealed class RedisFaultProxy : IAsyncDisposable +{ + private readonly ConcurrentDictionary _live = new(); + private readonly CancellationTokenSource _cts = new(); + private readonly TcpListener _listener; + private readonly string _targetHost; + private readonly int _targetPort; + private readonly int _port; + + private volatile bool _cut; + + private RedisFaultProxy(TcpListener listener, int port, string targetHost, int targetPort) + { + _listener = listener; + _port = port; + _targetHost = targetHost; + _targetPort = targetPort; + } + + /// + /// Gets a connection string pointing at the proxy. The timeouts are short so a cut surfaces as a + /// failure in seconds rather than in the library's minute-scale defaults. + /// + public string ConnectionString => string.Create( + CultureInfo.InvariantCulture, + $"127.0.0.1:{_port},abortConnect=false,connectTimeout=500,syncTimeout=2000,connectRetry=1"); + + /// + /// Starts a proxy in front of the server named by . + /// + /// The connection string of the server to forward to. + /// The running proxy. + public static RedisFaultProxy Start(string target) + { + var endpoint = ConfigurationOptions.Parse(target).EndPoints[0]; + var (host, port) = endpoint switch + { + DnsEndPoint dns => (dns.Host, dns.Port), + IPEndPoint ip => (ip.Address.ToString(), ip.Port), + _ => throw new NotSupportedException("Unsupported endpoint " + endpoint) + }; + + var listener = new TcpListener(IPAddress.Loopback, 0); + listener.Start(); + + var proxy = new RedisFaultProxy(listener, ((IPEndPoint)listener.LocalEndpoint).Port, host, port); + _ = Task.Run(proxy.AcceptAsync); + return proxy; + } + + /// + /// Takes Redis away from everything connected through the proxy. + /// + public void Cut() + { + _cut = true; + DropLiveConnections(); + } + + /// + /// Lets connections through again. Clients reconnect on their own schedule, so callers have to wait + /// for the connection to come back rather than assume it already has. + /// + public void Restore() => _cut = false; + + /// + public async ValueTask DisposeAsync() + { + _cut = true; + await _cts.CancelAsync().ConfigureAwait(false); + _listener.Stop(); + DropLiveConnections(); + _cts.Dispose(); + } + + private void DropLiveConnections() + { + foreach (var client in _live.Keys) + { + if (_live.TryRemove(client, out _)) + { + client.Dispose(); + } + } + } + + private async Task AcceptAsync() + { + while (!_cts.IsCancellationRequested) + { + TcpClient client; + try + { + client = await _listener.AcceptTcpClientAsync(_cts.Token).ConfigureAwait(false); + } + catch (Exception exception) when (exception is OperationCanceledException or SocketException or ObjectDisposedException) + { + return; + } + + if (_cut) + { + client.Dispose(); + continue; + } + + _ = Task.Run(() => ForwardAsync(client)); + } + } + + private async Task ForwardAsync(TcpClient client) + { + TcpClient? upstream = null; + try + { + upstream = new TcpClient(); + await upstream.ConnectAsync(_targetHost, _targetPort, _cts.Token).ConfigureAwait(false); + + _live[client] = 0; + _live[upstream] = 0; + + var clientStream = client.GetStream(); + var upstreamStream = upstream.GetStream(); + await Task.WhenAny( + CopyAsync(clientStream, upstreamStream), + CopyAsync(upstreamStream, clientStream)).ConfigureAwait(false); + } + catch (Exception exception) when (exception is IOException or SocketException or OperationCanceledException or ObjectDisposedException) + { + } + finally + { + _live.TryRemove(client, out _); + client.Dispose(); + if (upstream is not null) + { + _live.TryRemove(upstream, out _); + upstream.Dispose(); + } + } + } + + private async Task CopyAsync(NetworkStream from, NetworkStream to) + { + try + { + await from.CopyToAsync(to, _cts.Token).ConfigureAwait(false); + } + catch (Exception exception) when (exception is IOException or SocketException or OperationCanceledException or ObjectDisposedException) + { + } + } +} diff --git a/tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectReplicaTests.cs b/tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectReplicaTests.cs index 26fff9a1f1..9d8c3d2d49 100644 --- a/tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectReplicaTests.cs +++ b/tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectReplicaTests.cs @@ -113,7 +113,8 @@ public sealed class QuickConnectReplicaTests : IAsyncLifetime /// /// A secret is single use across the whole deployment: two replicas racing to exchange it must not - /// both hand out an access token. + /// both hand out an access token. One scheduling of one race settles nothing either way, so the race + /// is run repeatedly. /// /// A representing the asynchronous operation. [Fact] @@ -129,17 +130,57 @@ public sealed class QuickConnectReplicaTests : IAsyncLifetime var replicaB = await CreateReplicaAsync(dataSource, user); var replicaC = await CreateReplicaAsync(dataSource, user); - var initiated = await replicaA.Manager.TryConnect(_authorizationInfo); - await replicaB.Manager.AuthorizeRequest(user.Id, initiated.Code); + for (var attempt = 0; attempt < 25; attempt++) + { + var initiated = await replicaA.Manager.TryConnect(AuthorizationInfoFor(attempt)); + await replicaB.Manager.AuthorizeRequest(user.Id, initiated.Code); - var onA = ExchangeAsync(replicaA.Manager, initiated.Secret); - var onC = ExchangeAsync(replicaC.Manager, initiated.Secret); - var outcomes = await Task.WhenAll(onA, onC); + var outcomes = await Task.WhenAll( + Task.Run(() => ExchangeAsync(replicaA.Manager, initiated.Secret), cancellationToken), + Task.Run(() => ExchangeAsync(replicaC.Manager, initiated.Secret), cancellationToken)); - Assert.Single(outcomes, outcome => outcome is not null); + Assert.Single(outcomes, outcome => outcome is not null); - // And it stays consumed for every later attempt, on any replica. - await Assert.ThrowsAsync(() => replicaB.Manager.GetAuthorizedRequest(initiated.Secret)); + // And it stays consumed for every later attempt, on any replica. + await Assert.ThrowsAsync(() => replicaB.Manager.GetAuthorizedRequest(initiated.Secret)); + } + } + + /// + /// Two replicas authorizing one code at the same time mint one access token between them. A second + /// one would be live, attached to the same device and reachable by nobody. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task Authorize_RacedOnTwoReplicas_MintsOneAccessToken() + { + var cancellationToken = TestContext.Current.CancellationToken; + var connectionString = await _postgres.CreateDatabaseAsync("quickconnect_replica_authorize_race", cancellationToken); + + await using var dataSource = new NpgsqlDataSourceBuilder(connectionString).Build(); + var user = await CreateSchemaWithUserAsync(dataSource, cancellationToken); + + var replicaA = await CreateReplicaAsync(dataSource, user); + var replicaB = await CreateReplicaAsync(dataSource, user); + var replicaC = await CreateReplicaAsync(dataSource, user); + + for (var attempt = 0; attempt < 20; attempt++) + { + var authorizationInfo = AuthorizationInfoFor(attempt); + var initiated = await replicaA.Manager.TryConnect(authorizationInfo); + + var outcomes = await Task.WhenAll( + Task.Run(() => AuthorizeAsync(replicaB.Manager, user.Id, initiated.Code), cancellationToken), + Task.Run(() => AuthorizeAsync(replicaC.Manager, user.Id, initiated.Code), cancellationToken)); + + Assert.Single(outcomes, authorized => authorized); + + var devices = await replicaA.Devices.GetDevices(new DeviceQuery { DeviceId = authorizationInfo.DeviceId }); + var device = Assert.Single(devices.Items); + + var exchanged = await replicaA.Manager.GetAuthorizedRequest(initiated.Secret); + Assert.Equal(device.AccessToken, exchanged.AccessToken); + } } /// @@ -203,6 +244,26 @@ public sealed class QuickConnectReplicaTests : IAsyncLifetime await Assert.ThrowsAsync(() => replicaB.Manager.GetAuthorizedRequest(initiated.Secret)); } + private static AuthorizationInfo AuthorizationInfoFor(int attempt) => new AuthorizationInfo + { + Device = _authorizationInfo.Device, + DeviceId = string.Create(CultureInfo.InvariantCulture, $"device-{attempt}"), + Client = _authorizationInfo.Client, + Version = _authorizationInfo.Version + }; + + private static async Task AuthorizeAsync(IQuickConnect manager, Guid userId, string code) + { + try + { + return await manager.AuthorizeRequest(userId, code).ConfigureAwait(false); + } + catch (InvalidOperationException) + { + return false; + } + } + private static async Task ExchangeAsync(IQuickConnect manager, string secret) { try diff --git a/tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectStoreWiringTests.cs b/tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectStoreWiringTests.cs new file mode 100644 index 0000000000..fb58e45374 --- /dev/null +++ b/tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectStoreWiringTests.cs @@ -0,0 +1,142 @@ +using System; +using System.IO; +using System.Threading.Tasks; +using Emby.Server.Implementations.QuickConnect; +using Jellyfin.Server.Extensions; +using Jellyfin.Server.Tests.HighAvailability; +using MediaBrowser.Common.Configuration; +using MediaBrowser.Controller.Authentication; +using MediaBrowser.Controller.QuickConnect; +using MediaBrowser.Model.QuickConnect; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using Moq; +using StackExchange.Redis; +using Xunit; + +namespace Jellyfin.Server.Tests.QuickConnect; + +/// +/// Drives the whole configuration path a deployment uses: a bare +/// Jellyfin__TranscodeStore__RedisConnectionString environment variable, the server's own +/// configuration builder, the store registration, and a quick connect flow against a real valkey. +/// +[Trait("Category", "RequiresDocker")] +[Collection("JellyfinSectionConfiguration")] +public sealed class QuickConnectStoreWiringTests : IAsyncLifetime +{ + private const string RedisConnectionStringVariable = "Jellyfin__TranscodeStore__RedisConnectionString"; + + private RedisTestServer _redis = null!; + private string _configDirectory = string.Empty; + + /// + public async ValueTask InitializeAsync() + { + _redis = await RedisTestServer.StartAsync().ConfigureAwait(false); + _configDirectory = Directory.CreateTempSubdirectory("jellyfin-quickconnect-wiring").FullName; + await File.WriteAllTextAsync(Path.Combine(_configDirectory, "logging.default.json"), "{}").ConfigureAwait(false); + } + + /// + public async ValueTask DisposeAsync() + { + Environment.SetEnvironmentVariable(RedisConnectionStringVariable, null); + + if (_configDirectory.Length > 0) + { + Directory.Delete(_configDirectory, true); + } + + await _redis.DisposeAsync().ConfigureAwait(false); + } + + /// + /// The variable form deployments set selects the shared store, and that store really talks to valkey. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task ManifestStyleEnvironmentVariable_SelectsTheSharedStore() + { + Environment.SetEnvironmentVariable(RedisConnectionStringVariable, _redis.ConnectionString + ",abortConnect=false"); + + await using var provider = BuildProvider(); + + var store = provider.GetRequiredService(); + Assert.IsType(store); + + var request = NewRequest(); + await store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), TestContext.Current.CancellationToken); + + Assert.True(await store.TryClaimAuthorizationAsync(request.Secret, DateTime.UtcNow.AddMinutes(10), TestContext.Current.CancellationToken)); + + var redis = provider.GetRequiredService(); + Assert.True(await redis.GetDatabase().KeyExistsAsync("jellyfin:quickconnect:request:" + request.Secret)); + } + + /// + /// Without the variable the deployment is single-instance and gets the process-local store. + /// + [Fact] + public void NoEnvironmentVariable_SelectsTheProcessLocalStore() + { + Environment.SetEnvironmentVariable(RedisConnectionStringVariable, null); + + using var provider = BuildProvider(); + + Assert.IsType(provider.GetRequiredService()); + } + + /// + /// A configured but unreachable Redis degrades to the single-instance behaviour of a flow having to + /// complete against one instance, rather than taking quick connect down at startup. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task UnreachableRedisAtStartup_DegradesToTheProcessLocalStore() + { + Environment.SetEnvironmentVariable(RedisConnectionStringVariable, "127.0.0.1:1,connectTimeout=250,connectRetry=0"); + + await using var provider = BuildProvider(); + + var store = provider.GetRequiredService(); + Assert.IsType(store); + + // Quick connect still works, it just cannot span instances. + var request = NewRequest(); + await store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), TestContext.Current.CancellationToken); + Assert.True(await store.TryClaimAuthorizationAsync(request.Secret, DateTime.UtcNow.AddMinutes(10), TestContext.Current.CancellationToken)); + await store.SetAuthorizationAsync( + request.Secret, + new AuthenticationResult { AccessToken = "token-1" }, + DateTime.UtcNow.AddMinutes(10), + TestContext.Current.CancellationToken); + + Assert.Equal("token-1", (await store.TryConsumeAuthorizationAsync(request.Secret, TestContext.Current.CancellationToken))?.AccessToken); + Assert.Null(await store.TryConsumeAuthorizationAsync(request.Secret, TestContext.Current.CancellationToken)); + } + + private static QuickConnectResult NewRequest() => new QuickConnectResult( + Guid.NewGuid().ToString("N"), + Guid.NewGuid().ToString("N").Substring(0, 6), + DateTime.UtcNow, + "device-1", + "Living Room TV", + "Jellyfin Web", + "1.0.0"); + + private ServiceProvider BuildProvider() + { + var appPaths = new Mock(); + appPaths.Setup(paths => paths.ConfigurationDirectoryPath).Returns(_configDirectory); + IConfiguration configuration = Jellyfin.Server.Program.CreateAppConfiguration(new StartupOptions(), appPaths.Object); + + var services = new ServiceCollection(); + services.AddLogging(); + services.AddTranscodeSessionStore(configuration, NullLogger.Instance); + services.AddQuickConnectStore(configuration, NullLogger.Instance); + + return services.BuildServiceProvider(); + } +} diff --git a/tests/Jellyfin.Server.Tests/QuickConnect/RedisQuickConnectStoreDegradedTests.cs b/tests/Jellyfin.Server.Tests/QuickConnect/RedisQuickConnectStoreDegradedTests.cs new file mode 100644 index 0000000000..3c07faeae7 --- /dev/null +++ b/tests/Jellyfin.Server.Tests/QuickConnect/RedisQuickConnectStoreDegradedTests.cs @@ -0,0 +1,308 @@ +using System; +using System.Collections.Generic; +using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; +using Emby.Server.Implementations.QuickConnect; +using Jellyfin.Server.Tests.HighAvailability; +using MediaBrowser.Controller.Authentication; +using MediaBrowser.Model.QuickConnect; +using Microsoft.Extensions.Logging.Abstractions; +using StackExchange.Redis; +using Xunit; + +namespace Jellyfin.Server.Tests.QuickConnect; + +/// +/// What a does while its Redis is unreachable. Each instance talks +/// to the one real server through a proxy of its own, so an outage can be given to one instance and not +/// the others, and then taken back. +/// +[Trait("Category", "RequiresDocker")] +public sealed class RedisQuickConnectStoreDegradedTests : IAsyncLifetime +{ + private readonly List _proxies = new(); + private readonly List _connections = new(); + + private RedisTestServer _redis = null!; + + private static CancellationToken CancellationToken => TestContext.Current.CancellationToken; + + /// + public async ValueTask InitializeAsync() + { + _redis = await RedisTestServer.StartAsync().ConfigureAwait(false); + } + + /// + public async ValueTask DisposeAsync() + { + foreach (var connection in _connections) + { + await connection.DisposeAsync().ConfigureAwait(false); + } + + foreach (var proxy in _proxies) + { + await proxy.DisposeAsync().ConfigureAwait(false); + } + + await _redis.DisposeAsync().ConfigureAwait(false); + } + + /// + /// A request stored while Redis is unreachable is still resolvable on the instance that stored it, + /// so a flow whose three legs happen to land on one instance keeps working through the outage. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task PendingRequest_SurvivesAnOutage_OnTheInstanceThatStoredIt() + { + var instance = await CreateInstanceAsync(); + var request = NewRequest(); + + instance.Proxy.Cut(); + await instance.Store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), CancellationToken); + + Assert.Equal(request.Secret, (await instance.Store.GetRequestBySecretAsync(request.Secret, CancellationToken))?.Secret); + Assert.Equal(request.Secret, (await instance.Store.GetRequestByCodeAsync(request.Code, CancellationToken))?.Secret); + } + + /// + /// Once Redis answers again it is the only authority: a miss is a miss, not a reason to serve the + /// copy this instance kept while it was unreachable. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task PendingRequest_StoredDuringAnOutage_IsNotServedOnceRedisAnswersAgain() + { + var instance = await CreateInstanceAsync(); + var request = NewRequest(); + + instance.Proxy.Cut(); + await instance.Store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), CancellationToken); + Assert.NotNull(await instance.Store.GetRequestBySecretAsync(request.Secret, CancellationToken)); + + await RestoreAsync(instance); + + Assert.Null(await instance.Store.GetRequestBySecretAsync(request.Secret, CancellationToken)); + Assert.Null(await instance.Store.GetRequestByCodeAsync(request.Code, CancellationToken)); + } + + /// + /// A malformed stored value is a fault of its own, not a transport failure, so it is surfaced rather + /// than answered from the copy this instance happens to hold. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task PendingRequest_ThatIsMalformedInRedis_SurfacesInsteadOfDegrading() + { + var instance = await CreateInstanceAsync(); + var request = NewRequest(); + + instance.Proxy.Cut(); + await instance.Store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), CancellationToken); + await RestoreAsync(instance); + + await instance.Connection.GetDatabase().StringSetAsync( + "jellyfin:quickconnect:request:" + request.Secret, + "{ not json", + TimeSpan.FromMinutes(10)); + + await Assert.ThrowsAsync(() => instance.Store.GetRequestBySecretAsync(request.Secret, CancellationToken)); + } + + /// + /// An authorization write that failed leaves nothing behind on the instance, because the response + /// that never arrived may still have been applied and a second copy of an authorization is a second + /// access token. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task Authorization_ThatFailedToStore_LeavesNothingOnTheInstance() + { + var instance = await CreateInstanceAsync(); + var request = NewRequest(); + await instance.Store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), CancellationToken); + + instance.Proxy.Cut(); + await AssertTransportFailureAsync(() => instance.Store.SetAuthorizationAsync( + request.Secret, + new AuthenticationResult { AccessToken = "token-1" }, + DateTime.UtcNow.AddMinutes(10), + CancellationToken)); + + await RestoreAsync(instance); + + Assert.Null(await instance.Store.TryConsumeAuthorizationAsync(request.Secret, CancellationToken)); + } + + /// + /// The instance whose authorization write failed while the write landed anyway still hands the token + /// out exactly once, rather than once from Redis and again from a copy of its own. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task Authorization_IsHandedOutOnce_EvenAfterAFailedWriteOnTheSameInstance() + { + var instance = await CreateInstanceAsync(); + var request = NewRequest(); + var authentication = new AuthenticationResult { AccessToken = "token-1" }; + await instance.Store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), CancellationToken); + + instance.Proxy.Cut(); + await AssertTransportFailureAsync(() => instance.Store.SetAuthorizationAsync( + request.Secret, + authentication, + DateTime.UtcNow.AddMinutes(10), + CancellationToken)); + + await RestoreAsync(instance); + + // Stands in for that write having been applied before the response was lost. + await instance.Store.SetAuthorizationAsync(request.Secret, authentication, DateTime.UtcNow.AddMinutes(10), CancellationToken); + + Assert.Equal("token-1", (await instance.Store.TryConsumeAuthorizationAsync(request.Secret, CancellationToken))?.AccessToken); + Assert.Null(await instance.Store.TryConsumeAuthorizationAsync(request.Secret, CancellationToken)); + } + + /// + /// Exchanging during an outage fails loudly and spends nothing, so the token is still there to be + /// handed out once when Redis comes back. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task Exchange_DuringAnOutage_SurfacesTheFailureAndLeavesTheTokenUnspent() + { + var instance = await CreateInstanceAsync(); + var request = NewRequest(); + await instance.Store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), CancellationToken); + await instance.Store.SetAuthorizationAsync( + request.Secret, + new AuthenticationResult { AccessToken = "token-1" }, + DateTime.UtcNow.AddMinutes(10), + CancellationToken); + + instance.Proxy.Cut(); + await AssertTransportFailureAsync(() => instance.Store.TryConsumeAuthorizationAsync(request.Secret, CancellationToken)); + + await RestoreAsync(instance); + + Assert.Equal("token-1", (await instance.Store.TryConsumeAuthorizationAsync(request.Secret, CancellationToken))?.AccessToken); + Assert.Null(await instance.Store.TryConsumeAuthorizationAsync(request.Secret, CancellationToken)); + } + + /// + /// Authorizing during an outage fails loudly rather than claiming locally, because a claim only this + /// instance knows about does not stop another one minting a second access token. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task Claim_DuringAnOutage_SurfacesTheFailure() + { + var instance = await CreateInstanceAsync(); + var request = NewRequest(); + await instance.Store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), CancellationToken); + + instance.Proxy.Cut(); + await AssertTransportFailureAsync(() => instance.Store.TryClaimAuthorizationAsync( + request.Secret, + DateTime.UtcNow.AddMinutes(10), + CancellationToken)); + } + + /// + /// Two instances racing to authorize one request: exactly one of them may go on to mint a token. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task Claim_RacedOnTwoInstances_SucceedsOnce() + { + var first = await CreateInstanceAsync(); + var second = await CreateInstanceAsync(); + + for (var attempt = 0; attempt < 25; attempt++) + { + var request = NewRequest(); + var expiresUtc = DateTime.UtcNow.AddMinutes(10); + await first.Store.SetRequestAsync(request, expiresUtc, CancellationToken); + + var claims = await Task.WhenAll( + Task.Run(() => first.Store.TryClaimAuthorizationAsync(request.Secret, expiresUtc, CancellationToken), CancellationToken), + Task.Run(() => second.Store.TryClaimAuthorizationAsync(request.Secret, expiresUtc, CancellationToken), CancellationToken)); + + Assert.Single(claims, claimed => claimed); + } + } + + /// + /// A request that is unknown, already claimed or already authorized cannot be claimed. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task Claim_IsRefused_ForUnknownClaimedAndAuthorizedRequests() + { + var instance = await CreateInstanceAsync(); + var expiresUtc = DateTime.UtcNow.AddMinutes(10); + + Assert.False(await instance.Store.TryClaimAuthorizationAsync("unknown-secret", expiresUtc, CancellationToken)); + + var request = NewRequest(); + await instance.Store.SetRequestAsync(request, expiresUtc, CancellationToken); + Assert.True(await instance.Store.TryClaimAuthorizationAsync(request.Secret, expiresUtc, CancellationToken)); + Assert.False(await instance.Store.TryClaimAuthorizationAsync(request.Secret, expiresUtc, CancellationToken)); + + var authorized = NewRequest(); + authorized.Authenticated = true; + await instance.Store.SetRequestAsync(authorized, expiresUtc, CancellationToken); + Assert.False(await instance.Store.TryClaimAuthorizationAsync(authorized.Secret, expiresUtc, CancellationToken)); + } + + private static QuickConnectResult NewRequest() => new QuickConnectResult( + Guid.NewGuid().ToString("N"), + Guid.NewGuid().ToString("N").Substring(0, 6), + DateTime.UtcNow, + "device-1", + "Living Room TV", + "Jellyfin Web", + "1.0.0"); + + private static async Task AssertTransportFailureAsync(Func operation) + { + var exception = await Record.ExceptionAsync(operation); + + Assert.NotNull(exception); + Assert.True(exception is RedisException or TimeoutException, exception.ToString()); + } + + private static async Task RestoreAsync(Instance instance) + { + instance.Proxy.Restore(); + + for (var attempt = 1; ; attempt++) + { + try + { + await instance.Connection.GetDatabase().PingAsync(); + return; + } + catch (Exception exception) when (exception is RedisException or TimeoutException && attempt < 60) + { + await Task.Delay(TimeSpan.FromMilliseconds(500), CancellationToken); + } + } + } + + private async Task CreateInstanceAsync() + { + var proxy = RedisFaultProxy.Start(_redis.ConnectionString); + _proxies.Add(proxy); + + var connection = await ConnectionMultiplexer.ConnectAsync(proxy.ConnectionString).ConfigureAwait(false); + _connections.Add(connection); + + return new Instance(proxy, connection, new RedisQuickConnectStore(connection, NullLogger.Instance)); + } + + private sealed record Instance(RedisFaultProxy Proxy, IConnectionMultiplexer Connection, RedisQuickConnectStore Store); +}