unkin/jellyfin-ha-src
4
0
Fork 0
Code Issues 12 Pull Requests Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main' into benvin/session-directory

Browse Source 
# Conflicts:
#	.woodpecker/ci.yaml
#	Jellyfin.Server/CoreAppHost.cs
#	tests/Jellyfin.Server.Tests/HighAvailability/RedisTestServer.cs
This commit is contained in:
Unkin Agent
2026-09-26 20:43:52 +10:00
parent 577d24d190 2a09108d4f
commit af4e44173f
21 changed files with 1967 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Jellyfin.Server.Implementations.Tests/QuickConnect/QuickConnectManagerTests.cs
+69 -17
View File
@@ -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<IQuickConnectStore>(new InMemoryQuickConnectStore());
// User object contains circular references.
_fixture.Behaviors.OfType<ThrowingRecursionBehavior>().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<ArgumentException>(() => _quickConnectManager.TryConnect(
public async Task TryConnect_InvalidAuthorizationInfo_ThrowsArgumentException(string device, string deviceId, string client, string version)
=> await Assert.ThrowsAsync<ArgumentException>(() => _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<AuthenticationException>(() => _quickConnectManager.TryConnect(_quickConnectAuthInfo));
await Assert.ThrowsAsync<AuthenticationException>(() => _quickConnectManager.TryConnect(_quickConnectAuthInfo));
}
[Fact]
public void CheckRequestStatus_QuickConnectUnavailable_ThrowsAuthenticationException()
public async Task CheckRequestStatus_QuickConnectUnavailable_ThrowsAuthenticationException()
{
_config.QuickConnectAvailable = false;
Assert.Throws<AuthenticationException>(() => _quickConnectManager.CheckRequestStatus(string.Empty));
await Assert.ThrowsAsync<AuthenticationException>(() => _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<AuthenticationException>(() => _quickConnectManager.GetAuthorizedRequest(string.Empty));
await Assert.ThrowsAsync<AuthenticationException>(() => _quickConnectManager.GetAuthorizedRequest(string.Empty));
}
[Fact]
@@ -106,34 +109,83 @@ 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<ResourceNotFoundException>(() => _quickConnectManager.CheckRequestStatus("Unknown secret"));
await Assert.ThrowsAsync<ResourceNotFoundException>(() => _quickConnectManager.CheckRequestStatus("Unknown secret"));
}
[Fact]
public void GetAuthorizedRequest_UnknownSecret_ThrowsResourceNotFoundException()
public async Task GetAuthorizedRequest_UnknownSecret_ThrowsResourceNotFoundException()
{
_config.QuickConnectAvailable = true;
Assert.Throws<ResourceNotFoundException>(() => _quickConnectManager.GetAuthorizedRequest("Unknown secret"));
await Assert.ThrowsAsync<ResourceNotFoundException>(() => _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 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_ReturnsTheSameResult()
{
_config.QuickConnectAvailable = true;
var res = await _quickConnectManager.TryConnect(_quickConnectAuthInfo);
await _quickConnectManager.AuthorizeRequest(Guid.Empty, res.Code);
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<ConflictException>(() => _quickConnectManager.AuthorizeRequest(Guid.Empty, res.Code));
}
private async Task<bool> AuthorizeAsync(string code)
{
try
{
return await _quickConnectManager.AuthorizeRequest(Guid.Empty, code).ConfigureAwait(false);
}
catch (ConflictException)
{
return false;
}
}
}
}
tests/Jellyfin.Server.Tests/HighAvailability/JellyfinSectionConfigurationTests.cs
+1
View File
@@ -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.
/// </summary>
[Collection("JellyfinSectionConfiguration")]
public sealed class JellyfinSectionConfigurationTests : IDisposable
{
private const string RedisKey = "Jellyfin:TranscodeStore:RedisConnectionString";
tests/Jellyfin.Server.Tests/HighAvailability/RedisFaultProxy.cs
+223
View File
@@ -0,0 +1,223 @@
using System;
using System.Collections.Concurrent;
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;
namespace Jellyfin.Server.Tests.HighAvailability;
/// <summary>
/// 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.
/// </summary>
public sealed class RedisFaultProxy : IAsyncDisposable
{
private readonly ConcurrentDictionary<TcpClient, byte> _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 volatile byte[]? _cutAfterMarker;
private RedisFaultProxy(TcpListener listener, int port, string targetHost, int targetPort)
{
_listener = listener;
_port = port;
_targetHost = targetHost;
_targetPort = targetPort;
}
/// <summary>
/// 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.
/// </summary>
public string ConnectionString => string.Create(
CultureInfo.InvariantCulture,
$"127.0.0.1:{_port},abortConnect=false,connectTimeout=500,syncTimeout=2000,connectRetry=1");
/// <summary>
/// Starts a proxy in front of the server named by <paramref name="target"/>.
/// </summary>
/// <param name="target">The connection string of the server to forward to.</param>
/// <returns>The running proxy.</returns>
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;
}
/// <summary>
/// Takes Redis away from everything connected through the proxy.
/// </summary>
public void Cut()
{
_cut = true;
DropLiveConnections();
}
/// <summary>
/// Arms a cut for the moment after a command containing <paramref name="marker"/> 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.
/// </summary>
/// <param name="marker">Text that identifies the command to cut after.</param>
public void CutAfterForwarding(string marker) => _cutAfterMarker = Encoding.UTF8.GetBytes(marker);
/// <summary>
/// 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.
/// </summary>
public void Restore()
{
_cutAfterMarker = null;
_cut = false;
}
/// <inheritdoc/>
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;
// 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(
CopyFromClientAsync(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 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
{
await from.CopyToAsync(to, _cts.Token).ConfigureAwait(false);
}
catch (Exception exception) when (exception is IOException or SocketException or OperationCanceledException or ObjectDisposedException)
{
}
}
}
tests/Jellyfin.Server.Tests/HighAvailability/RedisTestServer.cs
+30 -5
View File
@@ -31,7 +31,7 @@ public sealed class RedisTestServer : IAsyncDisposable
public string ConnectionString { get; }
/// <summary>
/// Starts or attaches to a server and connects to it.
/// Starts or attaches to a server and waits until it accepts connections.
/// </summary>
/// <returns>The running server.</returns>
public static async Task<RedisTestServer> StartAsync()
@@ -39,19 +39,24 @@ public sealed class RedisTestServer : IAsyncDisposable
var provided = Environment.GetEnvironmentVariable(ConnectionStringVariable);
if (!string.IsNullOrWhiteSpace(provided))
{
return new RedisTestServer(null, provided);
var attached = new RedisTestServer(null, provided);
await attached.WaitUntilReadyAsync().ConfigureAwait(false);
return attached;
}
var container = new RedisBuilder("valkey/valkey:8-alpine").Build();
await container.StartAsync().ConfigureAwait(false);
return new RedisTestServer(container, container.GetConnectionString());
var started = new RedisTestServer(container, container.GetConnectionString());
await started.WaitUntilReadyAsync().ConfigureAwait(false);
return started;
}
/// <summary>
/// Opens a connection to the server.
/// 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.
/// </summary>
/// <returns>The connection.</returns>
/// <returns>A new multiplexer.</returns>
public async Task<IConnectionMultiplexer> ConnectAsync()
=> await ConnectionMultiplexer.ConnectAsync(ConnectionString).ConfigureAwait(false);
@@ -63,4 +68,24 @@ public sealed class RedisTestServer : IAsyncDisposable
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);
}
}
}
}
tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectReplicaTests.cs
+374
View File
@@ -0,0 +1,374 @@
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;
/// <summary>
/// Three independently constructed <see cref="QuickConnectManager"/> 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.
/// </summary>
[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<IConnectionMultiplexer> _connections = new();
private PostgreSqlTestServer _postgres = null!;
private RedisTestServer _redis = null!;
/// <inheritdoc/>
public async ValueTask InitializeAsync()
{
_postgres = await PostgreSqlTestServer.StartAsync().ConfigureAwait(false);
_redis = await RedisTestServer.StartAsync().ConfigureAwait(false);
}
/// <inheritdoc/>
public async ValueTask DisposeAsync()
{
foreach (var connection in _connections)
{
await connection.DisposeAsync().ConfigureAwait(false);
}
await _redis.DisposeAsync().ConfigureAwait(false);
await _postgres.DisposeAsync().ConfigureAwait(false);
}
/// <summary>
/// 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.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[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);
}
/// <summary>
/// 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.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[Fact]
public async Task Exchange_RepeatedOnTwoReplicas_ReturnsTheSameToken()
{
var cancellationToken = TestContext.Current.CancellationToken;
var connectionString = await _postgres.CreateDatabaseAsync("quickconnect_replica_reexchange", 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 < 10; attempt++)
{
var authorizationInfo = AuthorizationInfoFor(attempt);
var initiated = await replicaA.Manager.TryConnect(authorizationInfo);
await replicaB.Manager.AuthorizeRequest(user.Id, initiated.Code);
var exchanged = await Task.WhenAll(
Task.Run(() => ExchangeAsync(replicaA.Manager, initiated.Secret), cancellationToken),
Task.Run(() => ExchangeAsync(replicaC.Manager, initiated.Secret), cancellationToken));
Assert.All(exchanged, outcome => Assert.NotNull(outcome));
Assert.Equal(exchanged[0]!.AccessToken, exchanged[1]!.AccessToken);
// 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);
}
}
/// <summary>
/// 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.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[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);
}
}
/// <summary>
/// An expired request is rejected on a replica that never saw it created, rather than resolving to a
/// stale authorization.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[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<ResourceNotFoundException>(() => replicaB.Manager.CheckRequestStatus(initiated.Secret));
await Assert.ThrowsAsync<ResourceNotFoundException>(() => replicaB.Manager.AuthorizeRequest(user.Id, initiated.Code));
}
/// <summary>
/// An authorization that was never exchanged expires too, so a code authorized and then abandoned
/// cannot be redeemed later from another replica.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[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<ResourceNotFoundException>(() => 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<bool> AuthorizeAsync(IQuickConnect manager, Guid userId, string code)
{
try
{
return await manager.AuthorizeRequest(userId, code).ConfigureAwait(false);
}
catch (ConflictException)
{
return false;
}
}
private static async Task<AuthenticationResult?> ExchangeAsync(IQuickConnect manager, string secret)
{
try
{
return await manager.GetAuthorizedRequest(secret).ConfigureAwait(false);
}
catch (ResourceNotFoundException)
{
return null;
}
}
private static async Task<User> 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<JellyfinDbContext>();
var provider = new PostgreSqlDatabaseProvider(dataSource);
provider.Initialise(optionsBuilder, new DatabaseConfigurationOptions { DatabaseType = "PostgreSQL" });
return new JellyfinDbContext(
optionsBuilder.Options,
NullLogger<JellyfinDbContext>.Instance,
provider,
new NoLockBehavior(NullLogger<NoLockBehavior>.Instance));
}
private async Task<Replica> CreateReplicaAsync(NpgsqlDataSource dataSource, User user)
{
var connection = await _redis.ConnectAsync().ConfigureAwait(false);
_connections.Add(connection);
var userManager = new Mock<IUserManager>();
userManager.Setup(manager => manager.GetUserById(user.Id)).Returns(user);
var deviceManager = new DeviceManager(new DataSourceContextFactory(dataSource), userManager.Object);
var configManager = new Mock<IServerConfigurationManager>();
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<ISessionManager>();
sessionManager
.Setup(manager => manager.AuthenticateDirect(It.IsAny<AuthenticationRequest>()))
.Returns<AuthenticationRequest>(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<RedisQuickConnectStore>.Instance);
var manager = new QuickConnectManager(
configManager.Object,
NullLogger<QuickConnectManager>.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<JellyfinDbContext>
{
private readonly NpgsqlDataSource _dataSource;
public DataSourceContextFactory(NpgsqlDataSource dataSource)
{
_dataSource = dataSource;
}
public JellyfinDbContext CreateDbContext() => CreateContext(_dataSource);
}
}
tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectStatusCodeTests.cs
+261
View File
@@ -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;
/// <summary>
/// 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.
/// </summary>
[Trait("Category", "RequiresDocker")]
public sealed class QuickConnectStatusCodeTests : IAsyncLifetime
{
private readonly Mock<ISessionManager> _sessionManager = new();
private RedisTestServer _redis = null!;
private RedisFaultProxy _proxy = null!;
private IConnectionMultiplexer _connection = null!;
private QuickConnectManager _manager = null!;
private QuickConnectController _controller = null!;
/// <inheritdoc/>
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<IServerConfigurationManager>();
configManager.Setup(manager => manager.Configuration).Returns(new ServerConfiguration { QuickConnectAvailable = true });
_manager = new QuickConnectManager(
configManager.Object,
NullLogger<QuickConnectManager>.Instance,
_sessionManager.Object,
new RedisQuickConnectStore(_connection, NullLogger<RedisQuickConnectStore>.Instance));
_controller = new QuickConnectController(_manager, Mock.Of<IAuthorizationContext>());
}
/// <inheritdoc/>
public async ValueTask DisposeAsync()
{
await _connection.DisposeAsync().ConfigureAwait(false);
await _proxy.DisposeAsync().ConfigureAwait(false);
await _redis.DisposeAsync().ConfigureAwait(false);
}
/// <summary>
/// A secret valkey has never heard of is a 404, which is what ends a flow the user abandoned.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[Fact]
public async Task Poll_UnknownSecret_IsNotFound()
{
Assert.Equal(
StatusCodes.Status404NotFound,
await StatusCodeAsync(async () => StatusOf(await _controller.GetQuickConnectState(NewSecret()))));
}
/// <summary>
/// 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.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[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))));
}
/// <summary>
/// The exchange leg tells the two apart the same way.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[Fact]
public async Task Exchange_UnknownSecret_IsNotFound()
{
Assert.Equal(
StatusCodes.Status404NotFound,
await StatusCodeAsync(async () =>
{
await _manager.GetAuthorizedRequest(NewSecret()).ConfigureAwait(false);
return StatusCodes.Status200OK;
}));
}
/// <summary>
/// The exchange leg while valkey is unreachable.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[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;
}));
}
/// <summary>
/// The authorize leg while valkey is unreachable.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[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;
}));
}
/// <summary>
/// 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.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[Fact]
public async Task Authorize_AfterAMintThatFailed_IsConflictAndSaysToStartAgain()
{
var initiated = await _manager.TryConnect(AuthorizationInfo());
_sessionManager
.Setup(manager => manager.AuthenticateDirect(It.IsAny<AuthenticationRequest>()))
.ThrowsAsync(new InvalidOperationException("mint failed"));
await Assert.ThrowsAsync<InvalidOperationException>(() => _manager.AuthorizeRequest(Guid.NewGuid(), initiated.Code));
var retry = await Record.ExceptionAsync(() => _manager.AuthorizeRequest(Guid.NewGuid(), initiated.Code));
var conflict = Assert.IsType<ConflictException>(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;
}));
}
/// <summary>
/// A request that really was authorized still says so, so the accurate message above is not just a
/// blanket replacement for the old one.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[Fact]
public async Task Authorize_OfAnAuthorizedRequest_IsConflictAndSaysAlreadyAuthorized()
{
var initiated = await _manager.TryConnect(AuthorizationInfo());
var userId = Guid.NewGuid();
_sessionManager
.Setup(manager => manager.AuthenticateDirect(It.IsAny<AuthenticationRequest>()))
.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<ConflictException>(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<QuickConnectResult> result)
=> result.Result is IStatusCodeActionResult status
? status.StatusCode ?? StatusCodes.Status200OK
: StatusCodes.Status200OK;
private static async Task<int> StatusCodeAsync(Func<Task<int>> action)
{
var appPaths = new Mock<IServerApplicationPaths>();
appPaths.Setup(paths => paths.ProgramSystemPath).Returns("/program");
appPaths.Setup(paths => paths.ProgramDataPath).Returns("/data");
var configManager = new Mock<IServerConfigurationManager>();
configManager.Setup(manager => manager.ApplicationPaths).Returns(appPaths.Object);
var hostEnvironment = new Mock<IWebHostEnvironment>();
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<ExceptionMiddleware>.Instance,
configManager.Object,
hostEnvironment.Object);
await middleware.Invoke(context).ConfigureAwait(false);
return context.Response.StatusCode;
}
private async Task<string> InitiateAsync()
=> (await _manager.TryConnect(AuthorizationInfo()).ConfigureAwait(false)).Secret;
}
tests/Jellyfin.Server.Tests/QuickConnect/QuickConnectStoreWiringTests.cs
+147
View File
@@ -0,0 +1,147 @@
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;
/// <summary>
/// Drives the whole configuration path a deployment uses: a bare
/// <c>Jellyfin__TranscodeStore__RedisConnectionString</c> environment variable, the server's own
/// configuration builder, the store registration, and a quick connect flow against a real valkey.
/// </summary>
[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;
/// <inheritdoc/>
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);
}
/// <inheritdoc/>
public async ValueTask DisposeAsync()
{
Environment.SetEnvironmentVariable(RedisConnectionStringVariable, null);
if (_configDirectory.Length > 0)
{
Directory.Delete(_configDirectory, true);
}
await _redis.DisposeAsync().ConfigureAwait(false);
}
/// <summary>
/// The variable form deployments set selects the shared store, and that store really talks to valkey.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[Fact]
public async Task ManifestStyleEnvironmentVariable_SelectsTheSharedStore()
{
Environment.SetEnvironmentVariable(RedisConnectionStringVariable, _redis.ConnectionString);
await using var provider = BuildProvider();
var store = provider.GetRequiredService<IQuickConnectStore>();
Assert.IsType<RedisQuickConnectStore>(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<IConnectionMultiplexer>();
Assert.True(await redis.GetDatabase().KeyExistsAsync("jellyfin:quickconnect:request:" + request.Secret));
}
/// <summary>
/// Without the variable the deployment is single-instance and gets the process-local store, which
/// runs a whole flow on its own.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[Fact]
public async Task NoEnvironmentVariable_SelectsTheProcessLocalStore()
{
Environment.SetEnvironmentVariable(RedisConnectionStringVariable, null);
await using var provider = BuildProvider();
var store = provider.GetRequiredService<IQuickConnectStore>();
Assert.IsType<InMemoryQuickConnectStore>(store);
var cancellationToken = TestContext.Current.CancellationToken;
var request = NewRequest();
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),
cancellationToken);
Assert.Equal("token-1", (await store.GetAuthorizationAsync(request.Secret, cancellationToken))?.AccessToken);
Assert.Equal("token-1", (await store.GetAuthorizationAsync(request.Secret, cancellationToken))?.AccessToken);
}
/// <summary>
/// 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.
/// </summary>
[Fact]
public void UnreachableRedisAtStartup_FailsClosed()
{
Environment.SetEnvironmentVariable(RedisConnectionStringVariable, "127.0.0.1:1,connectTimeout=250,connectRetry=0");
using var provider = BuildProvider();
Assert.ThrowsAny<RedisConnectionException>(() => provider.GetRequiredService<IQuickConnectStore>());
}
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<IApplicationPaths>();
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();
}
}
tests/Jellyfin.Server.Tests/QuickConnect/RedisQuickConnectStoreDegradedTests.cs
+302
View File
@@ -0,0 +1,302 @@
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.Common.Extensions;
using MediaBrowser.Controller.Authentication;
using MediaBrowser.Controller.QuickConnect;
using MediaBrowser.Model.QuickConnect;
using Microsoft.Extensions.Logging.Abstractions;
using StackExchange.Redis;
using Xunit;
namespace Jellyfin.Server.Tests.QuickConnect;
/// <summary>
/// What a <see cref="RedisQuickConnectStore"/> 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.
/// </summary>
[Trait("Category", "RequiresDocker")]
public sealed class RedisQuickConnectStoreDegradedTests : IAsyncLifetime
{
private readonly List<RedisFaultProxy> _proxies = new();
private readonly List<IConnectionMultiplexer> _connections = new();
private RedisTestServer _redis = null!;
private static CancellationToken CancellationToken => TestContext.Current.CancellationToken;
/// <inheritdoc/>
public async ValueTask InitializeAsync()
{
_redis = await RedisTestServer.StartAsync().ConfigureAwait(false);
}
/// <inheritdoc/>
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);
}
/// <summary>
/// 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.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[Fact]
public async Task EveryLeg_WhileRedisIsUnreachable_ReportsUnavailable()
{
var instance = await CreateInstanceAsync();
var request = NewRequest();
var expiresUtc = DateTime.UtcNow.AddMinutes(10);
instance.Proxy.Cut();
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));
}
/// <summary>
/// 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.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[Fact]
public async Task EveryRead_AgainstAHealthyRedis_ReportsAMissAsAMiss()
{
var instance = await CreateInstanceAsync();
var request = NewRequest();
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));
}
/// <summary>
/// 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.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[Fact]
public async Task PendingRequest_WhenRedisGoesAwayMidWrite_IsResolvableByBothKeysOrNeither()
{
var instance = await CreateInstanceAsync();
var request = NewRequest();
// 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);
}
/// <summary>
/// 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.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[Fact]
public async Task PendingRequest_ThatIsMalformedInRedis_SurfacesAsItsOwnFault()
{
var instance = await CreateInstanceAsync();
var request = NewRequest();
await instance.Connection.GetDatabase().StringSetAsync(
"jellyfin:quickconnect:request:" + request.Secret,
"{ not json",
TimeSpan.FromMinutes(10));
await Assert.ThrowsAsync<JsonException>(() => instance.Store.GetRequestBySecretAsync(request.Secret, CancellationToken));
}
/// <summary>
/// A store whose Redis comes back answers from Redis again, with nothing carried over from the outage.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[Fact]
public async Task Store_AfterAnOutage_WorksAgain()
{
var instance = await CreateInstanceAsync();
var request = NewRequest();
instance.Proxy.Cut();
await AssertUnavailableAsync(() => instance.Store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), CancellationToken));
await RestoreAsync(instance);
Assert.Null(await instance.Store.GetRequestBySecretAsync(request.Secret, CancellationToken));
await instance.Store.SetRequestAsync(request, DateTime.UtcNow.AddMinutes(10), CancellationToken);
Assert.Equal(request.Secret, (await instance.Store.GetRequestByCodeAsync(request.Code, CancellationToken))?.Secret);
}
/// <summary>
/// Two instances racing to authorize one request: exactly one of them may go on to mint a token.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[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);
}
}
/// <summary>
/// A request that is unknown, already claimed or already authorized cannot be claimed.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[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));
}
/// <summary>
/// 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.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[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);
}
/// <summary>
/// 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.
/// </summary>
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
[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),
DateTime.UtcNow,
"device-1",
"Living Room TV",
"Jellyfin Web",
"1.0.0");
private static async Task AssertUnavailableAsync(Func<Task> operation)
{
var exception = await Record.ExceptionAsync(operation);
Assert.NotNull(exception);
Assert.IsType<ServiceUnavailableException>(exception);
}
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<Instance> 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<RedisQuickConnectStore>.Instance));
}
private sealed record Instance(RedisFaultProxy Proxy, IConnectionMultiplexer Connection, RedisQuickConnectStore Store);
}