jellyfin-ha-src/tests/Jellyfin.Server.Tests/QuickConnect/RedisQuickConnectStoreDegradedTests.cs

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);
}