From ba4d487c6512894f4605588972cf8e19cc1aceca Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Thu, 24 Sep 2026 23:02:56 +1000 Subject: [PATCH 1/3] share quick connect state between instances Move the pending requests and authorized secrets out of process so the initiate, authorize and exchange legs can land on different replicas. --- .woodpecker/ci.yaml | 6 +- .../QuickConnect/QuickConnectManager.cs | 80 ++--- .../QuickConnect/RedisQuickConnectStore.cs | 151 +++++++++ .../Controllers/QuickConnectController.cs | 6 +- Jellyfin.Api/Controllers/UserController.cs | 4 +- Jellyfin.Server/CoreAppHost.cs | 4 + ...ConnectStoreServiceCollectionExtensions.cs | 72 ++++ .../QuickConnect/IQuickConnect.cs | 6 +- .../QuickConnect/IQuickConnectStore.cs | 59 ++++ .../QuickConnect/InMemoryQuickConnectStore.cs | 89 +++++ .../QuickConnect/QuickConnectManagerTests.cs | 49 ++- .../HighAvailability/RedisTestServer.cs | 91 ++++++ .../Jellyfin.Server.Tests.csproj | 1 + .../QuickConnect/QuickConnectReplicaTests.cs | 308 ++++++++++++++++++ 14 files changed, 841 insertions(+), 85 deletions(-) create mode 100644 Emby.Server.Implementations/QuickConnect/RedisQuickConnectStore.cs create mode 100644 Jellyfin.Server/Extensions/QuickConnectStoreServiceCollectionExtensions.cs create mode 100644 MediaBrowser.Controller/QuickConnect/IQuickConnectStore.cs create mode 100644 MediaBrowser.Controller/QuickConnect/InMemoryQuickConnectStore.cs create mode 100644 tests/Jellyfin.Server.Tests/HighAvailability/RedisTestServer.cs create mode 100644 tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectReplicaTests.cs diff --git a/.woodpecker/ci.yaml b/.woodpecker/ci.yaml index c1c5e741f0..668be16db5 100644 --- a/.woodpecker/ci.yaml +++ b/.woodpecker/ci.yaml @@ -47,6 +47,7 @@ steps: # Its data directory lives on the step's ephemeral storage, not on the workspace volume. # Both projects attach to it through JELLYFIN_TEST_POSTGRES and give every test a database of # its own, so nothing here depends on a docker daemon. + # Valkey runs in the step for the same reason, attached through JELLYFIN_TEST_REDIS. - name: postgres-migration-chain image: mcr.microsoft.com/dotnet/sdk:10.0 depends_on: @@ -55,13 +56,16 @@ steps: DOTNET_CLI_TELEMETRY_OPTOUT: "1" DOTNET_NOLOGO: "1" JELLYFIN_TEST_POSTGRES: "Host=127.0.0.1;Port=5432;Database=postgres;Username=postgres" + JELLYFIN_TEST_REDIS: "127.0.0.1:6379" commands: - apt-get -o Acquire::Retries=3 update || apt-get -o Acquire::Retries=3 update - - DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends postgresql + - DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends postgresql valkey-server - install -d -o postgres -g postgres /tmp/pgdata /tmp/pgrun - PGBIN=$(ls -d /usr/lib/postgresql/*/bin | tail -1) - su postgres -c "$PGBIN/initdb -D /tmp/pgdata -A trust -U postgres" - su postgres -c "$PGBIN/pg_ctl -D /tmp/pgdata -o \"-c listen_addresses=127.0.0.1 -k /tmp/pgrun\" -l /tmp/pg.log -w start" + - valkey-server --daemonize yes --bind 127.0.0.1 --port 6379 --save '' + - valkey-cli ping - dotnet build tests/Jellyfin.Server.Tests/Jellyfin.Server.Tests.csproj -c Release - dotnet build tests/Jellyfin.Database.Tests.PostgreSQL/Jellyfin.Database.Tests.PostgreSQL.csproj -c Release - dotnet test tests/Jellyfin.Server.Tests/Jellyfin.Server.Tests.csproj -c Release --no-build --verbosity minimal --filter "Category=RequiresDocker" diff --git a/Emby.Server.Implementations/QuickConnect/QuickConnectManager.cs b/Emby.Server.Implementations/QuickConnect/QuickConnectManager.cs index c4bda96371..855205e4aa 100644 --- a/Emby.Server.Implementations/QuickConnect/QuickConnectManager.cs +++ b/Emby.Server.Implementations/QuickConnect/QuickConnectManager.cs @@ -1,7 +1,5 @@ using System; -using System.Collections.Concurrent; using System.Globalization; -using System.Linq; using System.Security.Cryptography; using System.Threading.Tasks; using MediaBrowser.Common.Extensions; @@ -30,12 +28,10 @@ namespace Emby.Server.Implementations.QuickConnect /// private const int Timeout = 10; - private readonly ConcurrentDictionary _currentRequests = new(); - private readonly ConcurrentDictionary _authorizedSecrets = new(); - private readonly IServerConfigurationManager _config; private readonly ILogger _logger; private readonly ISessionManager _sessionManager; + private readonly IQuickConnectStore _store; /// /// Initializes a new instance of the class. @@ -44,14 +40,17 @@ namespace Emby.Server.Implementations.QuickConnect /// Configuration. /// Logger. /// Session Manager. + /// Quick connect store. public QuickConnectManager( IServerConfigurationManager config, ILogger logger, - ISessionManager sessionManager) + ISessionManager sessionManager, + IQuickConnectStore store) { _config = config; _logger = logger; _sessionManager = sessionManager; + _store = store; } /// @@ -69,7 +68,7 @@ namespace Emby.Server.Implementations.QuickConnect } /// - public QuickConnectResult TryConnect(AuthorizationInfo authorizationInfo) + public async Task TryConnect(AuthorizationInfo authorizationInfo) { ArgumentException.ThrowIfNullOrEmpty(authorizationInfo.DeviceId); ArgumentException.ThrowIfNullOrEmpty(authorizationInfo.Device); @@ -77,7 +76,6 @@ namespace Emby.Server.Implementations.QuickConnect ArgumentException.ThrowIfNullOrEmpty(authorizationInfo.Version); AssertActive(); - ExpireRequests(); var secret = GenerateSecureRandom(); var code = GenerateCode(); @@ -90,19 +88,17 @@ namespace Emby.Server.Implementations.QuickConnect authorizationInfo.Client, authorizationInfo.Version); - _currentRequests[code] = result; + await _store.SetRequestAsync(result, ExpiryOf(result)).ConfigureAwait(false); return result; } /// - public QuickConnectResult CheckRequestStatus(string secret) + public async Task CheckRequestStatus(string secret) { AssertActive(); - ExpireRequests(); - string code = _currentRequests.Where(x => x.Value.Secret == secret).Select(x => x.Value.Code).DefaultIfEmpty(string.Empty).First(); - - if (!_currentRequests.TryGetValue(code, out QuickConnectResult? result)) + var result = await _store.GetRequestBySecretAsync(secret).ConfigureAwait(false); + if (result is null) { throw new ResourceNotFoundException("Unable to find request with provided secret"); } @@ -136,9 +132,9 @@ namespace Emby.Server.Implementations.QuickConnect public async Task AuthorizeRequest(Guid userId, string code) { AssertActive(); - ExpireRequests(); - if (!_currentRequests.TryGetValue(code, out QuickConnectResult? result)) + var result = await _store.GetRequestByCodeAsync(code).ConfigureAwait(false); + if (result is null) { throw new ResourceNotFoundException("Unable to find request"); } @@ -160,9 +156,10 @@ namespace Emby.Server.Implementations.QuickConnect AppVersion = result.AppVersion }).ConfigureAwait(false); - _authorizedSecrets[result.Secret] = (DateTime.UtcNow, authenticationResult); result.Authenticated = true; - _currentRequests[code] = result; + + await _store.SetAuthorizationAsync(result.Secret, authenticationResult, DateTime.UtcNow.AddMinutes(Timeout)).ConfigureAwait(false); + await _store.SetRequestAsync(result, ExpiryOf(result)).ConfigureAwait(false); _logger.LogDebug("Authorizing device with code {Code} to login as user {UserId}", code, userId); @@ -170,19 +167,21 @@ namespace Emby.Server.Implementations.QuickConnect } /// - public AuthenticationResult GetAuthorizedRequest(string secret) + public async Task GetAuthorizedRequest(string secret) { AssertActive(); - ExpireRequests(); - if (!_authorizedSecrets.TryGetValue(secret, out var result)) + var result = await _store.TryConsumeAuthorizationAsync(secret).ConfigureAwait(false); + if (result is null) { throw new ResourceNotFoundException("Unable to find request"); } - return result.AuthenticationResult; + return result; } + private static DateTime ExpiryOf(QuickConnectResult request) => request.DateAdded.AddMinutes(Timeout); + private string GenerateSecureRandom(int length = 32) { Span bytes = stackalloc byte[length]; @@ -190,42 +189,5 @@ namespace Emby.Server.Implementations.QuickConnect return Convert.ToHexString(bytes); } - - /// - /// Expire quick connect requests that are over the time limit. If is true, all requests are unconditionally expired. - /// - /// If true, all requests will be expired. - private void ExpireRequests(bool expireAll = false) - { - // All requests before this timestamp have expired - var minTime = DateTime.UtcNow.AddMinutes(-Timeout); - - // Expire stale connection requests - foreach (var (_, currentRequest) in _currentRequests) - { - if (expireAll || currentRequest.DateAdded < minTime) - { - var code = currentRequest.Code; - _logger.LogDebug("Removing expired request {Code}", code); - - if (!_currentRequests.TryRemove(code, out _)) - { - _logger.LogWarning("Request {Code} already expired", code); - } - } - } - - foreach (var (secret, (timestamp, _)) in _authorizedSecrets) - { - if (expireAll || timestamp < minTime) - { - _logger.LogDebug("Removing expired secret {Secret}", secret); - if (!_authorizedSecrets.TryRemove(secret, out _)) - { - _logger.LogWarning("Secret {Secret} already expired", secret); - } - } - } - } } } diff --git a/Emby.Server.Implementations/QuickConnect/RedisQuickConnectStore.cs b/Emby.Server.Implementations/QuickConnect/RedisQuickConnectStore.cs new file mode 100644 index 0000000000..715e0a6c6e --- /dev/null +++ b/Emby.Server.Implementations/QuickConnect/RedisQuickConnectStore.cs @@ -0,0 +1,151 @@ +using System; +using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; +using Jellyfin.Extensions.Json; +using MediaBrowser.Controller.Authentication; +using MediaBrowser.Controller.QuickConnect; +using MediaBrowser.Model.QuickConnect; +using Microsoft.Extensions.Logging; +using StackExchange.Redis; + +namespace Emby.Server.Implementations.QuickConnect; + +/// +/// 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. +/// +public sealed class RedisQuickConnectStore : IQuickConnectStore +{ + private const string KeyPrefix = "jellyfin:quickconnect:"; + + private readonly IDatabase _db; + private readonly InMemoryQuickConnectStore _fallback; + private readonly ILogger _logger; + + /// + /// Initializes a new instance of the class. + /// + /// The Redis connection multiplexer. + /// The logger. + public RedisQuickConnectStore(IConnectionMultiplexer redis, ILogger logger) + { + ArgumentNullException.ThrowIfNull(redis); + + _db = redis.GetDatabase(); + _fallback = new InMemoryQuickConnectStore(); + _logger = logger; + } + + /// + public async Task GetRequestBySecretAsync(string secret, CancellationToken cancellationToken = default) + { + try + { + var raw = await _db.StringGetAsync(RequestKey(secret)).ConfigureAwait(false); + if (raw.HasValue) + { + return JsonSerializer.Deserialize(raw.ToString(), JsonDefaults.Options); + } + } + catch (Exception ex) + { + LogDegraded(ex); + } + + return await _fallback.GetRequestBySecretAsync(secret, cancellationToken).ConfigureAwait(false); + } + + /// + public async Task GetRequestByCodeAsync(string code, CancellationToken cancellationToken = default) + { + try + { + var secret = await _db.StringGetAsync(CodeKey(code)).ConfigureAwait(false); + if (secret.HasValue) + { + return await GetRequestBySecretAsync(secret.ToString(), cancellationToken).ConfigureAwait(false); + } + } + catch (Exception ex) + { + LogDegraded(ex); + } + + return await _fallback.GetRequestByCodeAsync(code, cancellationToken).ConfigureAwait(false); + } + + /// + public async Task SetRequestAsync(QuickConnectResult request, DateTime expiresUtc, CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(request); + + var ttl = expiresUtc - DateTime.UtcNow; + if (ttl <= TimeSpan.Zero) + { + return; + } + + try + { + var json = JsonSerializer.Serialize(request, JsonDefaults.Options); + await _db.StringSetAsync(RequestKey(request.Secret), json, ttl).ConfigureAwait(false); + await _db.StringSetAsync(CodeKey(request.Code), request.Secret, ttl).ConfigureAwait(false); + } + catch (Exception ex) + { + LogDegraded(ex); + await _fallback.SetRequestAsync(request, expiresUtc, cancellationToken).ConfigureAwait(false); + } + } + + /// + public async Task SetAuthorizationAsync(string secret, AuthenticationResult authenticationResult, DateTime expiresUtc, CancellationToken cancellationToken = default) + { + var ttl = expiresUtc - DateTime.UtcNow; + if (ttl <= TimeSpan.Zero) + { + 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); + } + } + + /// + 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); + } + + return await _fallback.TryConsumeAuthorizationAsync(secret, cancellationToken).ConfigureAwait(false); + } + + private static string RequestKey(string secret) => KeyPrefix + "request:" + secret; + + private static string CodeKey(string code) => KeyPrefix + "code:" + code; + + 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."); +} diff --git a/Jellyfin.Api/Controllers/QuickConnectController.cs b/Jellyfin.Api/Controllers/QuickConnectController.cs index 5c7b38e137..eebda12ff3 100644 --- a/Jellyfin.Api/Controllers/QuickConnectController.cs +++ b/Jellyfin.Api/Controllers/QuickConnectController.cs @@ -59,7 +59,7 @@ public class QuickConnectController : BaseJellyfinApiController try { var auth = await _authContext.GetAuthorizationInfo(Request).ConfigureAwait(false); - return _quickConnect.TryConnect(auth); + return await _quickConnect.TryConnect(auth).ConfigureAwait(false); } catch (AuthenticationException) { @@ -77,11 +77,11 @@ public class QuickConnectController : BaseJellyfinApiController [HttpGet("Connect")] [ProducesResponseType(StatusCodes.Status200OK)] [ProducesResponseType(StatusCodes.Status404NotFound)] - public ActionResult GetQuickConnectState([FromQuery, Required] string secret) + public async Task> GetQuickConnectState([FromQuery, Required] string secret) { try { - return _quickConnect.CheckRequestStatus(secret); + return await _quickConnect.CheckRequestStatus(secret).ConfigureAwait(false); } catch (ResourceNotFoundException) { diff --git a/Jellyfin.Api/Controllers/UserController.cs b/Jellyfin.Api/Controllers/UserController.cs index 657bda4d15..c8aeb616a4 100644 --- a/Jellyfin.Api/Controllers/UserController.cs +++ b/Jellyfin.Api/Controllers/UserController.cs @@ -245,11 +245,11 @@ public class UserController : BaseJellyfinApiController [HttpPost("AuthenticateWithQuickConnect")] [ProducesResponseType(StatusCodes.Status200OK)] [Tags("Authentication")] - public ActionResult AuthenticateWithQuickConnect([FromBody, Required] QuickConnectDto request) + public async Task> AuthenticateWithQuickConnect([FromBody, Required] QuickConnectDto request) { try { - return _quickConnectManager.GetAuthorizedRequest(request.Secret); + return await _quickConnectManager.GetAuthorizedRequest(request.Secret).ConfigureAwait(false); } catch (SecurityException e) { diff --git a/Jellyfin.Server/CoreAppHost.cs b/Jellyfin.Server/CoreAppHost.cs index 9ac86bb8ed..4e30328870 100644 --- a/Jellyfin.Server/CoreAppHost.cs +++ b/Jellyfin.Server/CoreAppHost.cs @@ -116,6 +116,10 @@ namespace Jellyfin.Server // to the other instances. Redis-backed when configured, no-op otherwise. serviceCollection.AddConfigurationInvalidationBus(_startupConfig, Logger); + // Quick connect store: shares in-flight quick connect requests so the initiate, authorize and + // exchange legs can land on different instances. Redis-backed when configured, local otherwise. + serviceCollection.AddQuickConnectStore(_startupConfig, Logger); + foreach (var type in GetExportTypes()) { serviceCollection.AddSingleton(typeof(ILyricProvider), type); diff --git a/Jellyfin.Server/Extensions/QuickConnectStoreServiceCollectionExtensions.cs b/Jellyfin.Server/Extensions/QuickConnectStoreServiceCollectionExtensions.cs new file mode 100644 index 0000000000..36f270e66e --- /dev/null +++ b/Jellyfin.Server/Extensions/QuickConnectStoreServiceCollectionExtensions.cs @@ -0,0 +1,72 @@ +using System; +using Emby.Server.Implementations.QuickConnect; +using MediaBrowser.Controller.MediaEncoding; +using MediaBrowser.Controller.QuickConnect; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using StackExchange.Redis; + +namespace Jellyfin.Server.Extensions; + +/// +/// Extensions for registering the quick connect store. +/// +public static class QuickConnectStoreServiceCollectionExtensions +{ + /// + /// Registers the quick connect store, Redis-backed when a connection string is configured and + /// process-local otherwise, and reports the selected store at . + /// + /// + /// The connection string is only set for a multi-instance deployment, which is the only shape where + /// the initiate, authorize and exchange legs of one flow can land on different instances. + /// + /// The service collection. + /// The configuration to read the Redis connection string from. + /// The logger to report the selected store on. + /// The updated service collection. + public static IServiceCollection AddQuickConnectStore( + this IServiceCollection serviceCollection, + IConfiguration configuration, + ILogger logger) + { + ArgumentNullException.ThrowIfNull(configuration); + ArgumentNullException.ThrowIfNull(logger); + + if (string.IsNullOrEmpty(configuration[TranscodeStoreOptions.RedisConnectionStringKey])) + { + logger.LogInformation( + "Quick connect store: {Store}. A quick connect flow has to complete against one instance; set {Key} to share it.", + nameof(InMemoryQuickConnectStore), + TranscodeStoreOptions.RedisConnectionStringKey); + + return serviceCollection.AddSingleton(); + } + + logger.LogInformation( + "Quick connect store: {Store}. Quick connect flows complete across any instance.", + nameof(RedisQuickConnectStore)); + + return serviceCollection.AddSingleton(sp => + { + try + { + return new RedisQuickConnectStore( + sp.GetRequiredService(), + sp.GetRequiredService>()); + } + catch (Exception ex) + { + // Fail open: an unreachable Redis degrades to the single-instance behaviour of a flow + // having to complete against one instance, rather than taking quick connect down. + sp.GetRequiredService>().LogError( + ex, + "Redis is configured but unavailable, so quick connect flows will not complete across instances. Check {Key}.", + TranscodeStoreOptions.RedisConnectionStringKey); + + return new InMemoryQuickConnectStore(); + } + }); + } +} diff --git a/MediaBrowser.Controller/QuickConnect/IQuickConnect.cs b/MediaBrowser.Controller/QuickConnect/IQuickConnect.cs index ec3706773a..b585c30dbd 100644 --- a/MediaBrowser.Controller/QuickConnect/IQuickConnect.cs +++ b/MediaBrowser.Controller/QuickConnect/IQuickConnect.cs @@ -21,14 +21,14 @@ namespace MediaBrowser.Controller.QuickConnect /// /// The initiator authorization info. /// A quick connect result with tokens to proceed or throws an exception if not active. - QuickConnectResult TryConnect(AuthorizationInfo authorizationInfo); + Task TryConnect(AuthorizationInfo authorizationInfo); /// /// Checks the status of an individual request. /// /// Unique secret identifier of the request. /// Quick connect result. - QuickConnectResult CheckRequestStatus(string secret); + Task CheckRequestStatus(string secret); /// /// Authorizes a quick connect request to connect as the calling user. @@ -43,6 +43,6 @@ namespace MediaBrowser.Controller.QuickConnect /// /// The secret. /// The authentication result. - AuthenticationResult GetAuthorizedRequest(string secret); + Task GetAuthorizedRequest(string secret); } } diff --git a/MediaBrowser.Controller/QuickConnect/IQuickConnectStore.cs b/MediaBrowser.Controller/QuickConnect/IQuickConnectStore.cs new file mode 100644 index 0000000000..2d47c90507 --- /dev/null +++ b/MediaBrowser.Controller/QuickConnect/IQuickConnectStore.cs @@ -0,0 +1,59 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using MediaBrowser.Controller.Authentication; +using MediaBrowser.Model.QuickConnect; + +namespace MediaBrowser.Controller.QuickConnect; + +/// +/// Holds the state of in-flight quick connect requests. The three legs of a quick connect flow - +/// initiate, authorize and exchange - can each land on a different instance, so the state has to be +/// reachable from all of them. +/// +public interface IQuickConnectStore +{ + /// + /// Looks up a pending request by the secret handed to the initiating client. + /// + /// The request secret. + /// A cancellation token. + /// The request, or null when it is unknown or has expired. + Task GetRequestBySecretAsync(string secret, CancellationToken cancellationToken = default); + + /// + /// Looks up a pending request by the code shown to the user. + /// + /// The user facing code. + /// A cancellation token. + /// The request, or null when it is unknown or has expired. + Task GetRequestByCodeAsync(string code, CancellationToken cancellationToken = default); + + /// + /// Stores a new or updated request until . + /// + /// The request to store. + /// The instant the request stops being resolvable. + /// A cancellation token. + /// A representing the asynchronous operation. + Task SetRequestAsync(QuickConnectResult request, DateTime expiresUtc, CancellationToken cancellationToken = default); + + /// + /// Stores the authentication minted for an authorized request until . + /// + /// The request secret the client exchanges. + /// The authentication to hand out. + /// The instant the authentication stops being exchangeable. + /// A cancellation token. + /// A representing the asynchronous operation. + Task SetAuthorizationAsync(string secret, AuthenticationResult authenticationResult, DateTime expiresUtc, CancellationToken cancellationToken = default); + + /// + /// Atomically takes the authentication for and removes it, so that two + /// instances racing on the same secret cannot both hand out an access token. + /// + /// The request secret. + /// A cancellation token. + /// The authentication, or null when the secret is unknown, expired or already exchanged. + Task TryConsumeAuthorizationAsync(string secret, CancellationToken cancellationToken = default); +} diff --git a/MediaBrowser.Controller/QuickConnect/InMemoryQuickConnectStore.cs b/MediaBrowser.Controller/QuickConnect/InMemoryQuickConnectStore.cs new file mode 100644 index 0000000000..0b42bd08d3 --- /dev/null +++ b/MediaBrowser.Controller/QuickConnect/InMemoryQuickConnectStore.cs @@ -0,0 +1,89 @@ +using System; +using System.Collections.Concurrent; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using MediaBrowser.Controller.Authentication; +using MediaBrowser.Model.QuickConnect; + +namespace MediaBrowser.Controller.QuickConnect; + +/// +/// A process-local . It is the single-instance default, and the +/// fallback a shared store degrades to while its backend is unreachable, so quick connect keeps +/// working for clients whose three legs happen to land on one instance. +/// +public sealed class InMemoryQuickConnectStore : IQuickConnectStore +{ + private readonly ConcurrentDictionary> _requests = new(StringComparer.Ordinal); + private readonly ConcurrentDictionary> _authorizations = new(StringComparer.Ordinal); + + /// + public Task GetRequestBySecretAsync(string secret, CancellationToken cancellationToken = default) + { + Expire(); + return Task.FromResult(_requests.TryGetValue(secret, out var entry) ? entry.Value : null); + } + + /// + public Task GetRequestByCodeAsync(string code, CancellationToken cancellationToken = default) + { + Expire(); + return Task.FromResult(_requests.Values + .Select(entry => entry.Value) + .FirstOrDefault(request => string.Equals(request.Code, code, StringComparison.Ordinal))); + } + + /// + public Task SetRequestAsync(QuickConnectResult request, DateTime expiresUtc, CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(request); + + Expire(); + _requests[request.Secret] = new Entry(expiresUtc, request); + return Task.CompletedTask; + } + + /// + public Task SetAuthorizationAsync(string secret, AuthenticationResult authenticationResult, DateTime expiresUtc, CancellationToken cancellationToken = default) + { + Expire(); + _authorizations[secret] = new Entry(expiresUtc, authenticationResult); + return Task.CompletedTask; + } + + /// + public Task TryConsumeAuthorizationAsync(string secret, CancellationToken cancellationToken = default) + { + Expire(); + if (!_authorizations.TryRemove(secret, out var entry) || entry.ExpiresUtc <= DateTime.UtcNow) + { + return Task.FromResult(null); + } + + return Task.FromResult(entry.Value); + } + + private void Expire() + { + var now = DateTime.UtcNow; + + foreach (var (secret, entry) in _requests) + { + if (entry.ExpiresUtc <= now) + { + _requests.TryRemove(secret, out _); + } + } + + foreach (var (secret, entry) in _authorizations) + { + if (entry.ExpiresUtc <= now) + { + _authorizations.TryRemove(secret, out _); + } + } + } + + 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 30f72f5957..dffede2851 100644 --- a/tests/Jellyfin.Server.Implementations.Tests/QuickConnect/QuickConnectManagerTests.cs +++ b/tests/Jellyfin.Server.Implementations.Tests/QuickConnect/QuickConnectManagerTests.cs @@ -8,6 +8,7 @@ using MediaBrowser.Common.Extensions; using MediaBrowser.Controller.Authentication; using MediaBrowser.Controller.Configuration; using MediaBrowser.Controller.Net; +using MediaBrowser.Controller.QuickConnect; using MediaBrowser.Model.Configuration; using Moq; using Xunit; @@ -40,6 +41,8 @@ namespace Jellyfin.Server.Implementations.Tests.QuickConnect ConfigureMembers = true }).Inject(configManager.Object); + _fixture.Inject(new InMemoryQuickConnectStore()); + // User object contains circular references. _fixture.Behaviors.OfType().ToList() .ForEach(b => _fixture.Behaviors.Remove(b)); @@ -60,8 +63,8 @@ namespace Jellyfin.Server.Implementations.Tests.QuickConnect [InlineData("Device", "", "Client", "1.0.0")] [InlineData("Device", "DeviceId", "", "1.0.0")] [InlineData("Device", "DeviceId", "Client", "")] - public void TryConnect_InvalidAuthorizationInfo_ThrowsArgumentException(string device, string deviceId, string client, string version) - => Assert.Throws(() => _quickConnectManager.TryConnect( + public async Task TryConnect_InvalidAuthorizationInfo_ThrowsArgumentException(string device, string deviceId, string client, string version) + => await Assert.ThrowsAsync(() => _quickConnectManager.TryConnect( new AuthorizationInfo { Device = device, @@ -71,17 +74,17 @@ namespace Jellyfin.Server.Implementations.Tests.QuickConnect })); [Fact] - public void TryConnect_QuickConnectUnavailable_ThrowsAuthenticationException() + public async Task TryConnect_QuickConnectUnavailable_ThrowsAuthenticationException() { _config.QuickConnectAvailable = false; - Assert.Throws(() => _quickConnectManager.TryConnect(_quickConnectAuthInfo)); + await Assert.ThrowsAsync(() => _quickConnectManager.TryConnect(_quickConnectAuthInfo)); } [Fact] - public void CheckRequestStatus_QuickConnectUnavailable_ThrowsAuthenticationException() + public async Task CheckRequestStatus_QuickConnectUnavailable_ThrowsAuthenticationException() { _config.QuickConnectAvailable = false; - Assert.Throws(() => _quickConnectManager.CheckRequestStatus(string.Empty)); + await Assert.ThrowsAsync(() => _quickConnectManager.CheckRequestStatus(string.Empty)); } [Fact] @@ -92,10 +95,10 @@ namespace Jellyfin.Server.Implementations.Tests.QuickConnect } [Fact] - public void GetAuthorizedRequest_QuickConnectUnavailable_ThrowsAuthenticationException() + public async Task GetAuthorizedRequest_QuickConnectUnavailable_ThrowsAuthenticationException() { _config.QuickConnectAvailable = false; - Assert.Throws(() => _quickConnectManager.GetAuthorizedRequest(string.Empty)); + await Assert.ThrowsAsync(() => _quickConnectManager.GetAuthorizedRequest(string.Empty)); } [Fact] @@ -106,34 +109,46 @@ namespace Jellyfin.Server.Implementations.Tests.QuickConnect } [Fact] - public void CheckRequestStatus_QuickConnectAvailable_Success() + public async Task CheckRequestStatus_QuickConnectAvailable_Success() { _config.QuickConnectAvailable = true; - var res1 = _quickConnectManager.TryConnect(_quickConnectAuthInfo); - var res2 = _quickConnectManager.CheckRequestStatus(res1.Secret); - Assert.Equal(res1, res2); + var res1 = await _quickConnectManager.TryConnect(_quickConnectAuthInfo); + var res2 = await _quickConnectManager.CheckRequestStatus(res1.Secret); + Assert.Equal(res1.Secret, res2.Secret); + Assert.Equal(res1.Code, res2.Code); } [Fact] - public void CheckRequestStatus_UnknownSecret_ThrowsResourceNotFoundException() + public async Task CheckRequestStatus_UnknownSecret_ThrowsResourceNotFoundException() { _config.QuickConnectAvailable = true; - Assert.Throws(() => _quickConnectManager.CheckRequestStatus("Unknown secret")); + await Assert.ThrowsAsync(() => _quickConnectManager.CheckRequestStatus("Unknown secret")); } [Fact] - public void GetAuthorizedRequest_UnknownSecret_ThrowsResourceNotFoundException() + public async Task GetAuthorizedRequest_UnknownSecret_ThrowsResourceNotFoundException() { _config.QuickConnectAvailable = true; - Assert.Throws(() => _quickConnectManager.GetAuthorizedRequest("Unknown secret")); + await Assert.ThrowsAsync(() => _quickConnectManager.GetAuthorizedRequest("Unknown secret")); } [Fact] public async Task AuthorizeRequest_QuickConnectAvailable_Success() { _config.QuickConnectAvailable = true; - var res = _quickConnectManager.TryConnect(_quickConnectAuthInfo); + var res = await _quickConnectManager.TryConnect(_quickConnectAuthInfo); Assert.True(await _quickConnectManager.AuthorizeRequest(Guid.Empty, res.Code)); } + + [Fact] + public async Task GetAuthorizedRequest_SecondExchange_ThrowsResourceNotFoundException() + { + _config.QuickConnectAvailable = true; + var res = await _quickConnectManager.TryConnect(_quickConnectAuthInfo); + await _quickConnectManager.AuthorizeRequest(Guid.Empty, res.Code); + + Assert.NotNull(await _quickConnectManager.GetAuthorizedRequest(res.Secret)); + await Assert.ThrowsAsync(() => _quickConnectManager.GetAuthorizedRequest(res.Secret)); + } } } diff --git a/tests/Jellyfin.Server.Tests/HighAvailability/RedisTestServer.cs b/tests/Jellyfin.Server.Tests/HighAvailability/RedisTestServer.cs new file mode 100644 index 0000000000..a17850522c --- /dev/null +++ b/tests/Jellyfin.Server.Tests/HighAvailability/RedisTestServer.cs @@ -0,0 +1,91 @@ +using System; +using System.Threading.Tasks; +using StackExchange.Redis; +using Testcontainers.Redis; + +namespace Jellyfin.Server.Tests.HighAvailability; + +/// +/// Hands out a Redis server for the tests that need one. A server named by JELLYFIN_TEST_REDIS is +/// used as is, so CI can run one beside the step instead of a docker daemon of its own; without it a +/// container is started through testcontainers. +/// +public sealed class RedisTestServer : IAsyncDisposable +{ + /// + /// The connection string of an already running server. + /// + public const string ConnectionStringVariable = "JELLYFIN_TEST_REDIS"; + + private readonly RedisContainer? _container; + + private RedisTestServer(RedisContainer? container, string connectionString) + { + _container = container; + ConnectionString = connectionString; + } + + /// + /// Gets the connection string of the running server. + /// + public string ConnectionString { get; } + + /// + /// Starts or attaches to a Redis server and waits until it accepts connections. + /// + /// The running server. + public static async Task StartAsync() + { + var provided = Environment.GetEnvironmentVariable(ConnectionStringVariable); + if (!string.IsNullOrWhiteSpace(provided)) + { + var attached = new RedisTestServer(null, provided); + await attached.WaitUntilReadyAsync().ConfigureAwait(false); + return attached; + } + + var container = new RedisBuilder("redis:7-alpine").Build(); + await container.StartAsync().ConfigureAwait(false); + + var started = new RedisTestServer(container, container.GetConnectionString()); + await started.WaitUntilReadyAsync().ConfigureAwait(false); + return started; + } + + /// + /// Opens a connection of its own, so each in-process stand-in for a replica talks to the server the + /// way a separate pod would. + /// + /// A new multiplexer. + public async Task ConnectAsync() + => await ConnectionMultiplexer.ConnectAsync(ConnectionString).ConfigureAwait(false); + + /// + public async ValueTask DisposeAsync() + { + if (_container is not null) + { + await _container.DisposeAsync().ConfigureAwait(false); + } + } + + private async Task WaitUntilReadyAsync() + { + for (var attempt = 1; ; attempt++) + { + try + { + var connection = await ConnectionMultiplexer.ConnectAsync(ConnectionString).ConfigureAwait(false); + await using (connection.ConfigureAwait(false)) + { + await connection.GetDatabase().PingAsync().ConfigureAwait(false); + return; + } + } + catch (RedisException) when (attempt < 60) + { + await Task.Delay(TimeSpan.FromSeconds(1)).ConfigureAwait(false); + } + } + } +} diff --git a/tests/Jellyfin.Server.Tests/Jellyfin.Server.Tests.csproj b/tests/Jellyfin.Server.Tests/Jellyfin.Server.Tests.csproj index a96840baa1..b5d8f219d6 100644 --- a/tests/Jellyfin.Server.Tests/Jellyfin.Server.Tests.csproj +++ b/tests/Jellyfin.Server.Tests/Jellyfin.Server.Tests.csproj @@ -12,6 +12,7 @@ + all diff --git a/tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectReplicaTests.cs b/tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectReplicaTests.cs new file mode 100644 index 0000000000..26fff9a1f1 --- /dev/null +++ b/tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectReplicaTests.cs @@ -0,0 +1,308 @@ +using System; +using System.Collections.Generic; +using System.Globalization; +using System.Threading; +using System.Threading.Tasks; +using Emby.Server.Implementations.QuickConnect; +using Jellyfin.Data.Queries; +using Jellyfin.Database.Implementations; +using Jellyfin.Database.Implementations.DbConfiguration; +using Jellyfin.Database.Implementations.Entities; +using Jellyfin.Database.Implementations.Entities.Security; +using Jellyfin.Database.Implementations.Locking; +using Jellyfin.Database.Providers.PostgreSQL; +using Jellyfin.Server.Implementations.Devices; +using Jellyfin.Server.Tests.HighAvailability; +using Jellyfin.Server.Tests.Migrations; +using MediaBrowser.Common.Extensions; +using MediaBrowser.Controller.Authentication; +using MediaBrowser.Controller.Configuration; +using MediaBrowser.Controller.Devices; +using MediaBrowser.Controller.Library; +using MediaBrowser.Controller.Net; +using MediaBrowser.Controller.QuickConnect; +using MediaBrowser.Controller.Session; +using MediaBrowser.Model.Configuration; +using MediaBrowser.Model.Dto; +using MediaBrowser.Model.QuickConnect; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.Logging.Abstractions; +using Moq; +using Npgsql; +using StackExchange.Redis; +using Xunit; + +namespace Jellyfin.Server.Tests.QuickConnect; + +/// +/// Three independently constructed instances over one PostgreSQL +/// database and one Redis are the in-process stand-in for three replicas without sticky sessions: the +/// initiate, authorize and exchange legs of one flow each land on a different one. +/// +[Trait("Category", "RequiresDocker")] +public sealed class QuickConnectReplicaTests : IAsyncLifetime +{ + private static readonly AuthorizationInfo _authorizationInfo = new AuthorizationInfo + { + Device = "Living Room TV", + DeviceId = "device-1", + Client = "Jellyfin Web", + Version = "1.0.0" + }; + + private readonly List _connections = new(); + + private PostgreSqlTestServer _postgres = null!; + private RedisTestServer _redis = null!; + + /// + public async ValueTask InitializeAsync() + { + _postgres = await PostgreSqlTestServer.StartAsync().ConfigureAwait(false); + _redis = await RedisTestServer.StartAsync().ConfigureAwait(false); + } + + /// + public async ValueTask DisposeAsync() + { + foreach (var connection in _connections) + { + await connection.DisposeAsync().ConfigureAwait(false); + } + + await _redis.DisposeAsync().ConfigureAwait(false); + await _postgres.DisposeAsync().ConfigureAwait(false); + } + + /// + /// The three legs of a quick connect flow land on three different replicas, and the token the third + /// one hands out is the one the second one minted into the shared database. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task InitiateAuthorizeExchange_AcrossThreeReplicas_Succeeds() + { + var cancellationToken = TestContext.Current.CancellationToken; + var connectionString = await _postgres.CreateDatabaseAsync("quickconnect_replica_flow", cancellationToken); + + await using var dataSource = new NpgsqlDataSourceBuilder(connectionString).Build(); + var user = await CreateSchemaWithUserAsync(dataSource, cancellationToken); + + var replicaA = await CreateReplicaAsync(dataSource, user); + var replicaB = await CreateReplicaAsync(dataSource, user); + var replicaC = await CreateReplicaAsync(dataSource, user); + + var initiated = await replicaA.Manager.TryConnect(_authorizationInfo); + + // The code is shown to the user on whichever replica serves the dashboard. + Assert.True(await replicaB.Manager.AuthorizeRequest(user.Id, initiated.Code)); + + var polled = await replicaC.Manager.CheckRequestStatus(initiated.Secret); + Assert.True(polled.Authenticated); + Assert.Equal(initiated.Code, polled.Code); + Assert.Equal(_authorizationInfo.DeviceId, polled.DeviceId); + + var exchanged = await replicaC.Manager.GetAuthorizedRequest(initiated.Secret); + + Assert.False(string.IsNullOrEmpty(exchanged.AccessToken)); + Assert.Equal(user.Id, exchanged.User.Id); + + var devices = await replicaA.Devices.GetDevices(new DeviceQuery { AccessToken = exchanged.AccessToken }); + Assert.Equal(user.Id, Assert.Single(devices.Items).UserId); + } + + /// + /// A secret is single use across the whole deployment: two replicas racing to exchange it must not + /// both hand out an access token. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task Exchange_RacedOnTwoReplicas_SucceedsOnce() + { + var cancellationToken = TestContext.Current.CancellationToken; + var connectionString = await _postgres.CreateDatabaseAsync("quickconnect_replica_race", cancellationToken); + + await using var dataSource = new NpgsqlDataSourceBuilder(connectionString).Build(); + var user = await CreateSchemaWithUserAsync(dataSource, cancellationToken); + + var replicaA = await CreateReplicaAsync(dataSource, user); + var replicaB = await CreateReplicaAsync(dataSource, user); + var replicaC = await CreateReplicaAsync(dataSource, user); + + var initiated = await replicaA.Manager.TryConnect(_authorizationInfo); + 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); + + 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)); + } + + /// + /// An expired request is rejected on a replica that never saw it created, rather than resolving to a + /// stale authorization. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task ExpiredRequest_IsRejectedOnEveryReplica() + { + var cancellationToken = TestContext.Current.CancellationToken; + var connectionString = await _postgres.CreateDatabaseAsync("quickconnect_replica_expiry", cancellationToken); + + await using var dataSource = new NpgsqlDataSourceBuilder(connectionString).Build(); + var user = await CreateSchemaWithUserAsync(dataSource, cancellationToken); + + var replicaA = await CreateReplicaAsync(dataSource, user); + var replicaB = await CreateReplicaAsync(dataSource, user); + + var initiated = await replicaA.Manager.TryConnect(_authorizationInfo); + Assert.NotNull(await replicaB.Manager.CheckRequestStatus(initiated.Secret)); + + // Shorten the stored expiry instead of waiting out the ten minute timeout. + await replicaA.Store.SetRequestAsync(initiated, DateTime.UtcNow.AddSeconds(1), cancellationToken); + await Task.Delay(TimeSpan.FromSeconds(2), cancellationToken); + + await Assert.ThrowsAsync(() => replicaB.Manager.CheckRequestStatus(initiated.Secret)); + await Assert.ThrowsAsync(() => replicaB.Manager.AuthorizeRequest(user.Id, initiated.Code)); + } + + /// + /// An authorization that was never exchanged expires too, so a code authorized and then abandoned + /// cannot be redeemed later from another replica. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task ExpiredAuthorization_IsRejectedOnEveryReplica() + { + var cancellationToken = TestContext.Current.CancellationToken; + var connectionString = await _postgres.CreateDatabaseAsync("quickconnect_replica_auth_expiry", cancellationToken); + + await using var dataSource = new NpgsqlDataSourceBuilder(connectionString).Build(); + var user = await CreateSchemaWithUserAsync(dataSource, cancellationToken); + + var replicaA = await CreateReplicaAsync(dataSource, user); + var replicaB = await CreateReplicaAsync(dataSource, user); + + var initiated = await replicaA.Manager.TryConnect(_authorizationInfo); + await replicaA.Manager.AuthorizeRequest(user.Id, initiated.Code); + + var stored = await replicaA.Store.GetRequestBySecretAsync(initiated.Secret, cancellationToken); + Assert.True(stored?.Authenticated); + + await replicaA.Store.SetAuthorizationAsync( + initiated.Secret, + new AuthenticationResult { AccessToken = "stale" }, + DateTime.UtcNow.AddSeconds(1), + cancellationToken); + await Task.Delay(TimeSpan.FromSeconds(2), cancellationToken); + + await Assert.ThrowsAsync(() => replicaB.Manager.GetAuthorizedRequest(initiated.Secret)); + } + + private static async Task ExchangeAsync(IQuickConnect manager, string secret) + { + try + { + return await manager.GetAuthorizedRequest(secret).ConfigureAwait(false); + } + catch (ResourceNotFoundException) + { + return null; + } + } + + private static async Task CreateSchemaWithUserAsync(NpgsqlDataSource dataSource, CancellationToken cancellationToken) + { + var context = CreateContext(dataSource); + await using (context.ConfigureAwait(false)) + { + await context.Database.EnsureCreatedAsync(cancellationToken).ConfigureAwait(false); + + var user = new User("quickconnect-user", "provider", "provider"); + context.Users.Add(user); + await context.SaveChangesAsync(cancellationToken).ConfigureAwait(false); + + return user; + } + } + + private static JellyfinDbContext CreateContext(NpgsqlDataSource dataSource) + { + var optionsBuilder = new DbContextOptionsBuilder(); + var provider = new PostgreSqlDatabaseProvider(dataSource); + provider.Initialise(optionsBuilder, new DatabaseConfigurationOptions { DatabaseType = "PostgreSQL" }); + return new JellyfinDbContext( + optionsBuilder.Options, + NullLogger.Instance, + provider, + new NoLockBehavior(NullLogger.Instance)); + } + + private async Task CreateReplicaAsync(NpgsqlDataSource dataSource, User user) + { + var connection = await _redis.ConnectAsync().ConfigureAwait(false); + _connections.Add(connection); + + var userManager = new Mock(); + userManager.Setup(manager => manager.GetUserById(user.Id)).Returns(user); + var deviceManager = new DeviceManager(new DataSourceContextFactory(dataSource), userManager.Object); + + var configManager = new Mock(); + configManager.Setup(manager => manager.Configuration).Returns(new ServerConfiguration { QuickConnectAvailable = true }); + + // Stands in for SessionManager.AuthenticateDirect: the token has to be minted into the shared + // database, because the replica that exchanges the secret is not the one that authorized it. + var sessionManager = new Mock(); + sessionManager + .Setup(manager => manager.AuthenticateDirect(It.IsAny())) + .Returns(async request => + { + var device = await deviceManager.CreateDevice( + new Device(request.UserId, request.App, request.AppVersion, request.DeviceName, request.DeviceId)).ConfigureAwait(false); + + return new AuthenticationResult + { + AccessToken = device.AccessToken, + ServerId = "server-1", + User = new UserDto { Id = user.Id, Name = user.Username, ServerId = "server-1" }, + SessionInfo = new SessionInfoDto + { + Id = device.Id.ToString(CultureInfo.InvariantCulture), + UserId = user.Id, + UserName = user.Username, + Client = request.App, + DeviceId = request.DeviceId, + DeviceName = request.DeviceName, + ApplicationVersion = request.AppVersion + } + }; + }); + + var store = new RedisQuickConnectStore(connection, NullLogger.Instance); + var manager = new QuickConnectManager( + configManager.Object, + NullLogger.Instance, + sessionManager.Object, + store); + + return new Replica(manager, store, deviceManager); + } + + private sealed record Replica(IQuickConnect Manager, IQuickConnectStore Store, IDeviceManager Devices); + + private sealed class DataSourceContextFactory : IDbContextFactory + { + private readonly NpgsqlDataSource _dataSource; + + public DataSourceContextFactory(NpgsqlDataSource dataSource) + { + _dataSource = dataSource; + } + + public JellyfinDbContext CreateDbContext() => CreateContext(_dataSource); + } +} From 6c54a02240cdafeeea41c3a7b0dd9512b3b28c12 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Thu, 24 Sep 2026 23:38:35 +1000 Subject: [PATCH 2/3] make quick connect authorize atomic and stop the fallback shadowing redis --- .../QuickConnect/QuickConnectManager.cs | 6 + .../QuickConnect/RedisQuickConnectStore.cs | 106 +++--- .../QuickConnect/IQuickConnectStore.cs | 11 + .../QuickConnect/InMemoryQuickConnectStore.cs | 21 ++ .../QuickConnect/QuickConnectManagerTests.cs | 25 ++ .../JellyfinSectionConfigurationTests.cs | 1 + .../HighAvailability/RedisFaultProxy.cs | 170 ++++++++++ .../QuickConnect/QuickConnectReplicaTests.cs | 79 ++++- .../QuickConnectStoreWiringTests.cs | 142 ++++++++ .../RedisQuickConnectStoreDegradedTests.cs | 308 ++++++++++++++++++ 10 files changed, 819 insertions(+), 50 deletions(-) create mode 100644 tests/Jellyfin.Server.Tests/HighAvailability/RedisFaultProxy.cs create mode 100644 tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectStoreWiringTests.cs create mode 100644 tests/Jellyfin.Server.Tests/QuickConnect/RedisQuickConnectStoreDegradedTests.cs 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); +} From e1ca272d6ca39f7bf63cf98bcae1a35d33bdfbab Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sat, 26 Sep 2026 14:28:45 +1000 Subject: [PATCH 3/3] fail quick connect closed on an unreachable valkey and restore the idempotent exchange --- .../QuickConnect/QuickConnectManager.cs | 20 +- .../QuickConnect/RedisQuickConnectStore.cs | 93 +++---- .../Controllers/QuickConnectController.cs | 8 + Jellyfin.Api/Controllers/UserController.cs | 4 + .../Middleware/ExceptionMiddleware.cs | 2 + ...ConnectStoreServiceCollectionExtensions.cs | 27 +- .../Extensions/ConflictException.cs | 36 +++ .../Extensions/ServiceUnavailableException.cs | 37 +++ .../QuickConnect/IQuickConnect.cs | 8 +- .../QuickConnect/IQuickConnectStore.cs | 21 +- .../QuickConnect/InMemoryQuickConnectStore.cs | 27 +- .../QuickConnect/QuickConnectManagerTests.cs | 20 +- .../HighAvailability/RedisFaultProxy.cs | 57 +++- .../QuickConnect/QuickConnectReplicaTests.cs | 29 +- .../QuickConnectStatusCodeTests.cs | 261 ++++++++++++++++++ .../QuickConnectStoreWiringTests.cs | 53 ++-- .../RedisQuickConnectStoreDegradedTests.cs | 206 +++++++------- 17 files changed, 664 insertions(+), 245 deletions(-) create mode 100644 MediaBrowser.Common/Extensions/ConflictException.cs create mode 100644 MediaBrowser.Common/Extensions/ServiceUnavailableException.cs create mode 100644 tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectStatusCodeTests.cs diff --git a/Emby.Server.Implementations/QuickConnect/QuickConnectManager.cs b/Emby.Server.Implementations/QuickConnect/QuickConnectManager.cs index 5ca0c7de73..2c619f4178 100644 --- a/Emby.Server.Implementations/QuickConnect/QuickConnectManager.cs +++ b/Emby.Server.Implementations/QuickConnect/QuickConnectManager.cs @@ -141,7 +141,7 @@ namespace Emby.Server.Implementations.QuickConnect if (result.Authenticated) { - throw new InvalidOperationException("Request is already authorized"); + throw new ConflictException("Request is already authorized"); } // 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. @@ -150,7 +150,7 @@ namespace Emby.Server.Implementations.QuickConnect // 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"); + throw await RefusedClaimAsync(result.Secret).ConfigureAwait(false); } var authenticationResult = await _sessionManager.AuthenticateDirect(new AuthenticationRequest @@ -177,7 +177,7 @@ namespace Emby.Server.Implementations.QuickConnect { AssertActive(); - var result = await _store.TryConsumeAuthorizationAsync(secret).ConfigureAwait(false); + var result = await _store.GetAuthorizationAsync(secret).ConfigureAwait(false); if (result is null) { throw new ResourceNotFoundException("Unable to find request"); @@ -188,6 +188,20 @@ namespace Emby.Server.Implementations.QuickConnect private static DateTime ExpiryOf(QuickConnectResult request) => request.DateAdded.AddMinutes(Timeout); + /// + /// Explains a refused claim. The claim outlives a failed mint on purpose, so it can mean either + /// that the request is authorized or that authorizing it did not finish; the two are told apart + /// by re-reading the request rather than reported as the same thing. + /// + private async Task RefusedClaimAsync(string secret) + { + var current = await _store.GetRequestBySecretAsync(secret).ConfigureAwait(false); + + return current?.Authenticated == true + ? new ConflictException("Request is already authorized") + : new ConflictException("Request is being authorized elsewhere, or an earlier attempt to authorize it did not complete. Start quick connect again for a new code."); + } + private string GenerateSecureRandom(int length = 32) { Span bytes = stackalloc byte[length]; diff --git a/Emby.Server.Implementations/QuickConnect/RedisQuickConnectStore.cs b/Emby.Server.Implementations/QuickConnect/RedisQuickConnectStore.cs index e3396be7ba..f536fdb55a 100644 --- a/Emby.Server.Implementations/QuickConnect/RedisQuickConnectStore.cs +++ b/Emby.Server.Implementations/QuickConnect/RedisQuickConnectStore.cs @@ -3,6 +3,7 @@ using System.Text.Json; using System.Threading; using System.Threading.Tasks; using Jellyfin.Extensions.Json; +using MediaBrowser.Common.Extensions; using MediaBrowser.Controller.Authentication; using MediaBrowser.Controller.QuickConnect; using MediaBrowser.Model.QuickConnect; @@ -13,20 +14,27 @@ 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, 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. +/// of a quick connect flow land on different instances. Expiry is the key TTL and an authorization is +/// claimed with a Lua check-and-set, so only one instance can ever mint a given secret's access token. /// /// -/// 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. +/// There is no local fallback: a call Redis did not answer is inconclusive, and reporting it as a miss +/// would tell a polling client its secret is invalid. Quick connect is unavailable for as long as Redis +/// is, which password login is not. /// public sealed class RedisQuickConnectStore : IQuickConnectStore { private const string KeyPrefix = "jellyfin:quickconnect:"; + /// + /// Lua script writing the two keys a request is resolvable by in one step, so it can never be + /// reachable by its secret while the code the user is reading off the screen resolves to nothing. + /// + private const string SetRequestScript = @" +redis.call('SET', KEYS[1], ARGV[1], 'PX', ARGV[3]) +redis.call('SET', KEYS[2], ARGV[2], 'PX', ARGV[3]) +return 1"; + /// /// 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 @@ -40,7 +48,6 @@ 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; /// @@ -53,41 +60,23 @@ return 0"; ArgumentNullException.ThrowIfNull(redis); _db = redis.GetDatabase(); - _fallback = new InMemoryQuickConnectStore(); _logger = logger; } /// public async Task GetRequestBySecretAsync(string secret, CancellationToken cancellationToken = default) { - RedisValue raw; - try - { - raw = await _db.StringGetAsync(RequestKey(secret)).ConfigureAwait(false); - } - catch (Exception ex) when (IsTransportFailure(ex)) - { - LogDegraded(ex); - return await _fallback.GetRequestBySecretAsync(secret, cancellationToken).ConfigureAwait(false); - } + var raw = await CallAsync(() => _db.StringGetAsync(RequestKey(secret))).ConfigureAwait(false); - // A miss is an answer rather than a transport failure, so the fallback is not consulted for it. + // Deserialization is outside the guard: a malformed stored value is a fault of its own, not Redis + // being unavailable. return raw.HasValue ? JsonSerializer.Deserialize(raw.ToString(), JsonDefaults.Options) : null; } /// public async Task GetRequestByCodeAsync(string code, CancellationToken cancellationToken = default) { - RedisValue secret; - try - { - secret = await _db.StringGetAsync(CodeKey(code)).ConfigureAwait(false); - } - catch (Exception ex) when (IsTransportFailure(ex)) - { - LogDegraded(ex); - return await _fallback.GetRequestByCodeAsync(code, cancellationToken).ConfigureAwait(false); - } + var secret = await CallAsync(() => _db.StringGetAsync(CodeKey(code))).ConfigureAwait(false); return secret.HasValue ? await GetRequestBySecretAsync(secret.ToString(), cancellationToken).ConfigureAwait(false) @@ -105,17 +94,11 @@ return 0"; return; } - try - { - var json = JsonSerializer.Serialize(request, JsonDefaults.Options); - await _db.StringSetAsync(RequestKey(request.Secret), json, ttl).ConfigureAwait(false); - await _db.StringSetAsync(CodeKey(request.Code), request.Secret, ttl).ConfigureAwait(false); - } - catch (Exception ex) when (IsTransportFailure(ex)) - { - LogDegraded(ex); - await _fallback.SetRequestAsync(request, expiresUtc, cancellationToken).ConfigureAwait(false); - } + var json = JsonSerializer.Serialize(request, JsonDefaults.Options); + await CallAsync(() => _db.ScriptEvaluateAsync( + SetRequestScript, + keys: new RedisKey[] { RequestKey(request.Secret), CodeKey(request.Code) }, + values: new RedisValue[] { json, request.Secret, (long)ttl.TotalMilliseconds })).ConfigureAwait(false); } /// @@ -127,10 +110,10 @@ return 0"; return false; } - var claimed = (long?)await _db.ScriptEvaluateAsync( + var claimed = (long?)await CallAsync(() => _db.ScriptEvaluateAsync( ClaimAuthorizationScript, keys: new RedisKey[] { RequestKey(secret), ClaimKey(secret) }, - values: new RedisValue[] { (long)ttl.TotalMilliseconds }).ConfigureAwait(false); + values: new RedisValue[] { (long)ttl.TotalMilliseconds })).ConfigureAwait(false); return claimed == 1; } @@ -145,23 +128,19 @@ return 0"; } var json = JsonSerializer.Serialize(authenticationResult, JsonDefaults.Options); - await _db.StringSetAsync(AuthorizationKey(secret), json, ttl).ConfigureAwait(false); + await CallAsync(() => _db.StringSetAsync(AuthorizationKey(secret), json, ttl)).ConfigureAwait(false); } /// - public async Task TryConsumeAuthorizationAsync(string secret, CancellationToken cancellationToken = default) + public async Task GetAuthorizationAsync(string secret, CancellationToken cancellationToken = default) { - var raw = await _db.StringGetDeleteAsync(AuthorizationKey(secret)).ConfigureAwait(false); + var raw = await CallAsync(() => _db.StringGetAsync(AuthorizationKey(secret))).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; @@ -170,6 +149,16 @@ return 0"; private static string AuthorizationKey(string secret) => KeyPrefix + "auth:" + secret; - private void LogDegraded(Exception exception) - => _logger.LogWarning(exception, "Quick connect request state could not be shared through Redis; falling back to this instance only."); + private async Task CallAsync(Func> call) + { + try + { + return await call().ConfigureAwait(false); + } + catch (Exception exception) when (exception is RedisException or RedisCommandException or TimeoutException) + { + _logger.LogError(exception, "Quick connect state could not be reached in Redis."); + throw new ServiceUnavailableException("Quick connect is temporarily unavailable.", exception); + } + } } diff --git a/Jellyfin.Api/Controllers/QuickConnectController.cs b/Jellyfin.Api/Controllers/QuickConnectController.cs index eebda12ff3..54424deb06 100644 --- a/Jellyfin.Api/Controllers/QuickConnectController.cs +++ b/Jellyfin.Api/Controllers/QuickConnectController.cs @@ -50,10 +50,12 @@ public class QuickConnectController : BaseJellyfinApiController /// /// Quick connect request successfully created. /// Quick connect is not active on this server. + /// Quick connect state is unavailable. /// A with a secret and code for future use or an error message. [HttpPost("Initiate")] [ProducesResponseType(StatusCodes.Status200OK)] [ProducesResponseType(StatusCodes.Status401Unauthorized)] + [ProducesResponseType(StatusCodes.Status503ServiceUnavailable)] public async Task> InitiateQuickConnect() { try @@ -73,10 +75,12 @@ public class QuickConnectController : BaseJellyfinApiController /// Secret previously returned from the Initiate endpoint. /// Quick connect result returned. /// Unknown quick connect secret. + /// Quick connect state is unavailable. /// An updated . [HttpGet("Connect")] [ProducesResponseType(StatusCodes.Status200OK)] [ProducesResponseType(StatusCodes.Status404NotFound)] + [ProducesResponseType(StatusCodes.Status503ServiceUnavailable)] public async Task> GetQuickConnectState([FromQuery, Required] string secret) { try @@ -100,11 +104,15 @@ public class QuickConnectController : BaseJellyfinApiController /// The user the authorize. Access to the requested user is required. /// Quick connect result authorized successfully. /// Unknown user id. + /// Request is already authorized, or authorizing it did not complete. + /// Quick connect state is unavailable. /// Boolean indicating if the authorization was successful. [HttpPost("Authorize")] [Authorize] [ProducesResponseType(StatusCodes.Status200OK)] [ProducesResponseType(StatusCodes.Status403Forbidden)] + [ProducesResponseType(StatusCodes.Status409Conflict)] + [ProducesResponseType(StatusCodes.Status503ServiceUnavailable)] public async Task> AuthorizeQuickConnect([FromQuery, Required] string code, [FromQuery] Guid? userId = null) { userId = RequestHelpers.GetUserId(User, userId); diff --git a/Jellyfin.Api/Controllers/UserController.cs b/Jellyfin.Api/Controllers/UserController.cs index c8aeb616a4..48e473557b 100644 --- a/Jellyfin.Api/Controllers/UserController.cs +++ b/Jellyfin.Api/Controllers/UserController.cs @@ -241,9 +241,13 @@ public class UserController : BaseJellyfinApiController /// The request. /// User authenticated. /// Missing token. + /// Unknown or unauthorized quick connect secret. + /// Quick connect state is unavailable. /// A containing an with information about the new session. [HttpPost("AuthenticateWithQuickConnect")] [ProducesResponseType(StatusCodes.Status200OK)] + [ProducesResponseType(StatusCodes.Status404NotFound)] + [ProducesResponseType(StatusCodes.Status503ServiceUnavailable)] [Tags("Authentication")] public async Task> AuthenticateWithQuickConnect([FromBody, Required] QuickConnectDto request) { diff --git a/Jellyfin.Api/Middleware/ExceptionMiddleware.cs b/Jellyfin.Api/Middleware/ExceptionMiddleware.cs index acbb4877d4..8f5da04575 100644 --- a/Jellyfin.Api/Middleware/ExceptionMiddleware.cs +++ b/Jellyfin.Api/Middleware/ExceptionMiddleware.cs @@ -131,6 +131,8 @@ public class ExceptionMiddleware FileNotFoundException => StatusCodes.Status404NotFound, ResourceNotFoundException => StatusCodes.Status404NotFound, MethodNotAllowedException => StatusCodes.Status405MethodNotAllowed, + ConflictException => StatusCodes.Status409Conflict, + ServiceUnavailableException => StatusCodes.Status503ServiceUnavailable, _ => StatusCodes.Status500InternalServerError }; } diff --git a/Jellyfin.Server/Extensions/QuickConnectStoreServiceCollectionExtensions.cs b/Jellyfin.Server/Extensions/QuickConnectStoreServiceCollectionExtensions.cs index 36f270e66e..aa13634a9e 100644 --- a/Jellyfin.Server/Extensions/QuickConnectStoreServiceCollectionExtensions.cs +++ b/Jellyfin.Server/Extensions/QuickConnectStoreServiceCollectionExtensions.cs @@ -20,7 +20,9 @@ public static class QuickConnectStoreServiceCollectionExtensions /// /// /// The connection string is only set for a multi-instance deployment, which is the only shape where - /// the initiate, authorize and exchange legs of one flow can land on different instances. + /// the initiate, authorize and exchange legs of one flow can land on different instances. Set but + /// unreachable is a misconfigured deployment rather than a single-instance one, so it fails rather + /// than quietly handing out a store the other instances cannot see. /// /// The service collection. /// The configuration to read the Redis connection string from. @@ -48,25 +50,8 @@ public static class QuickConnectStoreServiceCollectionExtensions "Quick connect store: {Store}. Quick connect flows complete across any instance.", nameof(RedisQuickConnectStore)); - return serviceCollection.AddSingleton(sp => - { - try - { - return new RedisQuickConnectStore( - sp.GetRequiredService(), - sp.GetRequiredService>()); - } - catch (Exception ex) - { - // Fail open: an unreachable Redis degrades to the single-instance behaviour of a flow - // having to complete against one instance, rather than taking quick connect down. - sp.GetRequiredService>().LogError( - ex, - "Redis is configured but unavailable, so quick connect flows will not complete across instances. Check {Key}.", - TranscodeStoreOptions.RedisConnectionStringKey); - - return new InMemoryQuickConnectStore(); - } - }); + return serviceCollection.AddSingleton(sp => new RedisQuickConnectStore( + sp.GetRequiredService(), + sp.GetRequiredService>())); } } diff --git a/MediaBrowser.Common/Extensions/ConflictException.cs b/MediaBrowser.Common/Extensions/ConflictException.cs new file mode 100644 index 0000000000..433fc13643 --- /dev/null +++ b/MediaBrowser.Common/Extensions/ConflictException.cs @@ -0,0 +1,36 @@ +using System; + +namespace MediaBrowser.Common.Extensions +{ + /// + /// Thrown when the current state of a resource does not allow the requested operation. + /// + public class ConflictException : Exception + { + /// + /// Initializes a new instance of the class. + /// + public ConflictException() + { + } + + /// + /// Initializes a new instance of the class. + /// + /// The message. + public ConflictException(string message) + : base(message) + { + } + + /// + /// Initializes a new instance of the class. + /// + /// The message. + /// The exception that caused this one. + public ConflictException(string message, Exception innerException) + : base(message, innerException) + { + } + } +} diff --git a/MediaBrowser.Common/Extensions/ServiceUnavailableException.cs b/MediaBrowser.Common/Extensions/ServiceUnavailableException.cs new file mode 100644 index 0000000000..34fc8049d6 --- /dev/null +++ b/MediaBrowser.Common/Extensions/ServiceUnavailableException.cs @@ -0,0 +1,37 @@ +using System; + +namespace MediaBrowser.Common.Extensions +{ + /// + /// Thrown when an operation cannot be answered because a backing service is unreachable, rather than + /// because the thing it was asked about does not exist. + /// + public class ServiceUnavailableException : Exception + { + /// + /// Initializes a new instance of the class. + /// + public ServiceUnavailableException() + { + } + + /// + /// Initializes a new instance of the class. + /// + /// The message. + public ServiceUnavailableException(string message) + : base(message) + { + } + + /// + /// Initializes a new instance of the class. + /// + /// The message. + /// The exception that caused this one. + public ServiceUnavailableException(string message, Exception innerException) + : base(message, innerException) + { + } + } +} diff --git a/MediaBrowser.Controller/QuickConnect/IQuickConnect.cs b/MediaBrowser.Controller/QuickConnect/IQuickConnect.cs index b585c30dbd..eba710cd81 100644 --- a/MediaBrowser.Controller/QuickConnect/IQuickConnect.cs +++ b/MediaBrowser.Controller/QuickConnect/IQuickConnect.cs @@ -1,5 +1,6 @@ using System; using System.Threading.Tasks; +using MediaBrowser.Common.Extensions; using MediaBrowser.Controller.Authentication; using MediaBrowser.Controller.Net; using MediaBrowser.Model.QuickConnect; @@ -31,7 +32,9 @@ namespace MediaBrowser.Controller.QuickConnect Task CheckRequestStatus(string secret); /// - /// Authorizes a quick connect request to connect as the calling user. + /// Authorizes a quick connect request to connect as the calling user. A request can be authorized + /// once: a second attempt, including one following an attempt that failed part way, throws + /// and the user has to start quick connect again for a new code. /// /// User id. /// Identifying code for the request. @@ -39,7 +42,8 @@ namespace MediaBrowser.Controller.QuickConnect Task AuthorizeRequest(Guid userId, string code); /// - /// Gets the authorized request for the secret. + /// Gets the authorized request for the secret. The read does not consume the authorization, so a + /// client that retries the exchange gets the same access token until the authorization expires. /// /// The secret. /// The authentication result. diff --git a/MediaBrowser.Controller/QuickConnect/IQuickConnectStore.cs b/MediaBrowser.Controller/QuickConnect/IQuickConnectStore.cs index 4bc30d9e3d..fd2254eb17 100644 --- a/MediaBrowser.Controller/QuickConnect/IQuickConnectStore.cs +++ b/MediaBrowser.Controller/QuickConnect/IQuickConnectStore.cs @@ -1,6 +1,7 @@ using System; using System.Threading; using System.Threading.Tasks; +using MediaBrowser.Common.Extensions; using MediaBrowser.Controller.Authentication; using MediaBrowser.Model.QuickConnect; @@ -11,6 +12,10 @@ namespace MediaBrowser.Controller.QuickConnect; /// initiate, authorize and exchange - can each land on a different instance, so the state has to be /// reachable from all of them. /// +/// +/// A shared implementation that cannot reach its backend throws +/// rather than reporting a miss, because a miss tells a polling client its secret is invalid. +/// public interface IQuickConnectStore { /// @@ -30,7 +35,8 @@ public interface IQuickConnectStore Task GetRequestByCodeAsync(string code, CancellationToken cancellationToken = default); /// - /// Stores a new or updated request until . + /// Stores a new or updated request until , resolvable by both its secret + /// and its code or by neither. A request already past is not stored. /// /// The request to store. /// The instant the request stops being resolvable. @@ -41,12 +47,13 @@ public interface IQuickConnectStore /// /// 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. + /// released: a mint that failed after writing its token would otherwise be retried into a second one, + /// so a request whose authorization failed has to be started again. /// /// 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. + /// true when this caller may go on to authorize the request; false when it is unknown, expired, already authorized or claimed elsewhere. Task TryClaimAuthorizationAsync(string secret, DateTime expiresUtc, CancellationToken cancellationToken = default); /// @@ -60,11 +67,11 @@ public interface IQuickConnectStore Task SetAuthorizationAsync(string secret, AuthenticationResult authenticationResult, DateTime expiresUtc, CancellationToken cancellationToken = default); /// - /// Atomically takes the authentication for and removes it, so that two - /// instances racing on the same secret cannot both hand out an access token. + /// Reads the authentication for . The read does not consume it, so a client + /// that retries an exchange gets the same access token for as long as the authentication lives. /// /// The request secret. /// A cancellation token. - /// The authentication, or null when the secret is unknown, expired or already exchanged. - Task TryConsumeAuthorizationAsync(string secret, CancellationToken cancellationToken = default); + /// The authentication, or null when the secret is unknown or has expired. + Task GetAuthorizationAsync(string secret, CancellationToken cancellationToken = default); } diff --git a/MediaBrowser.Controller/QuickConnect/InMemoryQuickConnectStore.cs b/MediaBrowser.Controller/QuickConnect/InMemoryQuickConnectStore.cs index 2296449406..0eef358730 100644 --- a/MediaBrowser.Controller/QuickConnect/InMemoryQuickConnectStore.cs +++ b/MediaBrowser.Controller/QuickConnect/InMemoryQuickConnectStore.cs @@ -9,9 +9,9 @@ using MediaBrowser.Model.QuickConnect; namespace MediaBrowser.Controller.QuickConnect; /// -/// A process-local . It is the single-instance default, and the -/// fallback a shared store degrades to while its backend is unreachable, so quick connect keeps -/// working for clients whose three legs happen to land on one instance. +/// A process-local , and the single-instance default. Nothing it holds +/// is visible to another instance, so a deployment running more than one has to configure a shared +/// store instead. /// public sealed class InMemoryQuickConnectStore : IQuickConnectStore { @@ -41,7 +41,11 @@ public sealed class InMemoryQuickConnectStore : IQuickConnectStore ArgumentNullException.ThrowIfNull(request); Expire(); - _requests[request.Secret] = new Entry(expiresUtc, request); + if (expiresUtc > DateTime.UtcNow) + { + _requests[request.Secret] = new Entry(expiresUtc, request); + } + return Task.CompletedTask; } @@ -61,20 +65,19 @@ public sealed class InMemoryQuickConnectStore : IQuickConnectStore public Task SetAuthorizationAsync(string secret, AuthenticationResult authenticationResult, DateTime expiresUtc, CancellationToken cancellationToken = default) { Expire(); - _authorizations[secret] = new Entry(expiresUtc, authenticationResult); + if (expiresUtc > DateTime.UtcNow) + { + _authorizations[secret] = new Entry(expiresUtc, authenticationResult); + } + return Task.CompletedTask; } /// - public Task TryConsumeAuthorizationAsync(string secret, CancellationToken cancellationToken = default) + public Task GetAuthorizationAsync(string secret, CancellationToken cancellationToken = default) { Expire(); - if (!_authorizations.TryRemove(secret, out var entry) || entry.ExpiresUtc <= DateTime.UtcNow) - { - return Task.FromResult(null); - } - - return Task.FromResult(entry.Value); + return Task.FromResult(_authorizations.TryGetValue(secret, out var entry) ? entry.Value : null); } private void Expire() diff --git a/tests/Jellyfin.Server.Implementations.Tests/QuickConnect/QuickConnectManagerTests.cs b/tests/Jellyfin.Server.Implementations.Tests/QuickConnect/QuickConnectManagerTests.cs index 31eac9bba0..512a42ed77 100644 --- a/tests/Jellyfin.Server.Implementations.Tests/QuickConnect/QuickConnectManagerTests.cs +++ b/tests/Jellyfin.Server.Implementations.Tests/QuickConnect/QuickConnectManagerTests.cs @@ -154,14 +154,26 @@ namespace Jellyfin.Server.Implementations.Tests.QuickConnect } [Fact] - public async Task GetAuthorizedRequest_SecondExchange_ThrowsResourceNotFoundException() + public async Task GetAuthorizedRequest_SecondExchange_ReturnsTheSameResult() { _config.QuickConnectAvailable = true; var res = await _quickConnectManager.TryConnect(_quickConnectAuthInfo); await _quickConnectManager.AuthorizeRequest(Guid.Empty, res.Code); - Assert.NotNull(await _quickConnectManager.GetAuthorizedRequest(res.Secret)); - await Assert.ThrowsAsync(() => _quickConnectManager.GetAuthorizedRequest(res.Secret)); + var first = await _quickConnectManager.GetAuthorizedRequest(res.Secret); + var second = await _quickConnectManager.GetAuthorizedRequest(res.Secret); + + Assert.Same(first, second); + } + + [Fact] + public async Task AuthorizeRequest_OfAnAuthorizedRequest_ThrowsConflictException() + { + _config.QuickConnectAvailable = true; + var res = await _quickConnectManager.TryConnect(_quickConnectAuthInfo); + await _quickConnectManager.AuthorizeRequest(Guid.Empty, res.Code); + + await Assert.ThrowsAsync(() => _quickConnectManager.AuthorizeRequest(Guid.Empty, res.Code)); } private async Task AuthorizeAsync(string code) @@ -170,7 +182,7 @@ namespace Jellyfin.Server.Implementations.Tests.QuickConnect { return await _quickConnectManager.AuthorizeRequest(Guid.Empty, code).ConfigureAwait(false); } - catch (InvalidOperationException) + catch (ConflictException) { return false; } diff --git a/tests/Jellyfin.Server.Tests/HighAvailability/RedisFaultProxy.cs b/tests/Jellyfin.Server.Tests/HighAvailability/RedisFaultProxy.cs index 14d6c1a22c..6c4c6cb4a3 100644 --- a/tests/Jellyfin.Server.Tests/HighAvailability/RedisFaultProxy.cs +++ b/tests/Jellyfin.Server.Tests/HighAvailability/RedisFaultProxy.cs @@ -4,6 +4,7 @@ using System.Globalization; using System.IO; using System.Net; using System.Net.Sockets; +using System.Text; using System.Threading; using System.Threading.Tasks; using StackExchange.Redis; @@ -25,6 +26,7 @@ public sealed class RedisFaultProxy : IAsyncDisposable private readonly int _port; private volatile bool _cut; + private volatile byte[]? _cutAfterMarker; private RedisFaultProxy(TcpListener listener, int port, string targetHost, int targetPort) { @@ -74,11 +76,23 @@ public sealed class RedisFaultProxy : IAsyncDisposable DropLiveConnections(); } + /// + /// Arms a cut for the moment after a command containing has been forwarded + /// and answered, so a test can take Redis away between two round trips of one operation rather than + /// only before or after all of them. + /// + /// Text that identifies the command to cut after. + public void CutAfterForwarding(string marker) => _cutAfterMarker = Encoding.UTF8.GetBytes(marker); + /// /// 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 void Restore() + { + _cutAfterMarker = null; + _cut = false; + } /// public async ValueTask DisposeAsync() @@ -136,10 +150,17 @@ public sealed class RedisFaultProxy : IAsyncDisposable _live[client] = 0; _live[upstream] = 0; + // Registered first, then rechecked: a cut concurrent with this connect would otherwise drop + // the live connections before this pair joined them and leave it running through the outage. + if (_cut) + { + return; + } + var clientStream = client.GetStream(); var upstreamStream = upstream.GetStream(); await Task.WhenAny( - CopyAsync(clientStream, upstreamStream), + CopyFromClientAsync(clientStream, upstreamStream), CopyAsync(upstreamStream, clientStream)).ConfigureAwait(false); } catch (Exception exception) when (exception is IOException or SocketException or OperationCanceledException or ObjectDisposedException) @@ -157,6 +178,38 @@ public sealed class RedisFaultProxy : IAsyncDisposable } } + private async Task CopyFromClientAsync(NetworkStream from, NetworkStream to) + { + var buffer = new byte[16 * 1024]; + try + { + while (true) + { + var read = await from.ReadAsync(buffer, _cts.Token).ConfigureAwait(false); + if (read == 0) + { + return; + } + + await to.WriteAsync(buffer.AsMemory(0, read), _cts.Token).ConfigureAwait(false); + + var marker = _cutAfterMarker; + if (marker is not null && buffer.AsSpan(0, read).IndexOf(marker) >= 0) + { + _cutAfterMarker = null; + + // Long enough for the server to have applied the command that was just forwarded. + await Task.Delay(TimeSpan.FromMilliseconds(250), _cts.Token).ConfigureAwait(false); + Cut(); + return; + } + } + } + catch (Exception exception) when (exception is IOException or SocketException or OperationCanceledException or ObjectDisposedException) + { + } + } + private async Task CopyAsync(NetworkStream from, NetworkStream to) { try diff --git a/tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectReplicaTests.cs b/tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectReplicaTests.cs index 9d8c3d2d49..31341139cc 100644 --- a/tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectReplicaTests.cs +++ b/tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectReplicaTests.cs @@ -112,16 +112,15 @@ 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. One scheduling of one race settles nothing either way, so the race - /// is run repeatedly. + /// Exchanging a secret does not spend it: a client that retries, or whose retry lands on another + /// replica, gets the same access token back rather than a 404, and the device is minted once. /// /// A representing the asynchronous operation. [Fact] - public async Task Exchange_RacedOnTwoReplicas_SucceedsOnce() + public async Task Exchange_RepeatedOnTwoReplicas_ReturnsTheSameToken() { var cancellationToken = TestContext.Current.CancellationToken; - var connectionString = await _postgres.CreateDatabaseAsync("quickconnect_replica_race", cancellationToken); + var connectionString = await _postgres.CreateDatabaseAsync("quickconnect_replica_reexchange", cancellationToken); await using var dataSource = new NpgsqlDataSourceBuilder(connectionString).Build(); var user = await CreateSchemaWithUserAsync(dataSource, cancellationToken); @@ -130,19 +129,25 @@ public sealed class QuickConnectReplicaTests : IAsyncLifetime var replicaB = await CreateReplicaAsync(dataSource, user); var replicaC = await CreateReplicaAsync(dataSource, user); - for (var attempt = 0; attempt < 25; attempt++) + for (var attempt = 0; attempt < 10; attempt++) { - var initiated = await replicaA.Manager.TryConnect(AuthorizationInfoFor(attempt)); + var authorizationInfo = AuthorizationInfoFor(attempt); + var initiated = await replicaA.Manager.TryConnect(authorizationInfo); await replicaB.Manager.AuthorizeRequest(user.Id, initiated.Code); - var outcomes = await Task.WhenAll( + var exchanged = 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.All(exchanged, outcome => Assert.NotNull(outcome)); + Assert.Equal(exchanged[0]!.AccessToken, exchanged[1]!.AccessToken); - // And it stays consumed for every later attempt, on any replica. - await Assert.ThrowsAsync(() => replicaB.Manager.GetAuthorizedRequest(initiated.Secret)); + // Still there afterwards, on a replica that has not exchanged it yet. + var later = await replicaB.Manager.GetAuthorizedRequest(initiated.Secret); + Assert.Equal(exchanged[0]!.AccessToken, later.AccessToken); + + var devices = await replicaA.Devices.GetDevices(new DeviceQuery { DeviceId = authorizationInfo.DeviceId }); + Assert.Equal(later.AccessToken, Assert.Single(devices.Items).AccessToken); } } @@ -258,7 +263,7 @@ public sealed class QuickConnectReplicaTests : IAsyncLifetime { return await manager.AuthorizeRequest(userId, code).ConfigureAwait(false); } - catch (InvalidOperationException) + catch (ConflictException) { return false; } diff --git a/tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectStatusCodeTests.cs b/tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectStatusCodeTests.cs new file mode 100644 index 0000000000..55d3284de7 --- /dev/null +++ b/tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectStatusCodeTests.cs @@ -0,0 +1,261 @@ +using System; +using System.IO; +using System.Threading.Tasks; +using Emby.Server.Implementations.QuickConnect; +using Jellyfin.Api.Controllers; +using Jellyfin.Api.Middleware; +using Jellyfin.Server.Tests.HighAvailability; +using MediaBrowser.Common.Extensions; +using MediaBrowser.Controller; +using MediaBrowser.Controller.Authentication; +using MediaBrowser.Controller.Configuration; +using MediaBrowser.Controller.Net; +using MediaBrowser.Controller.QuickConnect; +using MediaBrowser.Controller.Session; +using MediaBrowser.Model.Configuration; +using MediaBrowser.Model.Dto; +using MediaBrowser.Model.QuickConnect; +using Microsoft.AspNetCore.Hosting; +using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.Mvc; +using Microsoft.AspNetCore.Mvc.Infrastructure; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging.Abstractions; +using Moq; +using StackExchange.Redis; +using Xunit; + +namespace Jellyfin.Server.Tests.QuickConnect; + +/// +/// The status a client actually sees. A missing key and an unreachable valkey are different answers and +/// must not collapse into one: telling a polling client its secret is unknown ends its flow, while 503 +/// tells it to keep trying. The exception the store really throws is run through the real exception +/// middleware, so the mapping is exercised rather than assumed. +/// +[Trait("Category", "RequiresDocker")] +public sealed class QuickConnectStatusCodeTests : IAsyncLifetime +{ + private readonly Mock _sessionManager = new(); + + private RedisTestServer _redis = null!; + private RedisFaultProxy _proxy = null!; + private IConnectionMultiplexer _connection = null!; + private QuickConnectManager _manager = null!; + private QuickConnectController _controller = null!; + + /// + public async ValueTask InitializeAsync() + { + _redis = await RedisTestServer.StartAsync().ConfigureAwait(false); + _proxy = RedisFaultProxy.Start(_redis.ConnectionString); + _connection = await ConnectionMultiplexer.ConnectAsync(_proxy.ConnectionString).ConfigureAwait(false); + + var configManager = new Mock(); + configManager.Setup(manager => manager.Configuration).Returns(new ServerConfiguration { QuickConnectAvailable = true }); + + _manager = new QuickConnectManager( + configManager.Object, + NullLogger.Instance, + _sessionManager.Object, + new RedisQuickConnectStore(_connection, NullLogger.Instance)); + + _controller = new QuickConnectController(_manager, Mock.Of()); + } + + /// + public async ValueTask DisposeAsync() + { + await _connection.DisposeAsync().ConfigureAwait(false); + await _proxy.DisposeAsync().ConfigureAwait(false); + await _redis.DisposeAsync().ConfigureAwait(false); + } + + /// + /// A secret valkey has never heard of is a 404, which is what ends a flow the user abandoned. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task Poll_UnknownSecret_IsNotFound() + { + Assert.Equal( + StatusCodes.Status404NotFound, + await StatusCodeAsync(async () => StatusOf(await _controller.GetQuickConnectState(NewSecret())))); + } + + /// + /// The same poll while valkey is unreachable is a 503. This is the bug the shared store is here to + /// avoid: a blip must not tell every polling client that its secret is invalid. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task Poll_WhileRedisIsUnreachable_IsServiceUnavailable() + { + var secret = await InitiateAsync(); + _proxy.Cut(); + + Assert.Equal( + StatusCodes.Status503ServiceUnavailable, + await StatusCodeAsync(async () => StatusOf(await _controller.GetQuickConnectState(secret)))); + } + + /// + /// The exchange leg tells the two apart the same way. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task Exchange_UnknownSecret_IsNotFound() + { + Assert.Equal( + StatusCodes.Status404NotFound, + await StatusCodeAsync(async () => + { + await _manager.GetAuthorizedRequest(NewSecret()).ConfigureAwait(false); + return StatusCodes.Status200OK; + })); + } + + /// + /// The exchange leg while valkey is unreachable. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task Exchange_WhileRedisIsUnreachable_IsServiceUnavailable() + { + var secret = await InitiateAsync(); + _proxy.Cut(); + + Assert.Equal( + StatusCodes.Status503ServiceUnavailable, + await StatusCodeAsync(async () => + { + await _manager.GetAuthorizedRequest(secret).ConfigureAwait(false); + return StatusCodes.Status200OK; + })); + } + + /// + /// The authorize leg while valkey is unreachable. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task Authorize_WhileRedisIsUnreachable_IsServiceUnavailable() + { + var initiated = await _manager.TryConnect(AuthorizationInfo()); + _proxy.Cut(); + + Assert.Equal( + StatusCodes.Status503ServiceUnavailable, + await StatusCodeAsync(async () => + { + await _manager.AuthorizeRequest(Guid.NewGuid(), initiated.Code).ConfigureAwait(false); + return StatusCodes.Status200OK; + })); + } + + /// + /// A mint that threw leaves its claim taken on purpose, because the write it failed on may have + /// landed. Retrying then has to say so and be a 409 the client can act on, not a 500 and not the + /// untrue claim that the request is already authorized. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task Authorize_AfterAMintThatFailed_IsConflictAndSaysToStartAgain() + { + var initiated = await _manager.TryConnect(AuthorizationInfo()); + + _sessionManager + .Setup(manager => manager.AuthenticateDirect(It.IsAny())) + .ThrowsAsync(new InvalidOperationException("mint failed")); + + await Assert.ThrowsAsync(() => _manager.AuthorizeRequest(Guid.NewGuid(), initiated.Code)); + + var retry = await Record.ExceptionAsync(() => _manager.AuthorizeRequest(Guid.NewGuid(), initiated.Code)); + + var conflict = Assert.IsType(retry); + Assert.DoesNotContain("already authorized", conflict.Message, StringComparison.Ordinal); + Assert.Contains("Start quick connect again", conflict.Message, StringComparison.Ordinal); + + Assert.Equal( + StatusCodes.Status409Conflict, + await StatusCodeAsync(async () => + { + await _manager.AuthorizeRequest(Guid.NewGuid(), initiated.Code).ConfigureAwait(false); + return StatusCodes.Status200OK; + })); + } + + /// + /// A request that really was authorized still says so, so the accurate message above is not just a + /// blanket replacement for the old one. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task Authorize_OfAnAuthorizedRequest_IsConflictAndSaysAlreadyAuthorized() + { + var initiated = await _manager.TryConnect(AuthorizationInfo()); + var userId = Guid.NewGuid(); + + _sessionManager + .Setup(manager => manager.AuthenticateDirect(It.IsAny())) + .ReturnsAsync(new AuthenticationResult + { + AccessToken = "token-1", + ServerId = "server-1", + User = new UserDto { Id = userId, Name = "user", ServerId = "server-1" } + }); + + Assert.True(await _manager.AuthorizeRequest(userId, initiated.Code)); + + var retry = await Record.ExceptionAsync(() => _manager.AuthorizeRequest(userId, initiated.Code)); + + Assert.Equal("Request is already authorized", Assert.IsType(retry).Message); + } + + private static AuthorizationInfo AuthorizationInfo() => new AuthorizationInfo + { + Device = "Living Room TV", + DeviceId = Guid.NewGuid().ToString("N"), + Client = "Jellyfin Web", + Version = "1.0.0" + }; + + private static string NewSecret() => Guid.NewGuid().ToString("N"); + + private static int StatusOf(ActionResult result) + => result.Result is IStatusCodeActionResult status + ? status.StatusCode ?? StatusCodes.Status200OK + : StatusCodes.Status200OK; + + private static async Task StatusCodeAsync(Func> action) + { + var appPaths = new Mock(); + appPaths.Setup(paths => paths.ProgramSystemPath).Returns("/program"); + appPaths.Setup(paths => paths.ProgramDataPath).Returns("/data"); + + var configManager = new Mock(); + configManager.Setup(manager => manager.ApplicationPaths).Returns(appPaths.Object); + + var hostEnvironment = new Mock(); + hostEnvironment.SetupGet(environment => environment.EnvironmentName).Returns(Environments.Production); + + var context = new DefaultHttpContext(); + context.Response.Body = new MemoryStream(); + + var middleware = new ExceptionMiddleware( + async _ => + { + context.Response.StatusCode = await action().ConfigureAwait(false); + }, + NullLogger.Instance, + configManager.Object, + hostEnvironment.Object); + + await middleware.Invoke(context).ConfigureAwait(false); + + return context.Response.StatusCode; + } + + private async Task InitiateAsync() + => (await _manager.TryConnect(AuthorizationInfo()).ConfigureAwait(false)).Secret; +} diff --git a/tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectStoreWiringTests.cs b/tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectStoreWiringTests.cs index fb58e45374..b492f44d98 100644 --- a/tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectStoreWiringTests.cs +++ b/tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectStoreWiringTests.cs @@ -59,7 +59,7 @@ public sealed class QuickConnectStoreWiringTests : IAsyncLifetime [Fact] public async Task ManifestStyleEnvironmentVariable_SelectsTheSharedStore() { - Environment.SetEnvironmentVariable(RedisConnectionStringVariable, _redis.ConnectionString + ",abortConnect=false"); + Environment.SetEnvironmentVariable(RedisConnectionStringVariable, _redis.ConnectionString); await using var provider = BuildProvider(); @@ -76,45 +76,50 @@ public sealed class QuickConnectStoreWiringTests : IAsyncLifetime } /// - /// 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. + /// Without the variable the deployment is single-instance and gets the process-local store, which + /// runs a whole flow on its own. /// /// A representing the asynchronous operation. [Fact] - public async Task UnreachableRedisAtStartup_DegradesToTheProcessLocalStore() + public async Task NoEnvironmentVariable_SelectsTheProcessLocalStore() { - Environment.SetEnvironmentVariable(RedisConnectionStringVariable, "127.0.0.1:1,connectTimeout=250,connectRetry=0"); + Environment.SetEnvironmentVariable(RedisConnectionStringVariable, null); await using var provider = BuildProvider(); var store = provider.GetRequiredService(); Assert.IsType(store); - // Quick connect still works, it just cannot span instances. + var cancellationToken = TestContext.Current.CancellationToken; 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.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), cancellationToken); + + Assert.Equal(request.Secret, (await store.GetRequestByCodeAsync(request.Code, cancellationToken))?.Secret); + Assert.True(await store.TryClaimAuthorizationAsync(request.Secret, DateTime.UtcNow.AddMinutes(10), cancellationToken)); + await store.SetAuthorizationAsync( request.Secret, new AuthenticationResult { AccessToken = "token-1" }, DateTime.UtcNow.AddMinutes(10), - TestContext.Current.CancellationToken); + cancellationToken); - Assert.Equal("token-1", (await store.TryConsumeAuthorizationAsync(request.Secret, TestContext.Current.CancellationToken))?.AccessToken); - Assert.Null(await store.TryConsumeAuthorizationAsync(request.Secret, TestContext.Current.CancellationToken)); + Assert.Equal("token-1", (await store.GetAuthorizationAsync(request.Secret, cancellationToken))?.AccessToken); + Assert.Equal("token-1", (await store.GetAuthorizationAsync(request.Secret, cancellationToken))?.AccessToken); + } + + /// + /// A connection string that is set but unreachable is a misconfigured multi-instance deployment. It + /// fails rather than handing out a store the other instances cannot see, which would put quick + /// connect back on the cross-instance behaviour this configuration exists to fix. + /// + [Fact] + public void UnreachableRedisAtStartup_FailsClosed() + { + Environment.SetEnvironmentVariable(RedisConnectionStringVariable, "127.0.0.1:1,connectTimeout=250,connectRetry=0"); + + using var provider = BuildProvider(); + + Assert.ThrowsAny(() => provider.GetRequiredService()); } private static QuickConnectResult NewRequest() => new QuickConnectResult( diff --git a/tests/Jellyfin.Server.Tests/QuickConnect/RedisQuickConnectStoreDegradedTests.cs b/tests/Jellyfin.Server.Tests/QuickConnect/RedisQuickConnectStoreDegradedTests.cs index 3c07faeae7..bfd4999705 100644 --- a/tests/Jellyfin.Server.Tests/QuickConnect/RedisQuickConnectStoreDegradedTests.cs +++ b/tests/Jellyfin.Server.Tests/QuickConnect/RedisQuickConnectStoreDegradedTests.cs @@ -5,7 +5,9 @@ using System.Threading; using System.Threading.Tasks; using Emby.Server.Implementations.QuickConnect; using Jellyfin.Server.Tests.HighAvailability; +using MediaBrowser.Common.Extensions; using MediaBrowser.Controller.Authentication; +using MediaBrowser.Controller.QuickConnect; using MediaBrowser.Model.QuickConnect; using Microsoft.Extensions.Logging.Abstractions; using StackExchange.Redis; @@ -51,58 +53,83 @@ public sealed class RedisQuickConnectStoreDegradedTests : IAsyncLifetime } /// - /// 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. + /// Every leg of the flow fails closed while Redis is unreachable. None of them may answer as though + /// Redis had said the request is unknown, because that tells a polling client its secret is invalid. /// /// A representing the asynchronous operation. [Fact] - public async Task PendingRequest_SurvivesAnOutage_OnTheInstanceThatStoredIt() + public async Task EveryLeg_WhileRedisIsUnreachable_ReportsUnavailable() { var instance = await CreateInstanceAsync(); var request = NewRequest(); + var expiresUtc = DateTime.UtcNow.AddMinutes(10); 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); + await AssertUnavailableAsync(() => instance.Store.SetRequestAsync(request, expiresUtc, CancellationToken)); + await AssertUnavailableAsync(() => instance.Store.GetRequestBySecretAsync(request.Secret, CancellationToken)); + await AssertUnavailableAsync(() => instance.Store.GetRequestByCodeAsync(request.Code, CancellationToken)); + await AssertUnavailableAsync(() => instance.Store.TryClaimAuthorizationAsync(request.Secret, expiresUtc, CancellationToken)); + await AssertUnavailableAsync(() => instance.Store.SetAuthorizationAsync( + request.Secret, + new AuthenticationResult { AccessToken = "token-1" }, + expiresUtc, + CancellationToken)); + await AssertUnavailableAsync(() => instance.Store.GetAuthorizationAsync(request.Secret, CancellationToken)); } /// - /// 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. + /// The same three reads against a Redis that is answering report a genuine miss as a miss, which is + /// what makes an outage and an unknown secret tellable apart by the callers above. /// /// A representing the asynchronous operation. [Fact] - public async Task PendingRequest_StoredDuringAnOutage_IsNotServedOnceRedisAnswersAgain() + public async Task EveryRead_AgainstAHealthyRedis_ReportsAMissAsAMiss() { 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)); + Assert.Null(await instance.Store.GetAuthorizationAsync(request.Secret, 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 request is resolvable by its secret and by its code together or not at all: a failure part way + /// through storing it must not leave a code on the user's screen that resolves to nothing for the + /// whole ten minutes the poll keeps succeeding. /// /// A representing the asynchronous operation. [Fact] - public async Task PendingRequest_ThatIsMalformedInRedis_SurfacesInsteadOfDegrading() + public async Task PendingRequest_WhenRedisGoesAwayMidWrite_IsResolvableByBothKeysOrNeither() { var instance = await CreateInstanceAsync(); var request = NewRequest(); - instance.Proxy.Cut(); - await instance.Store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), CancellationToken); - await RestoreAsync(instance); + // Redis is taken away the instant after it has applied the write of the secret key, which is + // where a two round trip write loses the code key. + instance.Proxy.CutAfterForwarding("request:" + request.Secret); + await Record.ExceptionAsync(() => instance.Store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), CancellationToken)); + + // Read straight from the server: the instance's own connection is the one that was cut. + var direct = await _redis.ConnectAsync(); + _connections.Add(direct); + var bySecret = await direct.GetDatabase().KeyExistsAsync("jellyfin:quickconnect:request:" + request.Secret); + var byCode = await direct.GetDatabase().KeyExistsAsync("jellyfin:quickconnect:code:" + request.Code); + + Assert.Equal(bySecret, byCode); + } + + /// + /// A malformed stored value is a fault of its own, not Redis being unavailable, so it is not reported + /// as either a miss or an outage. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task PendingRequest_ThatIsMalformedInRedis_SurfacesAsItsOwnFault() + { + var instance = await CreateInstanceAsync(); + var request = NewRequest(); await instance.Connection.GetDatabase().StringSetAsync( "jellyfin:quickconnect:request:" + request.Secret, @@ -113,102 +140,24 @@ public sealed class RedisQuickConnectStoreDegradedTests : IAsyncLifetime } /// - /// 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 store whose Redis comes back answers from Redis again, with nothing carried over from the outage. /// /// A representing the asynchronous operation. [Fact] - public async Task Authorization_ThatFailedToStore_LeavesNothingOnTheInstance() + public async Task Store_AfterAnOutage_WorksAgain() { 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 AssertUnavailableAsync(() => instance.Store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), CancellationToken)); await RestoreAsync(instance); - Assert.Null(await instance.Store.TryConsumeAuthorizationAsync(request.Secret, CancellationToken)); - } + Assert.Null(await instance.Store.GetRequestBySecretAsync(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)); + Assert.Equal(request.Secret, (await instance.Store.GetRequestByCodeAsync(request.Code, CancellationToken))?.Secret); } /// @@ -258,6 +207,51 @@ public sealed class RedisQuickConnectStoreDegradedTests : IAsyncLifetime Assert.False(await instance.Store.TryClaimAuthorizationAsync(authorized.Secret, expiresUtc, CancellationToken)); } + /// + /// An authorization is read, not spent: the same secret exchanged again on the same instance returns + /// the same access token for as long as the authorization lives. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task Authorization_IsReadRepeatedly_WithoutBeingSpent() + { + var instance = await CreateInstanceAsync(); + var request = NewRequest(); + await instance.Store.SetAuthorizationAsync( + request.Secret, + new AuthenticationResult { AccessToken = "token-1" }, + DateTime.UtcNow.AddMinutes(10), + CancellationToken); + + Assert.Equal("token-1", (await instance.Store.GetAuthorizationAsync(request.Secret, CancellationToken))?.AccessToken); + Assert.Equal("token-1", (await instance.Store.GetAuthorizationAsync(request.Secret, CancellationToken))?.AccessToken); + } + + /// + /// A write whose expiry has already passed is ignored, by the shared store and the process-local one + /// alike: a deployment must not get a different answer out of the two. + /// + /// A representing the asynchronous operation. + [Fact] + public async Task ElapsedExpiry_IsIgnoredByBothStores() + { + var shared = (await CreateInstanceAsync()).Store; + var local = new InMemoryQuickConnectStore(); + + foreach (var store in new IQuickConnectStore[] { shared, local }) + { + var request = NewRequest(); + var elapsed = DateTime.UtcNow.AddSeconds(-1); + + await store.SetRequestAsync(request, elapsed, CancellationToken); + await store.SetAuthorizationAsync(request.Secret, new AuthenticationResult { AccessToken = "token-1" }, elapsed, CancellationToken); + + Assert.Null(await store.GetRequestBySecretAsync(request.Secret, CancellationToken)); + Assert.Null(await store.GetRequestByCodeAsync(request.Code, CancellationToken)); + Assert.Null(await store.GetAuthorizationAsync(request.Secret, CancellationToken)); + } + } + private static QuickConnectResult NewRequest() => new QuickConnectResult( Guid.NewGuid().ToString("N"), Guid.NewGuid().ToString("N").Substring(0, 6), @@ -267,12 +261,12 @@ public sealed class RedisQuickConnectStoreDegradedTests : IAsyncLifetime "Jellyfin Web", "1.0.0"); - private static async Task AssertTransportFailureAsync(Func operation) + private static async Task AssertUnavailableAsync(Func operation) { var exception = await Record.ExceptionAsync(operation); Assert.NotNull(exception); - Assert.True(exception is RedisException or TimeoutException, exception.ToString()); + Assert.IsType(exception); } private static async Task RestoreAsync(Instance instance)