Compare commits
15 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 6c54a02240 | |||
| ba4d487c65 | |||
| 6691b785c3 | |||
| 2adb13f50f | |||
| 1c98f4a074 | |||
| 393994a454 | |||
| ad50c4e433 | |||
| 483c739fb1 | |||
| d39ec60e2c | |||
| a9d6c749fb | |||
| 44b62dcc64 | |||
| a025655b4d | |||
| face8ac653 | |||
| b662ffa48f | |||
| 1965c68a76 |
+9
-4
@@ -45,9 +45,9 @@ steps:
|
|||||||
# testcontainers, and a postgres service container deadlocks the step because the backend mounts
|
# testcontainers, and a postgres service container deadlocks the step because the backend mounts
|
||||||
# the ReadWriteOnce workspace volume into service pods and schedules them on another node.
|
# the ReadWriteOnce workspace volume into service pods and schedules them on another node.
|
||||||
# Its data directory lives on the step's ephemeral storage, not on the workspace volume.
|
# Its data directory lives on the step's ephemeral storage, not on the workspace volume.
|
||||||
# Scoped to Jellyfin.Server.Tests: the three classes in Jellyfin.Database.Tests.PostgreSQL still
|
# Both projects attach to it through JELLYFIN_TEST_POSTGRES and give every test a database of
|
||||||
# start their own container, and PostgreSqlProviderTests already fails on main - on an EF 10
|
# its own, so nothing here depends on a docker daemon.
|
||||||
# scalar query and on its own data - which a third test in the class then inherits.
|
# Valkey runs in the step for the same reason, attached through JELLYFIN_TEST_REDIS.
|
||||||
- name: postgres-migration-chain
|
- name: postgres-migration-chain
|
||||||
image: mcr.microsoft.com/dotnet/sdk:10.0
|
image: mcr.microsoft.com/dotnet/sdk:10.0
|
||||||
depends_on:
|
depends_on:
|
||||||
@@ -56,15 +56,20 @@ steps:
|
|||||||
DOTNET_CLI_TELEMETRY_OPTOUT: "1"
|
DOTNET_CLI_TELEMETRY_OPTOUT: "1"
|
||||||
DOTNET_NOLOGO: "1"
|
DOTNET_NOLOGO: "1"
|
||||||
JELLYFIN_TEST_POSTGRES: "Host=127.0.0.1;Port=5432;Database=postgres;Username=postgres"
|
JELLYFIN_TEST_POSTGRES: "Host=127.0.0.1;Port=5432;Database=postgres;Username=postgres"
|
||||||
|
JELLYFIN_TEST_REDIS: "127.0.0.1:6379"
|
||||||
commands:
|
commands:
|
||||||
- apt-get -o Acquire::Retries=3 update || apt-get -o Acquire::Retries=3 update
|
- 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
|
- install -d -o postgres -g postgres /tmp/pgdata /tmp/pgrun
|
||||||
- PGBIN=$(ls -d /usr/lib/postgresql/*/bin | tail -1)
|
- 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/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"
|
- 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.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"
|
- dotnet test tests/Jellyfin.Server.Tests/Jellyfin.Server.Tests.csproj -c Release --no-build --verbosity minimal --filter "Category=RequiresDocker"
|
||||||
|
- dotnet test tests/Jellyfin.Database.Tests.PostgreSQL/Jellyfin.Database.Tests.PostgreSQL.csproj -c Release --no-build --verbosity minimal --filter "Category=RequiresDocker"
|
||||||
backend_options:
|
backend_options:
|
||||||
kubernetes:
|
kubernetes:
|
||||||
serviceAccountName: jellyfin-ha-src
|
serviceAccountName: jellyfin-ha-src
|
||||||
|
|||||||
@@ -87,6 +87,12 @@ namespace Emby.Server.Implementations.AppBase
|
|||||||
/// <value>The application paths.</value>
|
/// <value>The application paths.</value>
|
||||||
public IApplicationPaths CommonApplicationPaths { get; private set; }
|
public IApplicationPaths CommonApplicationPaths { get; private set; }
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Gets or sets the bus announcing configuration writes to the other instances sharing this
|
||||||
|
/// configuration directory. Defaults to a no-op, which is the single-instance behaviour.
|
||||||
|
/// </summary>
|
||||||
|
public IConfigurationInvalidationBus InvalidationBus { get; set; } = NullConfigurationInvalidationBus.Instance;
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Gets or sets the system configuration.
|
/// Gets or sets the system configuration.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
@@ -169,6 +175,8 @@ namespace Emby.Server.Implementations.AppBase
|
|||||||
}
|
}
|
||||||
|
|
||||||
OnConfigurationUpdated();
|
OnConfigurationUpdated();
|
||||||
|
|
||||||
|
InvalidationBus.PublishLocalWrite(ConfigurationInvalidationScope.SystemConfiguration, null);
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
@@ -350,6 +358,29 @@ namespace Emby.Server.Implementations.AppBase
|
|||||||
}
|
}
|
||||||
|
|
||||||
OnNamedConfigurationUpdated(key, configuration);
|
OnNamedConfigurationUpdated(key, configuration);
|
||||||
|
|
||||||
|
InvalidationBus.PublishLocalWrite(ConfigurationInvalidationScope.NamedConfiguration, key);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public void InvalidateCachedConfiguration(string? key)
|
||||||
|
{
|
||||||
|
if (string.IsNullOrEmpty(key))
|
||||||
|
{
|
||||||
|
lock (_configurationSyncLock)
|
||||||
|
{
|
||||||
|
_configuration = null;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Reloads the system configuration off the shared file as a side effect of re-deriving
|
||||||
|
// the cache path from it, then tells the in-process consumers to re-read it.
|
||||||
|
OnConfigurationUpdated();
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
_configurations.TryRemove(key, out _);
|
||||||
|
|
||||||
|
OnNamedConfigurationUpdated(key, GetConfiguration(key));
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
|
|||||||
@@ -706,6 +706,8 @@ namespace Emby.Server.Implementations
|
|||||||
BaseItem.UserDataManager = Resolve<IUserDataManager>();
|
BaseItem.UserDataManager = Resolve<IUserDataManager>();
|
||||||
CollectionFolder.XmlSerializer = _xmlSerializer;
|
CollectionFolder.XmlSerializer = _xmlSerializer;
|
||||||
CollectionFolder.ApplicationHost = this;
|
CollectionFolder.ApplicationHost = this;
|
||||||
|
CollectionFolder.InvalidationBus = Resolve<IConfigurationInvalidationBus>();
|
||||||
|
ConfigurationManager.InvalidationBus = CollectionFolder.InvalidationBus;
|
||||||
Folder.UserViewManager = Resolve<IUserViewManager>();
|
Folder.UserViewManager = Resolve<IUserViewManager>();
|
||||||
Folder.CollectionManager = Resolve<ICollectionManager>();
|
Folder.CollectionManager = Resolve<ICollectionManager>();
|
||||||
Folder.LimitedConcurrencyLibraryScheduler = Resolve<ILimitedConcurrencyLibraryScheduler>();
|
Folder.LimitedConcurrencyLibraryScheduler = Resolve<ILimitedConcurrencyLibraryScheduler>();
|
||||||
@@ -790,6 +792,43 @@ namespace Emby.Server.Implementations
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Works out what a configuration update means for the ports this process bound at startup.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="boundHttpPort">The HTTP port this process is bound to.</param>
|
||||||
|
/// <param name="boundHttpsPort">The HTTPS port this process is bound to.</param>
|
||||||
|
/// <param name="configuredHttpPort">The HTTP port the shared configuration now carries.</param>
|
||||||
|
/// <param name="configuredHttpsPort">The HTTPS port the shared configuration now carries.</param>
|
||||||
|
/// <param name="isPortAuthorized">Whether the shared configuration still marks the port as authorized.</param>
|
||||||
|
/// <param name="isApplyingRemoteInvalidation">Whether this update is another instance's write being applied.</param>
|
||||||
|
/// <returns>What the update requires of this instance.</returns>
|
||||||
|
internal static PortChangeOutcome EvaluatePortChange(
|
||||||
|
int boundHttpPort,
|
||||||
|
int boundHttpsPort,
|
||||||
|
int configuredHttpPort,
|
||||||
|
int configuredHttpsPort,
|
||||||
|
bool isPortAuthorized,
|
||||||
|
bool isApplyingRemoteInvalidation)
|
||||||
|
{
|
||||||
|
// Nothing is bound yet, so nothing has gone stale.
|
||||||
|
if (boundHttpPort == 0 || boundHttpsPort == 0)
|
||||||
|
{
|
||||||
|
return default;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (configuredHttpPort == boundHttpPort && configuredHttpsPort == boundHttpsPort)
|
||||||
|
{
|
||||||
|
return default;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Whoever wrote the change, this process is still listening on a port the configuration no
|
||||||
|
// longer names, so the pending restart is reported either way. The authorization flag belongs
|
||||||
|
// to the instance that made the change: it cleared the flag along with the port, and clearing
|
||||||
|
// it again here would write shared configuration on that instance's behalf and announce it a
|
||||||
|
// second time.
|
||||||
|
return new PortChangeOutcome(true, isPortAuthorized && !isApplyingRemoteInvalidation);
|
||||||
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Called when [configuration updated].
|
/// Called when [configuration updated].
|
||||||
/// </summary>
|
/// </summary>
|
||||||
@@ -797,26 +836,24 @@ namespace Emby.Server.Implementations
|
|||||||
/// <param name="e">The <see cref="EventArgs"/> instance containing the event data.</param>
|
/// <param name="e">The <see cref="EventArgs"/> instance containing the event data.</param>
|
||||||
private void OnConfigurationUpdated(object sender, EventArgs e)
|
private void OnConfigurationUpdated(object sender, EventArgs e)
|
||||||
{
|
{
|
||||||
var requiresRestart = false;
|
|
||||||
var networkConfiguration = ConfigurationManager.GetNetworkConfiguration();
|
var networkConfiguration = ConfigurationManager.GetNetworkConfiguration();
|
||||||
|
|
||||||
// Don't do anything if these haven't been set yet
|
var portChange = EvaluatePortChange(
|
||||||
if (HttpPort != 0 && HttpsPort != 0)
|
HttpPort,
|
||||||
{
|
HttpsPort,
|
||||||
// Need to restart if ports have changed
|
networkConfiguration.InternalHttpPort,
|
||||||
if (networkConfiguration.InternalHttpPort != HttpPort
|
networkConfiguration.InternalHttpsPort,
|
||||||
|| networkConfiguration.InternalHttpsPort != HttpsPort)
|
ConfigurationManager.Configuration.IsPortAuthorized,
|
||||||
{
|
ConfigurationInvalidationContext.IsApplyingRemoteInvalidation);
|
||||||
if (ConfigurationManager.Configuration.IsPortAuthorized)
|
|
||||||
{
|
|
||||||
ConfigurationManager.Configuration.IsPortAuthorized = false;
|
|
||||||
ConfigurationManager.SaveConfiguration();
|
|
||||||
|
|
||||||
requiresRestart = true;
|
if (portChange.ClearsPortAuthorization)
|
||||||
}
|
{
|
||||||
}
|
ConfigurationManager.Configuration.IsPortAuthorized = false;
|
||||||
|
ConfigurationManager.SaveConfiguration();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
var requiresRestart = portChange.RequiresRestart;
|
||||||
|
|
||||||
if (ValidateSslCertificate(networkConfiguration))
|
if (ValidateSslCertificate(networkConfiguration))
|
||||||
{
|
{
|
||||||
requiresRestart = true;
|
requiresRestart = true;
|
||||||
|
|||||||
@@ -0,0 +1,82 @@
|
|||||||
|
using System;
|
||||||
|
using System.Threading;
|
||||||
|
using System.Threading.Tasks;
|
||||||
|
using MediaBrowser.Common.Configuration;
|
||||||
|
using MediaBrowser.Controller.Entities;
|
||||||
|
using Microsoft.Extensions.Hosting;
|
||||||
|
using Microsoft.Extensions.Logging;
|
||||||
|
|
||||||
|
namespace Emby.Server.Implementations.Configuration
|
||||||
|
{
|
||||||
|
/// <summary>
|
||||||
|
/// Applies the configuration invalidations published by the other instances sharing this
|
||||||
|
/// configuration directory, dropping the local cache entry so the next read comes off the shared file.
|
||||||
|
/// </summary>
|
||||||
|
public sealed class ConfigurationInvalidationSubscriber : IHostedService
|
||||||
|
{
|
||||||
|
private readonly IConfigurationInvalidationBus _bus;
|
||||||
|
private readonly IConfigurationManager _configurationManager;
|
||||||
|
private readonly ILogger<ConfigurationInvalidationSubscriber> _logger;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Initializes a new instance of the <see cref="ConfigurationInvalidationSubscriber"/> class.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="bus">The invalidation bus.</param>
|
||||||
|
/// <param name="configurationManager">The configuration manager holding the cached configuration.</param>
|
||||||
|
/// <param name="logger">The logger.</param>
|
||||||
|
public ConfigurationInvalidationSubscriber(
|
||||||
|
IConfigurationInvalidationBus bus,
|
||||||
|
IConfigurationManager configurationManager,
|
||||||
|
ILogger<ConfigurationInvalidationSubscriber> logger)
|
||||||
|
{
|
||||||
|
_bus = bus;
|
||||||
|
_configurationManager = configurationManager;
|
||||||
|
_logger = logger;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public Task StartAsync(CancellationToken cancellationToken)
|
||||||
|
{
|
||||||
|
_bus.Subscribe(Apply);
|
||||||
|
return Task.CompletedTask;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask;
|
||||||
|
|
||||||
|
private void Apply(ConfigurationInvalidation invalidation)
|
||||||
|
{
|
||||||
|
try
|
||||||
|
{
|
||||||
|
// Applying re-raises the same update events a local save raises, so the in-process
|
||||||
|
// consumers re-read. The scope tells those consumers that the write was somebody else's,
|
||||||
|
// so the ones that answer an update by writing neither repeat it nor publish it back.
|
||||||
|
using var scope = ConfigurationInvalidationContext.BeginApply();
|
||||||
|
|
||||||
|
switch (invalidation.Scope)
|
||||||
|
{
|
||||||
|
case ConfigurationInvalidationScope.SystemConfiguration:
|
||||||
|
_configurationManager.InvalidateCachedConfiguration(null);
|
||||||
|
break;
|
||||||
|
case ConfigurationInvalidationScope.NamedConfiguration when !string.IsNullOrEmpty(invalidation.Target):
|
||||||
|
_configurationManager.InvalidateCachedConfiguration(invalidation.Target);
|
||||||
|
break;
|
||||||
|
case ConfigurationInvalidationScope.LibraryOptions when !string.IsNullOrEmpty(invalidation.Target):
|
||||||
|
CollectionFolder.InvalidateLibraryOptions(invalidation.Target);
|
||||||
|
break;
|
||||||
|
case ConfigurationInvalidationScope.AllLibraryOptions:
|
||||||
|
CollectionFolder.InvalidateAllLibraryOptions();
|
||||||
|
break;
|
||||||
|
default:
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
_logger.LogDebug("Applied {Scope} invalidation for {Target} from {OriginId}.", invalidation.Scope, invalidation.Target, invalidation.OriginId);
|
||||||
|
}
|
||||||
|
catch (Exception ex)
|
||||||
|
{
|
||||||
|
_logger.LogWarning(ex, "Failed to apply a {Scope} invalidation for {Target}.", invalidation.Scope, invalidation.Target);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,96 @@
|
|||||||
|
using System;
|
||||||
|
using System.Text.Json;
|
||||||
|
using Jellyfin.Extensions.Json;
|
||||||
|
using MediaBrowser.Common.Configuration;
|
||||||
|
using Microsoft.Extensions.Logging;
|
||||||
|
using StackExchange.Redis;
|
||||||
|
|
||||||
|
namespace Emby.Server.Implementations.Configuration
|
||||||
|
{
|
||||||
|
/// <summary>
|
||||||
|
/// A Redis pub/sub <see cref="IConfigurationInvalidationBus"/>. Notices are broadcast on one channel
|
||||||
|
/// and every instance but the publisher applies them.
|
||||||
|
/// </summary>
|
||||||
|
public sealed class RedisConfigurationInvalidationBus : IConfigurationInvalidationBus
|
||||||
|
{
|
||||||
|
private const string ChannelName = "jellyfin:configinvalidation";
|
||||||
|
|
||||||
|
private static readonly JsonSerializerOptions _jsonOptions = JsonDefaults.Options;
|
||||||
|
|
||||||
|
private readonly ISubscriber _subscriber;
|
||||||
|
private readonly ILogger<RedisConfigurationInvalidationBus> _logger;
|
||||||
|
private readonly string _originId;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Initializes a new instance of the <see cref="RedisConfigurationInvalidationBus"/> class.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="redis">The Redis connection multiplexer.</param>
|
||||||
|
/// <param name="logger">The logger.</param>
|
||||||
|
public RedisConfigurationInvalidationBus(IConnectionMultiplexer redis, ILogger<RedisConfigurationInvalidationBus> logger)
|
||||||
|
{
|
||||||
|
ArgumentNullException.ThrowIfNull(redis);
|
||||||
|
|
||||||
|
_subscriber = redis.GetSubscriber();
|
||||||
|
_logger = logger;
|
||||||
|
_originId = Environment.GetEnvironmentVariable("JELLYFIN_INSTANCE_ID") ?? Environment.MachineName;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public void Publish(ConfigurationInvalidationScope scope, string? target)
|
||||||
|
{
|
||||||
|
var invalidation = new ConfigurationInvalidation
|
||||||
|
{
|
||||||
|
Scope = scope,
|
||||||
|
Target = target,
|
||||||
|
OriginId = _originId
|
||||||
|
};
|
||||||
|
|
||||||
|
try
|
||||||
|
{
|
||||||
|
// Fire and forget: an admin saving configuration must never wait on, or fail because of,
|
||||||
|
// the bus. The write has already reached the shared directory by this point.
|
||||||
|
_subscriber.Publish(
|
||||||
|
RedisChannel.Literal(ChannelName),
|
||||||
|
JsonSerializer.Serialize(invalidation, _jsonOptions),
|
||||||
|
CommandFlags.FireAndForget);
|
||||||
|
}
|
||||||
|
catch (Exception ex)
|
||||||
|
{
|
||||||
|
_logger.LogWarning(ex, "Failed to publish {Scope} invalidation for {Target}; other instances keep their cached copy until they restart.", scope, target);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public void Subscribe(Action<ConfigurationInvalidation> handler)
|
||||||
|
{
|
||||||
|
ArgumentNullException.ThrowIfNull(handler);
|
||||||
|
|
||||||
|
try
|
||||||
|
{
|
||||||
|
_subscriber.Subscribe(RedisChannel.Literal(ChannelName), (_, value) => Dispatch(handler, value));
|
||||||
|
}
|
||||||
|
catch (Exception ex)
|
||||||
|
{
|
||||||
|
_logger.LogWarning(ex, "Failed to subscribe to configuration invalidations; this instance keeps its cached configuration until it restarts.");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private void Dispatch(Action<ConfigurationInvalidation> handler, RedisValue value)
|
||||||
|
{
|
||||||
|
try
|
||||||
|
{
|
||||||
|
var invalidation = JsonSerializer.Deserialize<ConfigurationInvalidation>(value.ToString(), _jsonOptions);
|
||||||
|
if (invalidation is null || string.Equals(invalidation.OriginId, _originId, StringComparison.Ordinal))
|
||||||
|
{
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
handler(invalidation);
|
||||||
|
}
|
||||||
|
catch (Exception ex)
|
||||||
|
{
|
||||||
|
_logger.LogWarning(ex, "Failed to apply a configuration invalidation.");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,14 @@
|
|||||||
|
namespace Emby.Server.Implementations
|
||||||
|
{
|
||||||
|
/// <summary>
|
||||||
|
/// What a configuration update carrying different ports requires of the instance reading it.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="RequiresRestart">
|
||||||
|
/// Whether this process is still bound to a port the shared configuration no longer names, and so has
|
||||||
|
/// to report a pending restart.
|
||||||
|
/// </param>
|
||||||
|
/// <param name="ClearsPortAuthorization">
|
||||||
|
/// Whether this instance is the one that has to clear the port authorization flag and save it.
|
||||||
|
/// </param>
|
||||||
|
internal readonly record struct PortChangeOutcome(bool RequiresRestart, bool ClearsPortAuthorization);
|
||||||
|
}
|
||||||
@@ -1,7 +1,5 @@
|
|||||||
using System;
|
using System;
|
||||||
using System.Collections.Concurrent;
|
|
||||||
using System.Globalization;
|
using System.Globalization;
|
||||||
using System.Linq;
|
|
||||||
using System.Security.Cryptography;
|
using System.Security.Cryptography;
|
||||||
using System.Threading.Tasks;
|
using System.Threading.Tasks;
|
||||||
using MediaBrowser.Common.Extensions;
|
using MediaBrowser.Common.Extensions;
|
||||||
@@ -30,12 +28,10 @@ namespace Emby.Server.Implementations.QuickConnect
|
|||||||
/// </summary>
|
/// </summary>
|
||||||
private const int Timeout = 10;
|
private const int Timeout = 10;
|
||||||
|
|
||||||
private readonly ConcurrentDictionary<string, QuickConnectResult> _currentRequests = new();
|
|
||||||
private readonly ConcurrentDictionary<string, (DateTime Timestamp, AuthenticationResult AuthenticationResult)> _authorizedSecrets = new();
|
|
||||||
|
|
||||||
private readonly IServerConfigurationManager _config;
|
private readonly IServerConfigurationManager _config;
|
||||||
private readonly ILogger<QuickConnectManager> _logger;
|
private readonly ILogger<QuickConnectManager> _logger;
|
||||||
private readonly ISessionManager _sessionManager;
|
private readonly ISessionManager _sessionManager;
|
||||||
|
private readonly IQuickConnectStore _store;
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Initializes a new instance of the <see cref="QuickConnectManager"/> class.
|
/// Initializes a new instance of the <see cref="QuickConnectManager"/> class.
|
||||||
@@ -44,14 +40,17 @@ namespace Emby.Server.Implementations.QuickConnect
|
|||||||
/// <param name="config">Configuration.</param>
|
/// <param name="config">Configuration.</param>
|
||||||
/// <param name="logger">Logger.</param>
|
/// <param name="logger">Logger.</param>
|
||||||
/// <param name="sessionManager">Session Manager.</param>
|
/// <param name="sessionManager">Session Manager.</param>
|
||||||
|
/// <param name="store">Quick connect store.</param>
|
||||||
public QuickConnectManager(
|
public QuickConnectManager(
|
||||||
IServerConfigurationManager config,
|
IServerConfigurationManager config,
|
||||||
ILogger<QuickConnectManager> logger,
|
ILogger<QuickConnectManager> logger,
|
||||||
ISessionManager sessionManager)
|
ISessionManager sessionManager,
|
||||||
|
IQuickConnectStore store)
|
||||||
{
|
{
|
||||||
_config = config;
|
_config = config;
|
||||||
_logger = logger;
|
_logger = logger;
|
||||||
_sessionManager = sessionManager;
|
_sessionManager = sessionManager;
|
||||||
|
_store = store;
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <inheritdoc />
|
/// <inheritdoc />
|
||||||
@@ -69,7 +68,7 @@ namespace Emby.Server.Implementations.QuickConnect
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// <inheritdoc/>
|
/// <inheritdoc/>
|
||||||
public QuickConnectResult TryConnect(AuthorizationInfo authorizationInfo)
|
public async Task<QuickConnectResult> TryConnect(AuthorizationInfo authorizationInfo)
|
||||||
{
|
{
|
||||||
ArgumentException.ThrowIfNullOrEmpty(authorizationInfo.DeviceId);
|
ArgumentException.ThrowIfNullOrEmpty(authorizationInfo.DeviceId);
|
||||||
ArgumentException.ThrowIfNullOrEmpty(authorizationInfo.Device);
|
ArgumentException.ThrowIfNullOrEmpty(authorizationInfo.Device);
|
||||||
@@ -77,7 +76,6 @@ namespace Emby.Server.Implementations.QuickConnect
|
|||||||
ArgumentException.ThrowIfNullOrEmpty(authorizationInfo.Version);
|
ArgumentException.ThrowIfNullOrEmpty(authorizationInfo.Version);
|
||||||
|
|
||||||
AssertActive();
|
AssertActive();
|
||||||
ExpireRequests();
|
|
||||||
|
|
||||||
var secret = GenerateSecureRandom();
|
var secret = GenerateSecureRandom();
|
||||||
var code = GenerateCode();
|
var code = GenerateCode();
|
||||||
@@ -90,19 +88,17 @@ namespace Emby.Server.Implementations.QuickConnect
|
|||||||
authorizationInfo.Client,
|
authorizationInfo.Client,
|
||||||
authorizationInfo.Version);
|
authorizationInfo.Version);
|
||||||
|
|
||||||
_currentRequests[code] = result;
|
await _store.SetRequestAsync(result, ExpiryOf(result)).ConfigureAwait(false);
|
||||||
return result;
|
return result;
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <inheritdoc/>
|
/// <inheritdoc/>
|
||||||
public QuickConnectResult CheckRequestStatus(string secret)
|
public async Task<QuickConnectResult> CheckRequestStatus(string secret)
|
||||||
{
|
{
|
||||||
AssertActive();
|
AssertActive();
|
||||||
ExpireRequests();
|
|
||||||
|
|
||||||
string code = _currentRequests.Where(x => x.Value.Secret == secret).Select(x => x.Value.Code).DefaultIfEmpty(string.Empty).First();
|
var result = await _store.GetRequestBySecretAsync(secret).ConfigureAwait(false);
|
||||||
|
if (result is null)
|
||||||
if (!_currentRequests.TryGetValue(code, out QuickConnectResult? result))
|
|
||||||
{
|
{
|
||||||
throw new ResourceNotFoundException("Unable to find request with provided secret");
|
throw new ResourceNotFoundException("Unable to find request with provided secret");
|
||||||
}
|
}
|
||||||
@@ -136,9 +132,9 @@ namespace Emby.Server.Implementations.QuickConnect
|
|||||||
public async Task<bool> AuthorizeRequest(Guid userId, string code)
|
public async Task<bool> AuthorizeRequest(Guid userId, string code)
|
||||||
{
|
{
|
||||||
AssertActive();
|
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");
|
throw new ResourceNotFoundException("Unable to find request");
|
||||||
}
|
}
|
||||||
@@ -151,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.
|
// 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));
|
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
|
var authenticationResult = await _sessionManager.AuthenticateDirect(new AuthenticationRequest
|
||||||
{
|
{
|
||||||
UserId = userId,
|
UserId = userId,
|
||||||
@@ -160,9 +162,10 @@ namespace Emby.Server.Implementations.QuickConnect
|
|||||||
AppVersion = result.AppVersion
|
AppVersion = result.AppVersion
|
||||||
}).ConfigureAwait(false);
|
}).ConfigureAwait(false);
|
||||||
|
|
||||||
_authorizedSecrets[result.Secret] = (DateTime.UtcNow, authenticationResult);
|
|
||||||
result.Authenticated = true;
|
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);
|
_logger.LogDebug("Authorizing device with code {Code} to login as user {UserId}", code, userId);
|
||||||
|
|
||||||
@@ -170,19 +173,21 @@ namespace Emby.Server.Implementations.QuickConnect
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// <inheritdoc/>
|
/// <inheritdoc/>
|
||||||
public AuthenticationResult GetAuthorizedRequest(string secret)
|
public async Task<AuthenticationResult> GetAuthorizedRequest(string secret)
|
||||||
{
|
{
|
||||||
AssertActive();
|
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");
|
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)
|
private string GenerateSecureRandom(int length = 32)
|
||||||
{
|
{
|
||||||
Span<byte> bytes = stackalloc byte[length];
|
Span<byte> bytes = stackalloc byte[length];
|
||||||
@@ -190,42 +195,5 @@ namespace Emby.Server.Implementations.QuickConnect
|
|||||||
|
|
||||||
return Convert.ToHexString(bytes);
|
return Convert.ToHexString(bytes);
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>
|
|
||||||
/// Expire quick connect requests that are over the time limit. If <paramref name="expireAll"/> is true, all requests are unconditionally expired.
|
|
||||||
/// </summary>
|
|
||||||
/// <param name="expireAll">If true, all requests will be expired.</param>
|
|
||||||
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);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,175 @@
|
|||||||
|
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;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// A Redis-backed <see cref="IQuickConnectStore"/> 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 <c>GETDEL</c>, so only one instance can ever mint
|
||||||
|
/// a given secret's access token and only one can ever hand it out.
|
||||||
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// 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.
|
||||||
|
/// </remarks>
|
||||||
|
public sealed class RedisQuickConnectStore : IQuickConnectStore
|
||||||
|
{
|
||||||
|
private const string KeyPrefix = "jellyfin:quickconnect:";
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// 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 <c>SET NX</c>, so of two
|
||||||
|
/// instances racing on one code exactly one goes on to mint an access token.
|
||||||
|
/// </summary>
|
||||||
|
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<RedisQuickConnectStore> _logger;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Initializes a new instance of the <see cref="RedisQuickConnectStore"/> class.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="redis">The Redis connection multiplexer.</param>
|
||||||
|
/// <param name="logger">The logger.</param>
|
||||||
|
public RedisQuickConnectStore(IConnectionMultiplexer redis, ILogger<RedisQuickConnectStore> logger)
|
||||||
|
{
|
||||||
|
ArgumentNullException.ThrowIfNull(redis);
|
||||||
|
|
||||||
|
_db = redis.GetDatabase();
|
||||||
|
_fallback = new InMemoryQuickConnectStore();
|
||||||
|
_logger = logger;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public async Task<QuickConnectResult?> 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);
|
||||||
|
}
|
||||||
|
|
||||||
|
// A miss is an answer rather than a transport failure, so the fallback is not consulted for it.
|
||||||
|
return raw.HasValue ? JsonSerializer.Deserialize<QuickConnectResult>(raw.ToString(), JsonDefaults.Options) : null;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public async Task<QuickConnectResult?> 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);
|
||||||
|
}
|
||||||
|
|
||||||
|
return secret.HasValue
|
||||||
|
? await GetRequestBySecretAsync(secret.ToString(), cancellationToken).ConfigureAwait(false)
|
||||||
|
: null;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
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) when (IsTransportFailure(ex))
|
||||||
|
{
|
||||||
|
LogDegraded(ex);
|
||||||
|
await _fallback.SetRequestAsync(request, expiresUtc, cancellationToken).ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public async Task<bool> 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;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public async Task SetAuthorizationAsync(string secret, AuthenticationResult authenticationResult, DateTime expiresUtc, CancellationToken cancellationToken = default)
|
||||||
|
{
|
||||||
|
var ttl = expiresUtc - DateTime.UtcNow;
|
||||||
|
if (ttl <= TimeSpan.Zero)
|
||||||
|
{
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
var json = JsonSerializer.Serialize(authenticationResult, JsonDefaults.Options);
|
||||||
|
await _db.StringSetAsync(AuthorizationKey(secret), json, ttl).ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public async Task<AuthenticationResult?> TryConsumeAuthorizationAsync(string secret, CancellationToken cancellationToken = default)
|
||||||
|
{
|
||||||
|
var raw = await _db.StringGetDeleteAsync(AuthorizationKey(secret)).ConfigureAwait(false);
|
||||||
|
|
||||||
|
return raw.HasValue
|
||||||
|
? JsonSerializer.Deserialize<AuthenticationResult>(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 request state could not be shared through Redis; falling back to this instance only.");
|
||||||
|
}
|
||||||
@@ -59,7 +59,7 @@ public class QuickConnectController : BaseJellyfinApiController
|
|||||||
try
|
try
|
||||||
{
|
{
|
||||||
var auth = await _authContext.GetAuthorizationInfo(Request).ConfigureAwait(false);
|
var auth = await _authContext.GetAuthorizationInfo(Request).ConfigureAwait(false);
|
||||||
return _quickConnect.TryConnect(auth);
|
return await _quickConnect.TryConnect(auth).ConfigureAwait(false);
|
||||||
}
|
}
|
||||||
catch (AuthenticationException)
|
catch (AuthenticationException)
|
||||||
{
|
{
|
||||||
@@ -77,11 +77,11 @@ public class QuickConnectController : BaseJellyfinApiController
|
|||||||
[HttpGet("Connect")]
|
[HttpGet("Connect")]
|
||||||
[ProducesResponseType(StatusCodes.Status200OK)]
|
[ProducesResponseType(StatusCodes.Status200OK)]
|
||||||
[ProducesResponseType(StatusCodes.Status404NotFound)]
|
[ProducesResponseType(StatusCodes.Status404NotFound)]
|
||||||
public ActionResult<QuickConnectResult> GetQuickConnectState([FromQuery, Required] string secret)
|
public async Task<ActionResult<QuickConnectResult>> GetQuickConnectState([FromQuery, Required] string secret)
|
||||||
{
|
{
|
||||||
try
|
try
|
||||||
{
|
{
|
||||||
return _quickConnect.CheckRequestStatus(secret);
|
return await _quickConnect.CheckRequestStatus(secret).ConfigureAwait(false);
|
||||||
}
|
}
|
||||||
catch (ResourceNotFoundException)
|
catch (ResourceNotFoundException)
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -108,7 +108,7 @@ public class TvShowsController : BaseJellyfinApiController
|
|||||||
StartIndex = startIndex,
|
StartIndex = startIndex,
|
||||||
User = user,
|
User = user,
|
||||||
EnableTotalRecordCount = enableTotalRecordCount,
|
EnableTotalRecordCount = enableTotalRecordCount,
|
||||||
NextUpDateCutoff = nextUpDateCutoff ?? DateTime.MinValue,
|
NextUpDateCutoff = nextUpDateCutoff?.ToUniversalTime() ?? DateTime.SpecifyKind(DateTime.MinValue, DateTimeKind.Utc),
|
||||||
EnableResumable = enableResumable,
|
EnableResumable = enableResumable,
|
||||||
EnableRewatching = enableRewatching
|
EnableRewatching = enableRewatching
|
||||||
},
|
},
|
||||||
|
|||||||
@@ -245,11 +245,11 @@ public class UserController : BaseJellyfinApiController
|
|||||||
[HttpPost("AuthenticateWithQuickConnect")]
|
[HttpPost("AuthenticateWithQuickConnect")]
|
||||||
[ProducesResponseType(StatusCodes.Status200OK)]
|
[ProducesResponseType(StatusCodes.Status200OK)]
|
||||||
[Tags("Authentication")]
|
[Tags("Authentication")]
|
||||||
public ActionResult<AuthenticationResult> AuthenticateWithQuickConnect([FromBody, Required] QuickConnectDto request)
|
public async Task<ActionResult<AuthenticationResult>> AuthenticateWithQuickConnect([FromBody, Required] QuickConnectDto request)
|
||||||
{
|
{
|
||||||
try
|
try
|
||||||
{
|
{
|
||||||
return _quickConnectManager.GetAuthorizedRequest(request.Secret);
|
return await _quickConnectManager.GetAuthorizedRequest(request.Secret).ConfigureAwait(false);
|
||||||
}
|
}
|
||||||
catch (SecurityException e)
|
catch (SecurityException e)
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -112,6 +112,14 @@ namespace Jellyfin.Server
|
|||||||
// instance. Active by default once a Redis connection is configured, no-op otherwise.
|
// instance. Active by default once a Redis connection is configured, no-op otherwise.
|
||||||
serviceCollection.AddScanLeaderLease(_startupConfig, Logger);
|
serviceCollection.AddScanLeaderLease(_startupConfig, Logger);
|
||||||
|
|
||||||
|
// Configuration invalidation bus: propagates shared-configuration and library-option writes
|
||||||
|
// 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<ILyricProvider>())
|
foreach (var type in GetExportTypes<ILyricProvider>())
|
||||||
{
|
{
|
||||||
serviceCollection.AddSingleton(typeof(ILyricProvider), type);
|
serviceCollection.AddSingleton(typeof(ILyricProvider), type);
|
||||||
|
|||||||
@@ -0,0 +1,74 @@
|
|||||||
|
using System;
|
||||||
|
using Emby.Server.Implementations.Configuration;
|
||||||
|
using MediaBrowser.Common.Configuration;
|
||||||
|
using MediaBrowser.Controller.MediaEncoding;
|
||||||
|
using Microsoft.Extensions.Configuration;
|
||||||
|
using Microsoft.Extensions.DependencyInjection;
|
||||||
|
using Microsoft.Extensions.Logging;
|
||||||
|
using StackExchange.Redis;
|
||||||
|
|
||||||
|
namespace Jellyfin.Server.Extensions;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Extensions for registering the shared-configuration invalidation bus.
|
||||||
|
/// </summary>
|
||||||
|
public static class ConfigurationInvalidationServiceCollectionExtensions
|
||||||
|
{
|
||||||
|
/// <summary>
|
||||||
|
/// Registers the invalidation bus, Redis-backed when a connection string is configured and no-op
|
||||||
|
/// otherwise, and the subscriber applying the notices other instances publish.
|
||||||
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// The connection string is only set for a multi-instance deployment, which is the only shape where
|
||||||
|
/// one instance can write the shared configuration directory behind another's back.
|
||||||
|
/// </remarks>
|
||||||
|
/// <param name="serviceCollection">The service collection.</param>
|
||||||
|
/// <param name="configuration">The configuration to read the Redis connection string from.</param>
|
||||||
|
/// <param name="logger">The logger to report the selected bus on.</param>
|
||||||
|
/// <returns>The updated service collection.</returns>
|
||||||
|
public static IServiceCollection AddConfigurationInvalidationBus(
|
||||||
|
this IServiceCollection serviceCollection,
|
||||||
|
IConfiguration configuration,
|
||||||
|
ILogger logger)
|
||||||
|
{
|
||||||
|
ArgumentNullException.ThrowIfNull(configuration);
|
||||||
|
ArgumentNullException.ThrowIfNull(logger);
|
||||||
|
|
||||||
|
if (string.IsNullOrEmpty(configuration[TranscodeStoreOptions.RedisConnectionStringKey]))
|
||||||
|
{
|
||||||
|
logger.LogInformation(
|
||||||
|
"Configuration invalidation bus: {Bus}. Shared-configuration and library-visibility changes stay local to the instance that made them; set {Key} to propagate them.",
|
||||||
|
nameof(NullConfigurationInvalidationBus),
|
||||||
|
TranscodeStoreOptions.RedisConnectionStringKey);
|
||||||
|
|
||||||
|
return serviceCollection.AddSingleton<IConfigurationInvalidationBus>(NullConfigurationInvalidationBus.Instance);
|
||||||
|
}
|
||||||
|
|
||||||
|
logger.LogInformation(
|
||||||
|
"Configuration invalidation bus: {Bus}. Shared-configuration and library-visibility changes propagate to every instance.",
|
||||||
|
nameof(RedisConfigurationInvalidationBus));
|
||||||
|
|
||||||
|
serviceCollection.AddSingleton<IConfigurationInvalidationBus>(sp =>
|
||||||
|
{
|
||||||
|
try
|
||||||
|
{
|
||||||
|
return new RedisConfigurationInvalidationBus(
|
||||||
|
sp.GetRequiredService<IConnectionMultiplexer>(),
|
||||||
|
sp.GetRequiredService<ILogger<RedisConfigurationInvalidationBus>>());
|
||||||
|
}
|
||||||
|
catch (Exception ex)
|
||||||
|
{
|
||||||
|
// Fail open: an unreachable Redis degrades to the single-instance behaviour of every
|
||||||
|
// instance keeping its own cached configuration, rather than aborting startup.
|
||||||
|
sp.GetRequiredService<ILogger<CoreAppHost>>().LogError(
|
||||||
|
ex,
|
||||||
|
"Redis is configured but unavailable, so shared-configuration changes will not propagate between instances. Check {Key}.",
|
||||||
|
TranscodeStoreOptions.RedisConnectionStringKey);
|
||||||
|
|
||||||
|
return NullConfigurationInvalidationBus.Instance;
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
return serviceCollection.AddHostedService<ConfigurationInvalidationSubscriber>();
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Extensions for registering the quick connect store.
|
||||||
|
/// </summary>
|
||||||
|
public static class QuickConnectStoreServiceCollectionExtensions
|
||||||
|
{
|
||||||
|
/// <summary>
|
||||||
|
/// Registers the quick connect store, Redis-backed when a connection string is configured and
|
||||||
|
/// process-local otherwise, and reports the selected store at <see cref="LogLevel.Information"/>.
|
||||||
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// 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.
|
||||||
|
/// </remarks>
|
||||||
|
/// <param name="serviceCollection">The service collection.</param>
|
||||||
|
/// <param name="configuration">The configuration to read the Redis connection string from.</param>
|
||||||
|
/// <param name="logger">The logger to report the selected store on.</param>
|
||||||
|
/// <returns>The updated service collection.</returns>
|
||||||
|
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<IQuickConnectStore, InMemoryQuickConnectStore>();
|
||||||
|
}
|
||||||
|
|
||||||
|
logger.LogInformation(
|
||||||
|
"Quick connect store: {Store}. Quick connect flows complete across any instance.",
|
||||||
|
nameof(RedisQuickConnectStore));
|
||||||
|
|
||||||
|
return serviceCollection.AddSingleton<IQuickConnectStore>(sp =>
|
||||||
|
{
|
||||||
|
try
|
||||||
|
{
|
||||||
|
return new RedisQuickConnectStore(
|
||||||
|
sp.GetRequiredService<IConnectionMultiplexer>(),
|
||||||
|
sp.GetRequiredService<ILogger<RedisQuickConnectStore>>());
|
||||||
|
}
|
||||||
|
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<ILogger<CoreAppHost>>().LogError(
|
||||||
|
ex,
|
||||||
|
"Redis is configured but unavailable, so quick connect flows will not complete across instances. Check {Key}.",
|
||||||
|
TranscodeStoreOptions.RedisConnectionStringKey);
|
||||||
|
|
||||||
|
return new InMemoryQuickConnectStore();
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,26 @@
|
|||||||
|
namespace MediaBrowser.Common.Configuration
|
||||||
|
{
|
||||||
|
/// <summary>
|
||||||
|
/// A notice that one instance has written shared configuration, so every other instance has to drop
|
||||||
|
/// its locally cached copy and read the shared file again.
|
||||||
|
/// </summary>
|
||||||
|
public sealed class ConfigurationInvalidation
|
||||||
|
{
|
||||||
|
/// <summary>
|
||||||
|
/// Gets or sets the cache this notice refers to.
|
||||||
|
/// </summary>
|
||||||
|
public ConfigurationInvalidationScope Scope { get; set; }
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Gets or sets what was invalidated within the scope: the configuration key for
|
||||||
|
/// <see cref="ConfigurationInvalidationScope.NamedConfiguration"/>, the library path for
|
||||||
|
/// <see cref="ConfigurationInvalidationScope.LibraryOptions"/>, and <c>null</c> otherwise.
|
||||||
|
/// </summary>
|
||||||
|
public string? Target { get; set; }
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Gets or sets the identity of the instance that published the notice, so it can ignore its own.
|
||||||
|
/// </summary>
|
||||||
|
public string? OriginId { get; set; }
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,30 @@
|
|||||||
|
namespace MediaBrowser.Common.Configuration
|
||||||
|
{
|
||||||
|
/// <summary>
|
||||||
|
/// Extensions for <see cref="IConfigurationInvalidationBus"/>.
|
||||||
|
/// </summary>
|
||||||
|
public static class ConfigurationInvalidationBusExtensions
|
||||||
|
{
|
||||||
|
/// <summary>
|
||||||
|
/// Announces a write this instance originated, and stays silent for a write induced by an
|
||||||
|
/// invalidation another instance published.
|
||||||
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// This is the backstop for the write path: a consumer reacting to an applied invalidation by
|
||||||
|
/// writing - including a plugin that knows nothing about the bus - cannot turn that write into a
|
||||||
|
/// notice of its own.
|
||||||
|
/// </remarks>
|
||||||
|
/// <param name="bus">The bus.</param>
|
||||||
|
/// <param name="scope">The cache that was written.</param>
|
||||||
|
/// <param name="target">The configuration key or library path that was written, if any.</param>
|
||||||
|
public static void PublishLocalWrite(this IConfigurationInvalidationBus bus, ConfigurationInvalidationScope scope, string? target)
|
||||||
|
{
|
||||||
|
if (ConfigurationInvalidationContext.IsApplyingRemoteInvalidation)
|
||||||
|
{
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
bus.Publish(scope, target);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,57 @@
|
|||||||
|
using System;
|
||||||
|
using System.Threading;
|
||||||
|
|
||||||
|
namespace MediaBrowser.Common.Configuration
|
||||||
|
{
|
||||||
|
/// <summary>
|
||||||
|
/// Marks the flow of control that is applying an invalidation published by another instance, so the
|
||||||
|
/// reactions to it can tell a remote write apart from a local one.
|
||||||
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// Applying an invalidation raises the same update events a local save raises, because the in-process
|
||||||
|
/// consumers have to re-read the configuration either way. Some of those consumers answer an update by
|
||||||
|
/// writing, and that write must neither repeat what the publishing instance already did nor fan back
|
||||||
|
/// out over the bus. The flag rides the execution context, so it reaches the queued and asynchronous
|
||||||
|
/// event handlers as well as the synchronous ones.
|
||||||
|
/// </remarks>
|
||||||
|
public static class ConfigurationInvalidationContext
|
||||||
|
{
|
||||||
|
private static readonly AsyncLocal<bool> _applyingRemoteInvalidation = new();
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Gets a value indicating whether the current flow of control is applying an invalidation
|
||||||
|
/// published by another instance rather than handling a local save.
|
||||||
|
/// </summary>
|
||||||
|
public static bool IsApplyingRemoteInvalidation => _applyingRemoteInvalidation.Value;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Marks the current flow of control as applying a remote invalidation until the returned scope is
|
||||||
|
/// disposed.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>The scope to dispose once the invalidation has been applied.</returns>
|
||||||
|
public static IDisposable BeginApply() => new ApplyScope();
|
||||||
|
|
||||||
|
private sealed class ApplyScope : IDisposable
|
||||||
|
{
|
||||||
|
private readonly bool _previous;
|
||||||
|
private bool _disposed;
|
||||||
|
|
||||||
|
public ApplyScope()
|
||||||
|
{
|
||||||
|
_previous = _applyingRemoteInvalidation.Value;
|
||||||
|
_applyingRemoteInvalidation.Value = true;
|
||||||
|
}
|
||||||
|
|
||||||
|
public void Dispose()
|
||||||
|
{
|
||||||
|
if (_disposed)
|
||||||
|
{
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
_disposed = true;
|
||||||
|
_applyingRemoteInvalidation.Value = _previous;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,29 @@
|
|||||||
|
namespace MediaBrowser.Common.Configuration
|
||||||
|
{
|
||||||
|
/// <summary>
|
||||||
|
/// Identifies which locally cached copy of the shared configuration a
|
||||||
|
/// <see cref="ConfigurationInvalidation"/> refers to.
|
||||||
|
/// </summary>
|
||||||
|
public enum ConfigurationInvalidationScope
|
||||||
|
{
|
||||||
|
/// <summary>
|
||||||
|
/// The system configuration cached by the configuration manager.
|
||||||
|
/// </summary>
|
||||||
|
SystemConfiguration = 0,
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// A single named configuration, identified by its key.
|
||||||
|
/// </summary>
|
||||||
|
NamedConfiguration = 1,
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// The library options of a single collection folder, identified by its path.
|
||||||
|
/// </summary>
|
||||||
|
LibraryOptions = 2,
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// The library options of every collection folder, for changes to the library structure itself.
|
||||||
|
/// </summary>
|
||||||
|
AllLibraryOptions = 3
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,28 @@
|
|||||||
|
using System;
|
||||||
|
|
||||||
|
namespace MediaBrowser.Common.Configuration
|
||||||
|
{
|
||||||
|
/// <summary>
|
||||||
|
/// Carries cache-invalidation notices between the instances that share one configuration directory.
|
||||||
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// The shared directory carries the content; this bus only carries the fact that it changed. Every
|
||||||
|
/// implementation is expected to fail open: a bus that cannot deliver must not throw into the write
|
||||||
|
/// path, leaving each instance on its own locally cached copy until it restarts.
|
||||||
|
/// </remarks>
|
||||||
|
public interface IConfigurationInvalidationBus
|
||||||
|
{
|
||||||
|
/// <summary>
|
||||||
|
/// Announces that this instance has written shared configuration.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="scope">The cache that was written.</param>
|
||||||
|
/// <param name="target">The configuration key or library path that was written, if any.</param>
|
||||||
|
void Publish(ConfigurationInvalidationScope scope, string? target);
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Registers the handler invoked for notices published by other instances.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="handler">The handler applying the invalidation locally.</param>
|
||||||
|
void Subscribe(Action<ConfigurationInvalidation> handler);
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -85,6 +85,20 @@ namespace MediaBrowser.Common.Configuration
|
|||||||
/// </summary>
|
/// </summary>
|
||||||
/// <param name="factories">The factories.</param>
|
/// <param name="factories">The factories.</param>
|
||||||
void AddParts(IEnumerable<IConfigurationFactory> factories);
|
void AddParts(IEnumerable<IConfigurationFactory> factories);
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Drops the locally cached copy of configuration another instance has written to the shared
|
||||||
|
/// configuration directory, so the next read reloads it, and raises the local update event.
|
||||||
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// An implementation predating the invalidation bus keeps the default, which reports that it
|
||||||
|
/// cannot drop its cache rather than quietly leaving it stale. The caller applying a remote
|
||||||
|
/// notice treats that as a failed apply and logs it.
|
||||||
|
/// </remarks>
|
||||||
|
/// <param name="key">The named configuration key, or <c>null</c> for the system configuration.</param>
|
||||||
|
/// <exception cref="NotSupportedException">The implementation cannot drop its cached configuration.</exception>
|
||||||
|
void InvalidateCachedConfiguration(string? key)
|
||||||
|
=> throw new NotSupportedException(GetType().Name + " cannot drop configuration cached from the shared configuration directory, so writes by other instances will not be picked up.");
|
||||||
}
|
}
|
||||||
|
|
||||||
public static class ConfigurationManagerExtensions
|
public static class ConfigurationManagerExtensions
|
||||||
|
|||||||
@@ -0,0 +1,27 @@
|
|||||||
|
using System;
|
||||||
|
|
||||||
|
namespace MediaBrowser.Common.Configuration
|
||||||
|
{
|
||||||
|
/// <summary>
|
||||||
|
/// A no-op <see cref="IConfigurationInvalidationBus"/> used by single-instance installs and whenever
|
||||||
|
/// the shared bus is unavailable. Every instance keeps its own cached configuration, which is the
|
||||||
|
/// behaviour of an install that has only one.
|
||||||
|
/// </summary>
|
||||||
|
public sealed class NullConfigurationInvalidationBus : IConfigurationInvalidationBus
|
||||||
|
{
|
||||||
|
/// <summary>
|
||||||
|
/// Gets the shared instance.
|
||||||
|
/// </summary>
|
||||||
|
public static NullConfigurationInvalidationBus Instance { get; } = new NullConfigurationInvalidationBus();
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public void Publish(ConfigurationInvalidationScope scope, string? target)
|
||||||
|
{
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public void Subscribe(Action<ConfigurationInvalidation> handler)
|
||||||
|
{
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -14,6 +14,7 @@ using System.Threading.Tasks;
|
|||||||
using Jellyfin.Data.Enums;
|
using Jellyfin.Data.Enums;
|
||||||
using Jellyfin.Database.Implementations.Entities;
|
using Jellyfin.Database.Implementations.Entities;
|
||||||
using Jellyfin.Extensions.Json;
|
using Jellyfin.Extensions.Json;
|
||||||
|
using MediaBrowser.Common.Configuration;
|
||||||
using MediaBrowser.Controller.IO;
|
using MediaBrowser.Controller.IO;
|
||||||
using MediaBrowser.Controller.Library;
|
using MediaBrowser.Controller.Library;
|
||||||
using MediaBrowser.Controller.Providers;
|
using MediaBrowser.Controller.Providers;
|
||||||
@@ -70,6 +71,12 @@ namespace MediaBrowser.Controller.Entities
|
|||||||
|
|
||||||
public static IServerApplicationHost ApplicationHost { get; set; }
|
public static IServerApplicationHost ApplicationHost { get; set; }
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Gets or sets the bus announcing library option writes to the other instances sharing this
|
||||||
|
/// configuration directory. Defaults to a no-op, which is the single-instance behaviour.
|
||||||
|
/// </summary>
|
||||||
|
public static IConfigurationInvalidationBus InvalidationBus { get; set; } = NullConfigurationInvalidationBus.Instance;
|
||||||
|
|
||||||
[JsonIgnore]
|
[JsonIgnore]
|
||||||
public override bool SupportsPlayedStatus => false;
|
public override bool SupportsPlayedStatus => false;
|
||||||
|
|
||||||
@@ -188,11 +195,36 @@ namespace MediaBrowser.Controller.Entities
|
|||||||
XmlSerializer.SerializeToFile(clone, GetLibraryOptionsPath(path));
|
XmlSerializer.SerializeToFile(clone, GetLibraryOptionsPath(path));
|
||||||
|
|
||||||
LibraryOptionsUpdated?.Invoke(null, new LibraryOptionsUpdatedEventArgs(path, options));
|
LibraryOptionsUpdated?.Invoke(null, new LibraryOptionsUpdatedEventArgs(path, options));
|
||||||
|
|
||||||
|
InvalidationBus.PublishLocalWrite(ConfigurationInvalidationScope.LibraryOptions, path);
|
||||||
}
|
}
|
||||||
|
|
||||||
public static void OnCollectionFolderChange()
|
/// <summary>
|
||||||
|
/// Drops the cached options of one library so the next read comes off <c>options.xml</c> again.
|
||||||
|
/// Applied on the instances that did not write, and so does not publish.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="path">The library path.</param>
|
||||||
|
public static void InvalidateLibraryOptions(string path)
|
||||||
|
{
|
||||||
|
_libraryOptions.TryRemove(path, out _);
|
||||||
|
|
||||||
|
LibraryOptionsUpdated?.Invoke(null, new LibraryOptionsUpdatedEventArgs(path, GetLibraryOptions(path)));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Drops every cached library option set. Applied on the instances that did not write, and so does
|
||||||
|
/// not publish.
|
||||||
|
/// </summary>
|
||||||
|
public static void InvalidateAllLibraryOptions()
|
||||||
=> _libraryOptions.Clear();
|
=> _libraryOptions.Clear();
|
||||||
|
|
||||||
|
public static void OnCollectionFolderChange()
|
||||||
|
{
|
||||||
|
InvalidateAllLibraryOptions();
|
||||||
|
|
||||||
|
InvalidationBus.PublishLocalWrite(ConfigurationInvalidationScope.AllLibraryOptions, null);
|
||||||
|
}
|
||||||
|
|
||||||
public override bool IsSaveLocalMetadataEnabled()
|
public override bool IsSaveLocalMetadataEnabled()
|
||||||
{
|
{
|
||||||
return true;
|
return true;
|
||||||
|
|||||||
@@ -21,14 +21,14 @@ namespace MediaBrowser.Controller.QuickConnect
|
|||||||
/// </summary>
|
/// </summary>
|
||||||
/// <param name="authorizationInfo">The initiator authorization info.</param>
|
/// <param name="authorizationInfo">The initiator authorization info.</param>
|
||||||
/// <returns>A quick connect result with tokens to proceed or throws an exception if not active.</returns>
|
/// <returns>A quick connect result with tokens to proceed or throws an exception if not active.</returns>
|
||||||
QuickConnectResult TryConnect(AuthorizationInfo authorizationInfo);
|
Task<QuickConnectResult> TryConnect(AuthorizationInfo authorizationInfo);
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Checks the status of an individual request.
|
/// Checks the status of an individual request.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
/// <param name="secret">Unique secret identifier of the request.</param>
|
/// <param name="secret">Unique secret identifier of the request.</param>
|
||||||
/// <returns>Quick connect result.</returns>
|
/// <returns>Quick connect result.</returns>
|
||||||
QuickConnectResult CheckRequestStatus(string secret);
|
Task<QuickConnectResult> CheckRequestStatus(string secret);
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Authorizes a quick connect request to connect as the calling user.
|
/// Authorizes a quick connect request to connect as the calling user.
|
||||||
@@ -43,6 +43,6 @@ namespace MediaBrowser.Controller.QuickConnect
|
|||||||
/// </summary>
|
/// </summary>
|
||||||
/// <param name="secret">The secret.</param>
|
/// <param name="secret">The secret.</param>
|
||||||
/// <returns>The authentication result.</returns>
|
/// <returns>The authentication result.</returns>
|
||||||
AuthenticationResult GetAuthorizedRequest(string secret);
|
Task<AuthenticationResult> GetAuthorizedRequest(string secret);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,70 @@
|
|||||||
|
using System;
|
||||||
|
using System.Threading;
|
||||||
|
using System.Threading.Tasks;
|
||||||
|
using MediaBrowser.Controller.Authentication;
|
||||||
|
using MediaBrowser.Model.QuickConnect;
|
||||||
|
|
||||||
|
namespace MediaBrowser.Controller.QuickConnect;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// 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.
|
||||||
|
/// </summary>
|
||||||
|
public interface IQuickConnectStore
|
||||||
|
{
|
||||||
|
/// <summary>
|
||||||
|
/// Looks up a pending request by the secret handed to the initiating client.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="secret">The request secret.</param>
|
||||||
|
/// <param name="cancellationToken">A cancellation token.</param>
|
||||||
|
/// <returns>The request, or <c>null</c> when it is unknown or has expired.</returns>
|
||||||
|
Task<QuickConnectResult?> GetRequestBySecretAsync(string secret, CancellationToken cancellationToken = default);
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Looks up a pending request by the code shown to the user.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="code">The user facing code.</param>
|
||||||
|
/// <param name="cancellationToken">A cancellation token.</param>
|
||||||
|
/// <returns>The request, or <c>null</c> when it is unknown or has expired.</returns>
|
||||||
|
Task<QuickConnectResult?> GetRequestByCodeAsync(string code, CancellationToken cancellationToken = default);
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Stores a new or updated request until <paramref name="expiresUtc"/>.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="request">The request to store.</param>
|
||||||
|
/// <param name="expiresUtc">The instant the request stops being resolvable.</param>
|
||||||
|
/// <param name="cancellationToken">A cancellation token.</param>
|
||||||
|
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||||
|
Task SetRequestAsync(QuickConnectResult request, DateTime expiresUtc, CancellationToken cancellationToken = default);
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Atomically claims the sole right to authorize the request behind <paramref name="secret"/>, 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.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="secret">The request secret.</param>
|
||||||
|
/// <param name="expiresUtc">The instant the claim lapses, after which the request can be authorized again.</param>
|
||||||
|
/// <param name="cancellationToken">A cancellation token.</param>
|
||||||
|
/// <returns><c>true</c> when this caller may go on to authorize the request; <c>false</c> when it is unknown, expired, already authorized or being authorized elsewhere.</returns>
|
||||||
|
Task<bool> TryClaimAuthorizationAsync(string secret, DateTime expiresUtc, CancellationToken cancellationToken = default);
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Stores the authentication minted for an authorized request until <paramref name="expiresUtc"/>.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="secret">The request secret the client exchanges.</param>
|
||||||
|
/// <param name="authenticationResult">The authentication to hand out.</param>
|
||||||
|
/// <param name="expiresUtc">The instant the authentication stops being exchangeable.</param>
|
||||||
|
/// <param name="cancellationToken">A cancellation token.</param>
|
||||||
|
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||||
|
Task SetAuthorizationAsync(string secret, AuthenticationResult authenticationResult, DateTime expiresUtc, CancellationToken cancellationToken = default);
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Atomically takes the authentication for <paramref name="secret"/> and removes it, so that two
|
||||||
|
/// instances racing on the same secret cannot both hand out an access token.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="secret">The request secret.</param>
|
||||||
|
/// <param name="cancellationToken">A cancellation token.</param>
|
||||||
|
/// <returns>The authentication, or <c>null</c> when the secret is unknown, expired or already exchanged.</returns>
|
||||||
|
Task<AuthenticationResult?> TryConsumeAuthorizationAsync(string secret, CancellationToken cancellationToken = default);
|
||||||
|
}
|
||||||
@@ -0,0 +1,110 @@
|
|||||||
|
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;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// A process-local <see cref="IQuickConnectStore"/>. 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.
|
||||||
|
/// </summary>
|
||||||
|
public sealed class InMemoryQuickConnectStore : IQuickConnectStore
|
||||||
|
{
|
||||||
|
private readonly ConcurrentDictionary<string, Entry<QuickConnectResult>> _requests = new(StringComparer.Ordinal);
|
||||||
|
private readonly ConcurrentDictionary<string, Entry<AuthenticationResult>> _authorizations = new(StringComparer.Ordinal);
|
||||||
|
private readonly ConcurrentDictionary<string, DateTime> _authorizationClaims = new(StringComparer.Ordinal);
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public Task<QuickConnectResult?> GetRequestBySecretAsync(string secret, CancellationToken cancellationToken = default)
|
||||||
|
{
|
||||||
|
Expire();
|
||||||
|
return Task.FromResult(_requests.TryGetValue(secret, out var entry) ? entry.Value : null);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public Task<QuickConnectResult?> 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)));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public Task SetRequestAsync(QuickConnectResult request, DateTime expiresUtc, CancellationToken cancellationToken = default)
|
||||||
|
{
|
||||||
|
ArgumentNullException.ThrowIfNull(request);
|
||||||
|
|
||||||
|
Expire();
|
||||||
|
_requests[request.Secret] = new Entry<QuickConnectResult>(expiresUtc, request);
|
||||||
|
return Task.CompletedTask;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public Task<bool> 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));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public Task SetAuthorizationAsync(string secret, AuthenticationResult authenticationResult, DateTime expiresUtc, CancellationToken cancellationToken = default)
|
||||||
|
{
|
||||||
|
Expire();
|
||||||
|
_authorizations[secret] = new Entry<AuthenticationResult>(expiresUtc, authenticationResult);
|
||||||
|
return Task.CompletedTask;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public Task<AuthenticationResult?> TryConsumeAuthorizationAsync(string secret, CancellationToken cancellationToken = default)
|
||||||
|
{
|
||||||
|
Expire();
|
||||||
|
if (!_authorizations.TryRemove(secret, out var entry) || entry.ExpiresUtc <= DateTime.UtcNow)
|
||||||
|
{
|
||||||
|
return Task.FromResult<AuthenticationResult?>(null);
|
||||||
|
}
|
||||||
|
|
||||||
|
return Task.FromResult<AuthenticationResult?>(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 _);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
foreach (var (secret, expiresUtc) in _authorizationClaims)
|
||||||
|
{
|
||||||
|
if (expiresUtc <= now)
|
||||||
|
{
|
||||||
|
_authorizationClaims.TryRemove(secret, out _);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private sealed record Entry<T>(DateTime ExpiresUtc, T Value);
|
||||||
|
}
|
||||||
@@ -43,6 +43,14 @@ public sealed class ScanLeaderOptions
|
|||||||
"TaskExtractMediaSegments",
|
"TaskExtractMediaSegments",
|
||||||
"KeyframeExtraction",
|
"KeyframeExtraction",
|
||||||
"CleanupUserDataTask",
|
"CleanupUserDataTask",
|
||||||
"OptimizeDatabaseTask"
|
"OptimizeDatabaseTask",
|
||||||
|
"DownloadLyrics",
|
||||||
|
"DownloadSubtitles",
|
||||||
|
"TmdbRefreshUpcomingEpisodes",
|
||||||
|
"RefreshTrickplayImages",
|
||||||
|
"MoveTrickplayImages",
|
||||||
|
"RefreshInternetChannels",
|
||||||
|
"RefreshGuide",
|
||||||
|
"PluginUpdates"
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -12,7 +12,7 @@ public class NextUpQuery
|
|||||||
{
|
{
|
||||||
EnableImageTypes = Array.Empty<ImageType>();
|
EnableImageTypes = Array.Empty<ImageType>();
|
||||||
EnableTotalRecordCount = true;
|
EnableTotalRecordCount = true;
|
||||||
NextUpDateCutoff = DateTime.MinValue;
|
NextUpDateCutoff = DateTime.SpecifyKind(DateTime.MinValue, DateTimeKind.Utc);
|
||||||
EnableResumable = false;
|
EnableResumable = false;
|
||||||
EnableRewatching = false;
|
EnableRewatching = false;
|
||||||
}
|
}
|
||||||
@@ -56,7 +56,7 @@ public class NextUpQuery
|
|||||||
public bool EnableTotalRecordCount { get; set; }
|
public bool EnableTotalRecordCount { get; set; }
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Gets or sets a value indicating the oldest date for a show to appear in Next Up.
|
/// Gets or sets a value indicating the oldest date, in UTC, for a show to appear in Next Up.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
public DateTime NextUpDateCutoff { get; set; }
|
public DateTime NextUpDateCutoff { get; set; }
|
||||||
|
|
||||||
|
|||||||
@@ -444,6 +444,13 @@ public sealed class RecordingsManager : IRecordingsManager, IDisposable
|
|||||||
|
|
||||||
private async void OnNamedConfigurationUpdated(object? sender, ConfigurationUpdateEventArgs e)
|
private async void OnNamedConfigurationUpdated(object? sender, ConfigurationUpdateEventArgs e)
|
||||||
{
|
{
|
||||||
|
// The instance that wrote the change creates the folders; racing it from here would write the
|
||||||
|
// same virtual folder a second time.
|
||||||
|
if (ConfigurationInvalidationContext.IsApplyingRemoteInvalidation)
|
||||||
|
{
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
if (string.Equals(e.Key, "livetv", StringComparison.OrdinalIgnoreCase))
|
if (string.Equals(e.Key, "livetv", StringComparison.OrdinalIgnoreCase))
|
||||||
{
|
{
|
||||||
await CreateRecordingFolders().ConfigureAwait(false);
|
await CreateRecordingFolders().ConfigureAwait(false);
|
||||||
|
|||||||
@@ -18,6 +18,11 @@
|
|||||||
<PackageReference Include="coverlet.collector" />
|
<PackageReference Include="coverlet.collector" />
|
||||||
</ItemGroup>
|
</ItemGroup>
|
||||||
|
|
||||||
|
<ItemGroup>
|
||||||
|
<!-- Linked, not project-referenced: Jellyfin.Server.Tests drags the whole server into this output. -->
|
||||||
|
<Compile Include="..\Jellyfin.Server.Tests\Migrations\PostgreSqlTestServer.cs" Link="Migrations\PostgreSqlTestServer.cs" />
|
||||||
|
</ItemGroup>
|
||||||
|
|
||||||
<ItemGroup>
|
<ItemGroup>
|
||||||
<ProjectReference Include="..\..\src\Jellyfin.Database\Jellyfin.Database.Providers.PostgreSQL\Jellyfin.Database.Providers.PostgreSQL.csproj" />
|
<ProjectReference Include="..\..\src\Jellyfin.Database\Jellyfin.Database.Providers.PostgreSQL\Jellyfin.Database.Providers.PostgreSQL.csproj" />
|
||||||
<ProjectReference Include="..\..\src\Jellyfin.Database\Jellyfin.Database.Implementations\Jellyfin.Database.Implementations.csproj" />
|
<ProjectReference Include="..\..\src\Jellyfin.Database\Jellyfin.Database.Implementations\Jellyfin.Database.Implementations.csproj" />
|
||||||
|
|||||||
@@ -1,71 +1,68 @@
|
|||||||
using System;
|
using System;
|
||||||
using System.Collections.Generic;
|
using System.Collections.Generic;
|
||||||
|
using System.Threading;
|
||||||
using System.Threading.Tasks;
|
using System.Threading.Tasks;
|
||||||
using DotNet.Testcontainers.Builders;
|
|
||||||
using Jellyfin.Database.Implementations;
|
using Jellyfin.Database.Implementations;
|
||||||
using Jellyfin.Database.Implementations.DbConfiguration;
|
using Jellyfin.Database.Implementations.DbConfiguration;
|
||||||
using Jellyfin.Database.Implementations.Entities;
|
using Jellyfin.Database.Implementations.Entities;
|
||||||
using Jellyfin.Database.Implementations.Locking;
|
using Jellyfin.Database.Implementations.Locking;
|
||||||
using Jellyfin.Database.Providers.PostgreSQL;
|
using Jellyfin.Database.Providers.PostgreSQL;
|
||||||
|
using Jellyfin.Server.Tests.Migrations;
|
||||||
using Microsoft.EntityFrameworkCore;
|
using Microsoft.EntityFrameworkCore;
|
||||||
using Microsoft.Extensions.Logging.Abstractions;
|
using Microsoft.Extensions.Logging.Abstractions;
|
||||||
using Npgsql;
|
using Npgsql;
|
||||||
using Testcontainers.PostgreSql;
|
|
||||||
using Xunit;
|
using Xunit;
|
||||||
|
|
||||||
namespace Jellyfin.Database.Tests.PostgreSQL;
|
namespace Jellyfin.Database.Tests.PostgreSQL;
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Integration tests that verify concurrent access patterns against a real PostgreSQL 16 container.
|
/// Integration tests that verify concurrent access patterns against a real PostgreSQL server.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
[Xunit.Trait("Category", "RequiresDocker")]
|
[Xunit.Trait("Category", "RequiresDocker")]
|
||||||
public sealed class PostgreSqlConcurrencyTests : IAsyncLifetime
|
public sealed class PostgreSqlConcurrencyTests : IAsyncLifetime
|
||||||
{
|
{
|
||||||
private readonly PostgreSqlContainer _container;
|
private static int _databaseSequence;
|
||||||
|
|
||||||
|
private PostgreSqlTestServer? _server;
|
||||||
private NpgsqlDataSource? _dataSource;
|
private NpgsqlDataSource? _dataSource;
|
||||||
private PostgreSqlDatabaseProvider? _provider;
|
private PostgreSqlDatabaseProvider? _provider;
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Initializes a new instance of the <see cref="PostgreSqlConcurrencyTests"/> class.
|
/// Attaches to the test server, hands this test a database of its own and applies migrations to it.
|
||||||
/// </summary>
|
|
||||||
public PostgreSqlConcurrencyTests()
|
|
||||||
{
|
|
||||||
_container = new PostgreSqlBuilder("postgres:16-alpine")
|
|
||||||
.WithWaitStrategy(Wait.ForUnixContainer().UntilCommandIsCompleted("pg_isready"))
|
|
||||||
.Build();
|
|
||||||
}
|
|
||||||
|
|
||||||
/// <summary>
|
|
||||||
/// Starts the PostgreSQL container and applies migrations before any tests in the class run.
|
|
||||||
/// </summary>
|
/// </summary>
|
||||||
/// <returns>A <see cref="ValueTask"/> representing the asynchronous operation.</returns>
|
/// <returns>A <see cref="ValueTask"/> representing the asynchronous operation.</returns>
|
||||||
public async ValueTask InitializeAsync()
|
public async ValueTask InitializeAsync()
|
||||||
{
|
{
|
||||||
await _container.StartAsync().ConfigureAwait(false);
|
_server = await PostgreSqlTestServer.StartAsync().ConfigureAwait(false);
|
||||||
|
|
||||||
_dataSource = new NpgsqlDataSourceBuilder(_container.GetConnectionString()).Build();
|
var databaseName = FormattableString.Invariant($"pg_concurrency_{Interlocked.Increment(ref _databaseSequence)}");
|
||||||
|
var connectionString = await _server.CreateDatabaseAsync(databaseName, TestContext.Current.CancellationToken).ConfigureAwait(false);
|
||||||
|
_dataSource = new NpgsqlDataSourceBuilder(connectionString).Build();
|
||||||
_provider = new PostgreSqlDatabaseProvider(_dataSource);
|
_provider = new PostgreSqlDatabaseProvider(_dataSource);
|
||||||
|
|
||||||
// Apply migrations once for the whole test class.
|
|
||||||
var context = CreateContext();
|
var context = CreateContext();
|
||||||
await using (context.ConfigureAwait(false))
|
await using (context.ConfigureAwait(false))
|
||||||
{
|
{
|
||||||
await context.Database.MigrateAsync().ConfigureAwait(false);
|
await context.Database.MigrateAsync(TestContext.Current.CancellationToken).ConfigureAwait(false);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Stops and removes the PostgreSQL container after all tests in the class have run.
|
/// Releases the data source and the test server.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
/// <returns>A <see cref="ValueTask"/> representing the asynchronous operation.</returns>
|
/// <returns>A <see cref="ValueTask"/> representing the asynchronous operation.</returns>
|
||||||
public async ValueTask DisposeAsync()
|
public async ValueTask DisposeAsync()
|
||||||
{
|
{
|
||||||
|
// InitializeAsync can fail before the data source exists; its error must not be masked by an NRE here.
|
||||||
if (_dataSource is not null)
|
if (_dataSource is not null)
|
||||||
{
|
{
|
||||||
await _dataSource.DisposeAsync().ConfigureAwait(false);
|
await _dataSource.DisposeAsync().ConfigureAwait(false);
|
||||||
}
|
}
|
||||||
|
|
||||||
await _container.DisposeAsync().ConfigureAwait(false);
|
if (_server is not null)
|
||||||
|
{
|
||||||
|
await _server.DisposeAsync().ConfigureAwait(false);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
|
|||||||
@@ -1,62 +1,68 @@
|
|||||||
|
using System;
|
||||||
|
using System.Threading;
|
||||||
using System.Threading.Tasks;
|
using System.Threading.Tasks;
|
||||||
using DotNet.Testcontainers.Builders;
|
|
||||||
using Jellyfin.Database.Implementations;
|
using Jellyfin.Database.Implementations;
|
||||||
using Jellyfin.Database.Implementations.DbConfiguration;
|
using Jellyfin.Database.Implementations.DbConfiguration;
|
||||||
using Jellyfin.Database.Implementations.Locking;
|
using Jellyfin.Database.Implementations.Locking;
|
||||||
using Jellyfin.Database.Providers.PostgreSQL;
|
using Jellyfin.Database.Providers.PostgreSQL;
|
||||||
|
using Jellyfin.Server.Tests.Migrations;
|
||||||
using Microsoft.EntityFrameworkCore;
|
using Microsoft.EntityFrameworkCore;
|
||||||
using Microsoft.Extensions.Logging.Abstractions;
|
using Microsoft.Extensions.Logging.Abstractions;
|
||||||
using Npgsql;
|
using Npgsql;
|
||||||
using Testcontainers.PostgreSql;
|
|
||||||
using Xunit;
|
using Xunit;
|
||||||
|
|
||||||
namespace Jellyfin.Database.Tests.PostgreSQL;
|
namespace Jellyfin.Database.Tests.PostgreSQL;
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Integration tests that validate PostgreSQL migrations against a real container.
|
/// Integration tests that validate PostgreSQL migrations against a real server.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
[Xunit.Trait("Category", "RequiresDocker")]
|
[Xunit.Trait("Category", "RequiresDocker")]
|
||||||
public sealed class PostgreSqlMigrationTests : IAsyncLifetime
|
public sealed class PostgreSqlMigrationTests : IAsyncLifetime
|
||||||
{
|
{
|
||||||
private readonly PostgreSqlContainer _container;
|
private static int _databaseSequence;
|
||||||
|
|
||||||
|
private PostgreSqlTestServer? _server;
|
||||||
|
private NpgsqlDataSource? _dataSource;
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Initializes a new instance of the <see cref="PostgreSqlMigrationTests"/> class.
|
/// Attaches to the test server and hands this test an empty database of its own.
|
||||||
/// </summary>
|
|
||||||
public PostgreSqlMigrationTests()
|
|
||||||
{
|
|
||||||
_container = new PostgreSqlBuilder("postgres:16-alpine")
|
|
||||||
.WithWaitStrategy(Wait.ForUnixContainer().UntilCommandIsCompleted("pg_isready"))
|
|
||||||
.Build();
|
|
||||||
}
|
|
||||||
|
|
||||||
/// <summary>
|
|
||||||
/// Starts the PostgreSQL container before any tests in the class run.
|
|
||||||
/// </summary>
|
/// </summary>
|
||||||
/// <returns>A <see cref="ValueTask"/> representing the asynchronous operation.</returns>
|
/// <returns>A <see cref="ValueTask"/> representing the asynchronous operation.</returns>
|
||||||
public async ValueTask InitializeAsync()
|
public async ValueTask InitializeAsync()
|
||||||
{
|
{
|
||||||
await _container.StartAsync().ConfigureAwait(false);
|
_server = await PostgreSqlTestServer.StartAsync().ConfigureAwait(false);
|
||||||
|
|
||||||
|
var databaseName = FormattableString.Invariant($"pg_migration_{Interlocked.Increment(ref _databaseSequence)}");
|
||||||
|
var connectionString = await _server.CreateDatabaseAsync(databaseName, TestContext.Current.CancellationToken).ConfigureAwait(false);
|
||||||
|
_dataSource = new NpgsqlDataSourceBuilder(connectionString).Build();
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Stops and removes the PostgreSQL container after all tests in the class have run.
|
/// Releases the data source and the test server.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
/// <returns>A <see cref="ValueTask"/> representing the asynchronous operation.</returns>
|
/// <returns>A <see cref="ValueTask"/> representing the asynchronous operation.</returns>
|
||||||
public async ValueTask DisposeAsync()
|
public async ValueTask DisposeAsync()
|
||||||
{
|
{
|
||||||
await _container.DisposeAsync().ConfigureAwait(false);
|
// InitializeAsync can fail before the data source exists; its error must not be masked by an NRE here.
|
||||||
|
if (_dataSource is not null)
|
||||||
|
{
|
||||||
|
await _dataSource.DisposeAsync().ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
|
||||||
|
if (_server is not null)
|
||||||
|
{
|
||||||
|
await _server.DisposeAsync().ConfigureAwait(false);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Verifies that the <c>InitialPostgreSql</c> migration applies cleanly to a fresh PostgreSQL 16 container.
|
/// Verifies that the <c>InitialPostgreSql</c> migration applies cleanly to a fresh database.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||||
[Fact]
|
[Fact]
|
||||||
public async Task MigrateAsync_AppliesInitialMigrationCleanly()
|
public async Task MigrateAsync_AppliesInitialMigrationCleanly()
|
||||||
{
|
{
|
||||||
await using var dataSource = new NpgsqlDataSourceBuilder(_container.GetConnectionString()).Build();
|
var context = CreateContext(_dataSource!);
|
||||||
var context = CreateContext(dataSource);
|
|
||||||
await using (context)
|
await using (context)
|
||||||
{
|
{
|
||||||
await context.Database.MigrateAsync(TestContext.Current.CancellationToken);
|
await context.Database.MigrateAsync(TestContext.Current.CancellationToken);
|
||||||
@@ -73,11 +79,7 @@ public sealed class PostgreSqlMigrationTests : IAsyncLifetime
|
|||||||
[Fact]
|
[Fact]
|
||||||
public void CheckForUnappliedMigrations_PostgreSql()
|
public void CheckForUnappliedMigrations_PostgreSql()
|
||||||
{
|
{
|
||||||
// Use a dummy connection string; HasPendingModelChanges() is a purely in-memory check
|
using var context = CreateContext(_dataSource!);
|
||||||
// that compares the current compiled model with the migration snapshots — no real DB needed.
|
|
||||||
const string dummyConnectionString = "Host=localhost;Database=jellyfin;Username=postgres;Password=postgres";
|
|
||||||
using var dataSource = new NpgsqlDataSourceBuilder(dummyConnectionString).Build();
|
|
||||||
using var context = CreateContext(dataSource);
|
|
||||||
|
|
||||||
Assert.False(
|
Assert.False(
|
||||||
context.Database.HasPendingModelChanges(),
|
context.Database.HasPendingModelChanges(),
|
||||||
|
|||||||
@@ -2,71 +2,67 @@ using System;
|
|||||||
using System.Linq;
|
using System.Linq;
|
||||||
using System.Threading;
|
using System.Threading;
|
||||||
using System.Threading.Tasks;
|
using System.Threading.Tasks;
|
||||||
using DotNet.Testcontainers.Builders;
|
|
||||||
using Jellyfin.Database.Implementations;
|
using Jellyfin.Database.Implementations;
|
||||||
using Jellyfin.Database.Implementations.DbConfiguration;
|
using Jellyfin.Database.Implementations.DbConfiguration;
|
||||||
using Jellyfin.Database.Implementations.Entities;
|
using Jellyfin.Database.Implementations.Entities;
|
||||||
using Jellyfin.Database.Implementations.Locking;
|
using Jellyfin.Database.Implementations.Locking;
|
||||||
using Jellyfin.Database.Providers.PostgreSQL;
|
using Jellyfin.Database.Providers.PostgreSQL;
|
||||||
|
using Jellyfin.Server.Tests.Migrations;
|
||||||
using Microsoft.EntityFrameworkCore;
|
using Microsoft.EntityFrameworkCore;
|
||||||
using Microsoft.Extensions.Logging.Abstractions;
|
using Microsoft.Extensions.Logging.Abstractions;
|
||||||
using Npgsql;
|
using Npgsql;
|
||||||
using Testcontainers.PostgreSql;
|
|
||||||
using Xunit;
|
using Xunit;
|
||||||
|
|
||||||
namespace Jellyfin.Database.Tests.PostgreSQL;
|
namespace Jellyfin.Database.Tests.PostgreSQL;
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Integration tests for CRUD operations, optimisation, and purge against a real PostgreSQL 16 container.
|
/// Integration tests for CRUD operations, optimisation, and purge against a real PostgreSQL server.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
[Xunit.Trait("Category", "RequiresDocker")]
|
[Xunit.Trait("Category", "RequiresDocker")]
|
||||||
public sealed class PostgreSqlProviderTests : IAsyncLifetime
|
public sealed class PostgreSqlProviderTests : IAsyncLifetime
|
||||||
{
|
{
|
||||||
private readonly PostgreSqlContainer _container;
|
private static int _databaseSequence;
|
||||||
|
|
||||||
|
private PostgreSqlTestServer? _server;
|
||||||
private NpgsqlDataSource? _dataSource;
|
private NpgsqlDataSource? _dataSource;
|
||||||
private PostgreSqlDatabaseProvider? _provider;
|
private PostgreSqlDatabaseProvider? _provider;
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Initializes a new instance of the <see cref="PostgreSqlProviderTests"/> class.
|
/// Attaches to the test server, hands this test a database of its own and applies migrations to it.
|
||||||
/// </summary>
|
|
||||||
public PostgreSqlProviderTests()
|
|
||||||
{
|
|
||||||
_container = new PostgreSqlBuilder("postgres:16-alpine")
|
|
||||||
.WithWaitStrategy(Wait.ForUnixContainer().UntilCommandIsCompleted("pg_isready"))
|
|
||||||
.Build();
|
|
||||||
}
|
|
||||||
|
|
||||||
/// <summary>
|
|
||||||
/// Starts the PostgreSQL container and applies migrations before any tests in the class run.
|
|
||||||
/// </summary>
|
/// </summary>
|
||||||
/// <returns>A <see cref="ValueTask"/> representing the asynchronous operation.</returns>
|
/// <returns>A <see cref="ValueTask"/> representing the asynchronous operation.</returns>
|
||||||
public async ValueTask InitializeAsync()
|
public async ValueTask InitializeAsync()
|
||||||
{
|
{
|
||||||
await _container.StartAsync().ConfigureAwait(false);
|
_server = await PostgreSqlTestServer.StartAsync().ConfigureAwait(false);
|
||||||
|
|
||||||
_dataSource = new NpgsqlDataSourceBuilder(_container.GetConnectionString()).Build();
|
var databaseName = FormattableString.Invariant($"pg_provider_{Interlocked.Increment(ref _databaseSequence)}");
|
||||||
|
var connectionString = await _server.CreateDatabaseAsync(databaseName, TestContext.Current.CancellationToken).ConfigureAwait(false);
|
||||||
|
_dataSource = new NpgsqlDataSourceBuilder(connectionString).Build();
|
||||||
_provider = new PostgreSqlDatabaseProvider(_dataSource);
|
_provider = new PostgreSqlDatabaseProvider(_dataSource);
|
||||||
|
|
||||||
// Apply migrations once for the whole test class.
|
|
||||||
var context = CreateContext();
|
var context = CreateContext();
|
||||||
await using (context.ConfigureAwait(false))
|
await using (context.ConfigureAwait(false))
|
||||||
{
|
{
|
||||||
await context.Database.MigrateAsync().ConfigureAwait(false);
|
await context.Database.MigrateAsync(TestContext.Current.CancellationToken).ConfigureAwait(false);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Stops and removes the PostgreSQL container after all tests in the class have run.
|
/// Releases the data source and the test server.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
/// <returns>A <see cref="ValueTask"/> representing the asynchronous operation.</returns>
|
/// <returns>A <see cref="ValueTask"/> representing the asynchronous operation.</returns>
|
||||||
public async ValueTask DisposeAsync()
|
public async ValueTask DisposeAsync()
|
||||||
{
|
{
|
||||||
|
// InitializeAsync can fail before the data source exists; its error must not be masked by an NRE here.
|
||||||
if (_dataSource is not null)
|
if (_dataSource is not null)
|
||||||
{
|
{
|
||||||
await _dataSource.DisposeAsync().ConfigureAwait(false);
|
await _dataSource.DisposeAsync().ConfigureAwait(false);
|
||||||
}
|
}
|
||||||
|
|
||||||
await _container.DisposeAsync().ConfigureAwait(false);
|
if (_server is not null)
|
||||||
|
{
|
||||||
|
await _server.DisposeAsync().ConfigureAwait(false);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
@@ -155,11 +151,15 @@ public sealed class PostgreSqlProviderTests : IAsyncLifetime
|
|||||||
var ctx = CreateContext();
|
var ctx = CreateContext();
|
||||||
await using (ctx)
|
await using (ctx)
|
||||||
{
|
{
|
||||||
var userId = Guid.NewGuid();
|
// DisplayPreferences.UserId is a foreign key onto Users, which PostgreSQL enforces and SQLite does not.
|
||||||
|
var user = new User("prefsuser", "Jellyfin.Server.Implementations.Users.DefaultAuthenticationProvider", "Jellyfin.Server.Implementations.Users.DefaultPasswordResetProvider");
|
||||||
|
ctx.Users.Add(user);
|
||||||
|
await ctx.SaveChangesAsync(TestContext.Current.CancellationToken);
|
||||||
|
|
||||||
var itemId = Guid.NewGuid();
|
var itemId = Guid.NewGuid();
|
||||||
|
|
||||||
// Create
|
// Create
|
||||||
var prefs = new DisplayPreferences(userId, itemId, "TestClient");
|
var prefs = new DisplayPreferences(user.Id, itemId, "TestClient");
|
||||||
ctx.DisplayPreferences.Add(prefs);
|
ctx.DisplayPreferences.Add(prefs);
|
||||||
await ctx.SaveChangesAsync(TestContext.Current.CancellationToken);
|
await ctx.SaveChangesAsync(TestContext.Current.CancellationToken);
|
||||||
|
|
||||||
@@ -278,7 +278,7 @@ public sealed class PostgreSqlProviderTests : IAsyncLifetime
|
|||||||
|
|
||||||
// session_replication_role should be reset to 'origin' (default)
|
// session_replication_role should be reset to 'origin' (default)
|
||||||
var role = await ctx.Database
|
var role = await ctx.Database
|
||||||
.SqlQueryRaw<string>("SELECT current_setting('session_replication_role')")
|
.SqlQueryRaw<string>("SELECT current_setting('session_replication_role') AS \"Value\"")
|
||||||
.FirstAsync(TestContext.Current.CancellationToken);
|
.FirstAsync(TestContext.Current.CancellationToken);
|
||||||
Assert.Equal("origin", role);
|
Assert.Equal("origin", role);
|
||||||
}
|
}
|
||||||
|
|||||||
+81
@@ -0,0 +1,81 @@
|
|||||||
|
using Emby.Server.Implementations;
|
||||||
|
using Xunit;
|
||||||
|
|
||||||
|
namespace Jellyfin.Server.Implementations.Tests.Configuration;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// The decision <c>ApplicationHost.OnConfigurationUpdated</c> makes about a port change. The ports this
|
||||||
|
/// process bound are fixed for its lifetime and the pending-restart flag is per-process, so an instance
|
||||||
|
/// applying another instance's port change still has to notice its own binding went stale - while leaving
|
||||||
|
/// the authorization write, and the notice that follows it, to the instance that made the change.
|
||||||
|
/// </summary>
|
||||||
|
public static class ApplicationHostPortChangeTests
|
||||||
|
{
|
||||||
|
/// <summary>
|
||||||
|
/// The local case, unchanged: clear the authorization flag and report the pending restart.
|
||||||
|
/// </summary>
|
||||||
|
[Fact]
|
||||||
|
public static void LocalPortChange_ClearsAuthorizationAndRequiresRestart()
|
||||||
|
{
|
||||||
|
var outcome = ApplicationHost.EvaluatePortChange(8096, 8920, 9096, 8920, true, false);
|
||||||
|
|
||||||
|
Assert.True(outcome.RequiresRestart);
|
||||||
|
Assert.True(outcome.ClearsPortAuthorization);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// The cross-instance case: the peer wrote the new port and cleared the flag with it, so this
|
||||||
|
/// instance must not write, but it is still listening on the old port and has to say so.
|
||||||
|
/// </summary>
|
||||||
|
[Theory]
|
||||||
|
[InlineData(true)]
|
||||||
|
[InlineData(false)]
|
||||||
|
public static void RemotePortChange_RequiresRestartWithoutWriting(bool isPortAuthorized)
|
||||||
|
{
|
||||||
|
var outcome = ApplicationHost.EvaluatePortChange(8096, 8920, 9096, 8920, isPortAuthorized, true);
|
||||||
|
|
||||||
|
Assert.True(outcome.RequiresRestart);
|
||||||
|
Assert.False(outcome.ClearsPortAuthorization);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// A second update while a port change is already pending must not write the flag again, and the
|
||||||
|
/// binding is still stale.
|
||||||
|
/// </summary>
|
||||||
|
[Fact]
|
||||||
|
public static void LocalPortChange_WithAuthorizationAlreadyCleared_RequiresRestartWithoutWriting()
|
||||||
|
{
|
||||||
|
var outcome = ApplicationHost.EvaluatePortChange(8096, 8920, 9096, 8920, false, false);
|
||||||
|
|
||||||
|
Assert.True(outcome.RequiresRestart);
|
||||||
|
Assert.False(outcome.ClearsPortAuthorization);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// An update that leaves the ports alone is not a port change, whoever wrote it.
|
||||||
|
/// </summary>
|
||||||
|
[Theory]
|
||||||
|
[InlineData(false)]
|
||||||
|
[InlineData(true)]
|
||||||
|
public static void UnchangedPorts_DoNothing(bool isApplyingRemoteInvalidation)
|
||||||
|
{
|
||||||
|
var outcome = ApplicationHost.EvaluatePortChange(8096, 8920, 8096, 8920, true, isApplyingRemoteInvalidation);
|
||||||
|
|
||||||
|
Assert.False(outcome.RequiresRestart);
|
||||||
|
Assert.False(outcome.ClearsPortAuthorization);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Nothing is decided before the ports have been bound.
|
||||||
|
/// </summary>
|
||||||
|
[Theory]
|
||||||
|
[InlineData(0, 8920)]
|
||||||
|
[InlineData(8096, 0)]
|
||||||
|
public static void UnboundPorts_DoNothing(int boundHttpPort, int boundHttpsPort)
|
||||||
|
{
|
||||||
|
var outcome = ApplicationHost.EvaluatePortChange(boundHttpPort, boundHttpsPort, 9096, 9920, true, false);
|
||||||
|
|
||||||
|
Assert.False(outcome.RequiresRestart);
|
||||||
|
Assert.False(outcome.ClearsPortAuthorization);
|
||||||
|
}
|
||||||
|
}
|
||||||
+73
@@ -0,0 +1,73 @@
|
|||||||
|
using System;
|
||||||
|
using System.Collections.Generic;
|
||||||
|
using System.Linq;
|
||||||
|
using MediaBrowser.Common.Configuration;
|
||||||
|
|
||||||
|
namespace Jellyfin.Server.Implementations.Tests.Configuration;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// An in-process stand-in for the Redis pub/sub bus: every endpoint connected to one fabric receives
|
||||||
|
/// what the others publish, and never its own notices.
|
||||||
|
/// </summary>
|
||||||
|
internal sealed class FakeInvalidationBusFabric
|
||||||
|
{
|
||||||
|
private readonly List<Endpoint> _endpoints = new();
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Connects a new instance to the fabric.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="originId">The identity of the connecting instance.</param>
|
||||||
|
/// <returns>The bus of that instance.</returns>
|
||||||
|
public IConfigurationInvalidationBus Connect(string originId)
|
||||||
|
{
|
||||||
|
var endpoint = new Endpoint(this, originId);
|
||||||
|
lock (_endpoints)
|
||||||
|
{
|
||||||
|
_endpoints.Add(endpoint);
|
||||||
|
}
|
||||||
|
|
||||||
|
return endpoint;
|
||||||
|
}
|
||||||
|
|
||||||
|
private void Broadcast(ConfigurationInvalidation invalidation)
|
||||||
|
{
|
||||||
|
Endpoint[] endpoints;
|
||||||
|
lock (_endpoints)
|
||||||
|
{
|
||||||
|
endpoints = _endpoints.ToArray();
|
||||||
|
}
|
||||||
|
|
||||||
|
foreach (var endpoint in endpoints.Where(e => !string.Equals(e.OriginId, invalidation.OriginId, StringComparison.Ordinal)))
|
||||||
|
{
|
||||||
|
endpoint.Deliver(invalidation);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private sealed class Endpoint : IConfigurationInvalidationBus
|
||||||
|
{
|
||||||
|
private readonly FakeInvalidationBusFabric _fabric;
|
||||||
|
private readonly List<Action<ConfigurationInvalidation>> _handlers = new();
|
||||||
|
|
||||||
|
public Endpoint(FakeInvalidationBusFabric fabric, string originId)
|
||||||
|
{
|
||||||
|
_fabric = fabric;
|
||||||
|
OriginId = originId;
|
||||||
|
}
|
||||||
|
|
||||||
|
public string OriginId { get; }
|
||||||
|
|
||||||
|
public void Publish(ConfigurationInvalidationScope scope, string? target)
|
||||||
|
=> _fabric.Broadcast(new ConfigurationInvalidation { Scope = scope, Target = target, OriginId = OriginId });
|
||||||
|
|
||||||
|
public void Subscribe(Action<ConfigurationInvalidation> handler)
|
||||||
|
=> _handlers.Add(handler);
|
||||||
|
|
||||||
|
public void Deliver(ConfigurationInvalidation invalidation)
|
||||||
|
{
|
||||||
|
foreach (var handler in _handlers)
|
||||||
|
{
|
||||||
|
handler(invalidation);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
+161
@@ -0,0 +1,161 @@
|
|||||||
|
using System;
|
||||||
|
using System.IO;
|
||||||
|
using System.Threading;
|
||||||
|
using System.Threading.Tasks;
|
||||||
|
using Emby.Server.Implementations.Configuration;
|
||||||
|
using Emby.Server.Implementations.Serialization;
|
||||||
|
using Jellyfin.Data;
|
||||||
|
using Jellyfin.Database.Implementations.Entities;
|
||||||
|
using Jellyfin.Database.Implementations.Enums;
|
||||||
|
using MediaBrowser.Common.Configuration;
|
||||||
|
using MediaBrowser.Controller;
|
||||||
|
using MediaBrowser.Controller.Entities;
|
||||||
|
using MediaBrowser.Model.Configuration;
|
||||||
|
using Microsoft.Extensions.Logging.Abstractions;
|
||||||
|
using Moq;
|
||||||
|
using Xunit;
|
||||||
|
|
||||||
|
namespace Jellyfin.Server.Implementations.Tests.Configuration;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Library options are cached in a process-wide dictionary, so the replica that did not serve the admin's
|
||||||
|
/// request is the one under test here: the other replica's write reaches the shared library directory, and
|
||||||
|
/// this one has to stop answering out of its own stale copy.
|
||||||
|
/// </summary>
|
||||||
|
/// <remarks>
|
||||||
|
/// Disabling a library is an access revocation that overrides every per-user check, so a stale replica
|
||||||
|
/// keeps serving content that is supposed to be hidden from everyone.
|
||||||
|
/// </remarks>
|
||||||
|
public sealed class LibraryVisibilityPropagationTests : IDisposable
|
||||||
|
{
|
||||||
|
private readonly string _libraryPath;
|
||||||
|
private readonly MyXmlSerializer _serializer = new MyXmlSerializer();
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Initializes a new instance of the <see cref="LibraryVisibilityPropagationTests"/> class.
|
||||||
|
/// </summary>
|
||||||
|
public LibraryVisibilityPropagationTests()
|
||||||
|
{
|
||||||
|
_libraryPath = Path.Combine(Path.GetTempPath(), "jf-library-prop-" + Guid.NewGuid().ToString("N"));
|
||||||
|
Directory.CreateDirectory(_libraryPath);
|
||||||
|
|
||||||
|
var applicationHost = new Mock<IServerApplicationHost>();
|
||||||
|
applicationHost.Setup(host => host.ExpandVirtualPath(It.IsAny<string>())).Returns<string>(path => path);
|
||||||
|
applicationHost.Setup(host => host.ReverseVirtualPath(It.IsAny<string>())).Returns<string>(path => path);
|
||||||
|
|
||||||
|
CollectionFolder.XmlSerializer = _serializer;
|
||||||
|
CollectionFolder.ApplicationHost = applicationHost.Object;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public void Dispose()
|
||||||
|
{
|
||||||
|
CollectionFolder.InvalidationBus = NullConfigurationInvalidationBus.Instance;
|
||||||
|
CollectionFolder.InvalidateAllLibraryOptions();
|
||||||
|
|
||||||
|
try
|
||||||
|
{
|
||||||
|
Directory.Delete(_libraryPath, true);
|
||||||
|
}
|
||||||
|
catch (IOException)
|
||||||
|
{
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Disabling a library on one replica has to hide it on every replica. Until it does, the ones that did
|
||||||
|
/// not serve the request keep the library visible to every user.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||||
|
[Fact]
|
||||||
|
public async Task LibraryDisabledOnAnotherInstance_IsNotVisibleHere()
|
||||||
|
{
|
||||||
|
var fabric = new FakeInvalidationBusFabric();
|
||||||
|
var otherInstance = fabric.Connect("pod-a");
|
||||||
|
await SubscribeThisInstanceAsync(fabric);
|
||||||
|
|
||||||
|
WriteSharedOptions(enabled: true);
|
||||||
|
|
||||||
|
var user = CreateUser();
|
||||||
|
var library = new CollectionFolder { Path = _libraryPath, Name = "Movies" };
|
||||||
|
|
||||||
|
// This replica answers out of its cache from here on.
|
||||||
|
Assert.True(library.IsVisible(user));
|
||||||
|
|
||||||
|
// The admin disables the library on the other replica: it writes the shared directory and says so.
|
||||||
|
WriteSharedOptions(enabled: false);
|
||||||
|
otherInstance.Publish(ConfigurationInvalidationScope.LibraryOptions, _libraryPath);
|
||||||
|
|
||||||
|
Assert.False(library.IsVisible(user));
|
||||||
|
Assert.False(CollectionFolder.GetLibraryOptions(_libraryPath).Enabled);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// A path remap made on another replica has to reach this one, or it keeps resolving media against a
|
||||||
|
/// path that is no longer the library's.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||||
|
[Fact]
|
||||||
|
public async Task LibraryPathRemappedOnAnotherInstance_IsSeenHere()
|
||||||
|
{
|
||||||
|
var fabric = new FakeInvalidationBusFabric();
|
||||||
|
var otherInstance = fabric.Connect("pod-a");
|
||||||
|
await SubscribeThisInstanceAsync(fabric);
|
||||||
|
|
||||||
|
WriteSharedOptions(enabled: true, mediaPath: "/media/old");
|
||||||
|
Assert.Equal("/media/old", CollectionFolder.GetLibraryOptions(_libraryPath).PathInfos[0].Path);
|
||||||
|
|
||||||
|
WriteSharedOptions(enabled: true, mediaPath: "/media/new");
|
||||||
|
otherInstance.Publish(ConfigurationInvalidationScope.LibraryOptions, _libraryPath);
|
||||||
|
|
||||||
|
Assert.Equal("/media/new", CollectionFolder.GetLibraryOptions(_libraryPath).PathInfos[0].Path);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Saving library options here has to tell the other replicas, which is the half of the exchange the
|
||||||
|
/// tests above take as given.
|
||||||
|
/// </summary>
|
||||||
|
[Fact]
|
||||||
|
public void SaveLibraryOptions_AnnouncesTheLibraryToTheOtherInstances()
|
||||||
|
{
|
||||||
|
var fabric = new FakeInvalidationBusFabric();
|
||||||
|
ConfigurationInvalidation? received = null;
|
||||||
|
|
||||||
|
var otherInstance = fabric.Connect("pod-b");
|
||||||
|
otherInstance.Subscribe(invalidation => received = invalidation);
|
||||||
|
CollectionFolder.InvalidationBus = fabric.Connect("pod-a");
|
||||||
|
|
||||||
|
CollectionFolder.SaveLibraryOptions(_libraryPath, new LibraryOptions { Enabled = false });
|
||||||
|
|
||||||
|
Assert.NotNull(received);
|
||||||
|
Assert.Equal(ConfigurationInvalidationScope.LibraryOptions, received.Scope);
|
||||||
|
Assert.Equal(_libraryPath, received.Target);
|
||||||
|
}
|
||||||
|
|
||||||
|
private async Task SubscribeThisInstanceAsync(FakeInvalidationBusFabric fabric)
|
||||||
|
{
|
||||||
|
var bus = fabric.Connect("pod-b");
|
||||||
|
CollectionFolder.InvalidationBus = bus;
|
||||||
|
|
||||||
|
var subscriber = new ConfigurationInvalidationSubscriber(
|
||||||
|
bus,
|
||||||
|
Mock.Of<IConfigurationManager>(),
|
||||||
|
NullLogger<ConfigurationInvalidationSubscriber>.Instance);
|
||||||
|
|
||||||
|
await subscriber.StartAsync(CancellationToken.None);
|
||||||
|
}
|
||||||
|
|
||||||
|
private void WriteSharedOptions(bool enabled, string mediaPath = "/media")
|
||||||
|
{
|
||||||
|
// Written the way the other replica writes it, straight onto the shared directory.
|
||||||
|
var options = new LibraryOptions { Enabled = enabled, PathInfos = [new MediaPathInfo(mediaPath)] };
|
||||||
|
_serializer.SerializeToFile(options, Path.Combine(_libraryPath, "options.xml"));
|
||||||
|
}
|
||||||
|
|
||||||
|
private static User CreateUser()
|
||||||
|
{
|
||||||
|
var user = new User("propagation", "auth", "reset");
|
||||||
|
user.SetPermission(PermissionKind.EnableAllFolders, true);
|
||||||
|
return user;
|
||||||
|
}
|
||||||
|
}
|
||||||
+104
@@ -0,0 +1,104 @@
|
|||||||
|
using System;
|
||||||
|
using System.Threading.Tasks;
|
||||||
|
using Emby.Server.Implementations.Configuration;
|
||||||
|
using MediaBrowser.Common.Configuration;
|
||||||
|
using Microsoft.Extensions.Logging.Abstractions;
|
||||||
|
using StackExchange.Redis;
|
||||||
|
using Testcontainers.Redis;
|
||||||
|
using Xunit;
|
||||||
|
|
||||||
|
namespace Jellyfin.Server.Implementations.Tests.Configuration;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Round-trips <see cref="RedisConfigurationInvalidationBus"/> through a real Redis, the transport two
|
||||||
|
/// replicas actually use to tell each other that the shared configuration directory has changed.
|
||||||
|
/// </summary>
|
||||||
|
[Trait("Category", "RequiresDocker")]
|
||||||
|
public sealed class RedisConfigurationInvalidationBusTests : IAsyncLifetime
|
||||||
|
{
|
||||||
|
private readonly RedisContainer _container;
|
||||||
|
private IConnectionMultiplexer? _redis;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Initializes a new instance of the <see cref="RedisConfigurationInvalidationBusTests"/> class.
|
||||||
|
/// </summary>
|
||||||
|
public RedisConfigurationInvalidationBusTests()
|
||||||
|
{
|
||||||
|
_container = new RedisBuilder("redis:7-alpine").Build();
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public async ValueTask InitializeAsync()
|
||||||
|
{
|
||||||
|
await _container.StartAsync();
|
||||||
|
_redis = await ConnectionMultiplexer.ConnectAsync(_container.GetConnectionString());
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public async ValueTask DisposeAsync()
|
||||||
|
{
|
||||||
|
if (_redis is not null)
|
||||||
|
{
|
||||||
|
await _redis.DisposeAsync();
|
||||||
|
}
|
||||||
|
|
||||||
|
await _container.DisposeAsync();
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// A notice published by one replica reaches the other, carrying enough to invalidate one entry.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||||
|
[Fact]
|
||||||
|
public async Task Publish_ReachesTheOtherInstance()
|
||||||
|
{
|
||||||
|
var received = new TaskCompletionSource<ConfigurationInvalidation>();
|
||||||
|
var instanceA = CreateBus("pod-a");
|
||||||
|
var instanceB = CreateBus("pod-b");
|
||||||
|
|
||||||
|
instanceB.Subscribe(invalidation => received.TrySetResult(invalidation));
|
||||||
|
|
||||||
|
instanceA.Publish(ConfigurationInvalidationScope.LibraryOptions, "/media/movies");
|
||||||
|
|
||||||
|
var invalidation = await received.Task.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken);
|
||||||
|
Assert.Equal(ConfigurationInvalidationScope.LibraryOptions, invalidation.Scope);
|
||||||
|
Assert.Equal("/media/movies", invalidation.Target);
|
||||||
|
Assert.Equal("pod-a", invalidation.OriginId);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// The publishing replica has already applied the change to its own cache, so it must not act on its
|
||||||
|
/// own notice and reload what it just wrote.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||||
|
[Fact]
|
||||||
|
public async Task Publish_IsNotDeliveredToThePublisher()
|
||||||
|
{
|
||||||
|
var ownNotice = new TaskCompletionSource<ConfigurationInvalidation>();
|
||||||
|
var otherNotice = new TaskCompletionSource<ConfigurationInvalidation>();
|
||||||
|
var instanceA = CreateBus("pod-a");
|
||||||
|
var instanceB = CreateBus("pod-b");
|
||||||
|
|
||||||
|
instanceA.Subscribe(invalidation => ownNotice.TrySetResult(invalidation));
|
||||||
|
instanceB.Subscribe(invalidation => otherNotice.TrySetResult(invalidation));
|
||||||
|
|
||||||
|
instanceA.Publish(ConfigurationInvalidationScope.SystemConfiguration, null);
|
||||||
|
|
||||||
|
// Ordering is per channel, so B having the notice means A would have had it too.
|
||||||
|
await otherNotice.Task.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken);
|
||||||
|
Assert.False(ownNotice.Task.IsCompleted);
|
||||||
|
}
|
||||||
|
|
||||||
|
private RedisConfigurationInvalidationBus CreateBus(string originId)
|
||||||
|
{
|
||||||
|
Environment.SetEnvironmentVariable("JELLYFIN_INSTANCE_ID", originId);
|
||||||
|
try
|
||||||
|
{
|
||||||
|
return new RedisConfigurationInvalidationBus(_redis!, NullLogger<RedisConfigurationInvalidationBus>.Instance);
|
||||||
|
}
|
||||||
|
finally
|
||||||
|
{
|
||||||
|
Environment.SetEnvironmentVariable("JELLYFIN_INSTANCE_ID", null);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
+205
@@ -0,0 +1,205 @@
|
|||||||
|
using System;
|
||||||
|
using System.IO;
|
||||||
|
using System.Threading;
|
||||||
|
using System.Threading.Tasks;
|
||||||
|
using Emby.Server.Implementations;
|
||||||
|
using Emby.Server.Implementations.Configuration;
|
||||||
|
using Emby.Server.Implementations.Serialization;
|
||||||
|
using MediaBrowser.Common.Configuration;
|
||||||
|
using MediaBrowser.Common.Net;
|
||||||
|
using MediaBrowser.Controller.Configuration;
|
||||||
|
using Microsoft.Extensions.Logging.Abstractions;
|
||||||
|
using Xunit;
|
||||||
|
|
||||||
|
namespace Jellyfin.Server.Implementations.Tests.Configuration;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Applying an invalidation re-raises the same update events a local save raises, and some consumers of
|
||||||
|
/// those events answer an update by writing - <c>RecordingsManager</c> creating the recording folders for
|
||||||
|
/// the <c>livetv</c> key, <c>ApplicationHost</c> clearing <c>IsPortAuthorized</c> on a port change. The
|
||||||
|
/// instance that did not write must not repeat those writes, and no write it is induced into must reach
|
||||||
|
/// the bus.
|
||||||
|
/// </summary>
|
||||||
|
public sealed class RemoteInvalidationApplyTests : IDisposable
|
||||||
|
{
|
||||||
|
private readonly string _root;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Initializes a new instance of the <see cref="RemoteInvalidationApplyTests"/> class.
|
||||||
|
/// </summary>
|
||||||
|
public RemoteInvalidationApplyTests()
|
||||||
|
{
|
||||||
|
_root = Path.Combine(Path.GetTempPath(), "jf-config-apply-" + Guid.NewGuid().ToString("N"));
|
||||||
|
Directory.CreateDirectory(_root);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public void Dispose()
|
||||||
|
{
|
||||||
|
try
|
||||||
|
{
|
||||||
|
Directory.Delete(_root, true);
|
||||||
|
}
|
||||||
|
catch (IOException)
|
||||||
|
{
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// A consumer that knows nothing about the bus - a plugin, or anything reached transitively from one -
|
||||||
|
/// can answer an applied invalidation by writing. That write must not become a notice of its own.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||||
|
[Fact]
|
||||||
|
public async Task NamedInvalidation_InducingAWriteOnTheReceiver_DoesNotPublishBack()
|
||||||
|
{
|
||||||
|
var fabric = new FakeInvalidationBusFabric();
|
||||||
|
var instanceA = CreateInstance(fabric, "pod-a");
|
||||||
|
var instanceB = await CreateSubscribedInstanceAsync(fabric, "pod-b");
|
||||||
|
|
||||||
|
var receivedByA = 0;
|
||||||
|
instanceA.InvalidationBus.Subscribe(_ => Interlocked.Increment(ref receivedByA));
|
||||||
|
|
||||||
|
var writesByB = 0;
|
||||||
|
instanceB.NamedConfigurationUpdated += (_, e) =>
|
||||||
|
{
|
||||||
|
// One shot: the induced write raises the event again on this instance.
|
||||||
|
if (Interlocked.Increment(ref writesByB) > 1)
|
||||||
|
{
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
var configuration = (NetworkConfiguration)instanceB.GetConfiguration(e.Key);
|
||||||
|
configuration.PublishedServerUriBySubnet = ["10.0.0.0/8=example"];
|
||||||
|
instanceB.SaveConfiguration(e.Key, configuration);
|
||||||
|
};
|
||||||
|
|
||||||
|
var updated = instanceA.GetNetworkConfiguration();
|
||||||
|
updated.EnableRemoteAccess = false;
|
||||||
|
instanceA.SaveConfiguration(NetworkConfigurationStore.StoreKey, updated);
|
||||||
|
|
||||||
|
Assert.Equal(0, receivedByA);
|
||||||
|
|
||||||
|
// The point of the bus still holds: B is not left on its stale copy.
|
||||||
|
Assert.False(instanceB.GetNetworkConfiguration().EnableRemoteAccess);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// The shape of <c>RecordingsManager</c>: a consumer that answers a named configuration update by
|
||||||
|
/// writing has to be able to tell that the write was another instance's, and skip it.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||||
|
[Fact]
|
||||||
|
public async Task NamedInvalidation_WithAWriteTriggeringConsumer_DoesNotDuplicateTheWrite()
|
||||||
|
{
|
||||||
|
var fabric = new FakeInvalidationBusFabric();
|
||||||
|
var instanceA = CreateInstance(fabric, "pod-a");
|
||||||
|
var instanceB = await CreateSubscribedInstanceAsync(fabric, "pod-b");
|
||||||
|
|
||||||
|
var receivedByA = 0;
|
||||||
|
instanceA.InvalidationBus.Subscribe(_ => Interlocked.Increment(ref receivedByA));
|
||||||
|
|
||||||
|
var writesByB = 0;
|
||||||
|
instanceB.NamedConfigurationUpdated += (_, e) =>
|
||||||
|
{
|
||||||
|
if (ConfigurationInvalidationContext.IsApplyingRemoteInvalidation)
|
||||||
|
{
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
Interlocked.Increment(ref writesByB);
|
||||||
|
};
|
||||||
|
|
||||||
|
var updated = instanceA.GetNetworkConfiguration();
|
||||||
|
updated.EnableRemoteAccess = false;
|
||||||
|
instanceA.SaveConfiguration(NetworkConfigurationStore.StoreKey, updated);
|
||||||
|
|
||||||
|
Assert.Equal(0, writesByB);
|
||||||
|
Assert.Equal(0, receivedByA);
|
||||||
|
Assert.False(instanceB.GetNetworkConfiguration().EnableRemoteAccess);
|
||||||
|
|
||||||
|
// A read-only consumer is still told, which is what the invalidation exists for.
|
||||||
|
var refreshes = 0;
|
||||||
|
instanceB.NamedConfigurationUpdated += (_, _) => Interlocked.Increment(ref refreshes);
|
||||||
|
|
||||||
|
updated.EnableRemoteAccess = true;
|
||||||
|
instanceA.SaveConfiguration(NetworkConfigurationStore.StoreKey, updated);
|
||||||
|
|
||||||
|
Assert.Equal(1, refreshes);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// The <c>ApplicationHost.IsPortAuthorized</c> class of consumer: the system configuration event is
|
||||||
|
/// queued rather than raised inline, so the fix has to survive the hop onto the thread pool.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||||
|
[Fact]
|
||||||
|
public async Task SystemInvalidation_InducingAQueuedWriteOnTheReceiver_DoesNotPublishBack()
|
||||||
|
{
|
||||||
|
var fabric = new FakeInvalidationBusFabric();
|
||||||
|
var instanceA = CreateInstance(fabric, "pod-a");
|
||||||
|
var instanceB = await CreateSubscribedInstanceAsync(fabric, "pod-b");
|
||||||
|
|
||||||
|
var receivedByA = 0;
|
||||||
|
instanceA.InvalidationBus.Subscribe(_ => Interlocked.Increment(ref receivedByA));
|
||||||
|
|
||||||
|
var applied = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||||
|
var handled = 0;
|
||||||
|
instanceB.ConfigurationUpdated += (_, _) =>
|
||||||
|
{
|
||||||
|
if (Interlocked.Increment(ref handled) > 1)
|
||||||
|
{
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
instanceB.Configuration.IsPortAuthorized = false;
|
||||||
|
instanceB.SaveConfiguration();
|
||||||
|
applied.TrySetResult();
|
||||||
|
};
|
||||||
|
|
||||||
|
instanceA.Configuration.QuickConnectAvailable = false;
|
||||||
|
instanceA.SaveConfiguration();
|
||||||
|
|
||||||
|
await applied.Task.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken);
|
||||||
|
|
||||||
|
Assert.Equal(0, receivedByA);
|
||||||
|
Assert.False(instanceB.Configuration.QuickConnectAvailable);
|
||||||
|
}
|
||||||
|
|
||||||
|
private async Task<ServerConfigurationManager> CreateSubscribedInstanceAsync(FakeInvalidationBusFabric fabric, string originId)
|
||||||
|
{
|
||||||
|
var instance = CreateInstance(fabric, originId);
|
||||||
|
var subscriber = new ConfigurationInvalidationSubscriber(
|
||||||
|
instance.InvalidationBus,
|
||||||
|
instance,
|
||||||
|
NullLogger<ConfigurationInvalidationSubscriber>.Instance);
|
||||||
|
|
||||||
|
await subscriber.StartAsync(CancellationToken.None);
|
||||||
|
return instance;
|
||||||
|
}
|
||||||
|
|
||||||
|
private ServerConfigurationManager CreateInstance(FakeInvalidationBusFabric fabric, string originId)
|
||||||
|
{
|
||||||
|
var paths = new ServerApplicationPaths(
|
||||||
|
Ensure("data"),
|
||||||
|
Ensure("log"),
|
||||||
|
Ensure("config"),
|
||||||
|
Ensure("cache"),
|
||||||
|
Ensure("web"));
|
||||||
|
|
||||||
|
var manager = new ServerConfigurationManager(paths, NullLoggerFactory.Instance, new MyXmlSerializer())
|
||||||
|
{
|
||||||
|
InvalidationBus = fabric.Connect(originId)
|
||||||
|
};
|
||||||
|
|
||||||
|
manager.AddParts([new NetworkConfigurationFactory()]);
|
||||||
|
return manager;
|
||||||
|
}
|
||||||
|
|
||||||
|
private string Ensure(string name)
|
||||||
|
{
|
||||||
|
var path = Path.Combine(_root, name);
|
||||||
|
Directory.CreateDirectory(path);
|
||||||
|
return path;
|
||||||
|
}
|
||||||
|
}
|
||||||
+147
@@ -0,0 +1,147 @@
|
|||||||
|
using System;
|
||||||
|
using System.IO;
|
||||||
|
using System.Threading;
|
||||||
|
using System.Threading.Tasks;
|
||||||
|
using Emby.Server.Implementations;
|
||||||
|
using Emby.Server.Implementations.Configuration;
|
||||||
|
using Emby.Server.Implementations.Serialization;
|
||||||
|
using MediaBrowser.Common.Configuration;
|
||||||
|
using MediaBrowser.Common.Net;
|
||||||
|
using MediaBrowser.Controller.Configuration;
|
||||||
|
using Microsoft.Extensions.Logging.Abstractions;
|
||||||
|
using StackExchange.Redis;
|
||||||
|
using Xunit;
|
||||||
|
|
||||||
|
namespace Jellyfin.Server.Implementations.Tests.Configuration;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Two independently constructed <see cref="ServerConfigurationManager"/> instances over one configuration
|
||||||
|
/// directory are the in-process stand-in for two replicas sharing one <c>/config</c> mount: what either of
|
||||||
|
/// them writes, the other has to pick up without being restarted.
|
||||||
|
/// </summary>
|
||||||
|
public sealed class SharedConfigurationPropagationTests : IDisposable
|
||||||
|
{
|
||||||
|
private readonly string _root;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Initializes a new instance of the <see cref="SharedConfigurationPropagationTests"/> class.
|
||||||
|
/// </summary>
|
||||||
|
public SharedConfigurationPropagationTests()
|
||||||
|
{
|
||||||
|
_root = Path.Combine(Path.GetTempPath(), "jf-config-prop-" + Guid.NewGuid().ToString("N"));
|
||||||
|
Directory.CreateDirectory(_root);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
public void Dispose()
|
||||||
|
{
|
||||||
|
try
|
||||||
|
{
|
||||||
|
Directory.Delete(_root, true);
|
||||||
|
}
|
||||||
|
catch (IOException)
|
||||||
|
{
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// A system configuration setting tightened on one replica has to hold on every other replica, not
|
||||||
|
/// only on the one that served the admin's request.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||||
|
[Fact]
|
||||||
|
public async Task SystemConfigurationSavedOnOneInstance_IsSeenByAnotherWithoutRestart()
|
||||||
|
{
|
||||||
|
var fabric = new FakeInvalidationBusFabric();
|
||||||
|
var instanceA = CreateInstance(fabric, "pod-a");
|
||||||
|
var instanceB = await CreateSubscribedInstanceAsync(fabric, "pod-b");
|
||||||
|
|
||||||
|
// B has the pre-change configuration in hand before A writes, as a running replica would.
|
||||||
|
Assert.True(instanceB.Configuration.QuickConnectAvailable);
|
||||||
|
|
||||||
|
instanceA.Configuration.QuickConnectAvailable = false;
|
||||||
|
instanceA.SaveConfiguration();
|
||||||
|
|
||||||
|
Assert.False(instanceB.Configuration.QuickConnectAvailable);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// The same has to hold for the named configurations, which are cached per key and never reloaded.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||||
|
[Fact]
|
||||||
|
public async Task NamedConfigurationSavedOnOneInstance_IsSeenByAnotherWithoutRestart()
|
||||||
|
{
|
||||||
|
var fabric = new FakeInvalidationBusFabric();
|
||||||
|
var instanceA = CreateInstance(fabric, "pod-a");
|
||||||
|
var instanceB = await CreateSubscribedInstanceAsync(fabric, "pod-b");
|
||||||
|
|
||||||
|
Assert.True(instanceB.GetNetworkConfiguration().EnableRemoteAccess);
|
||||||
|
|
||||||
|
var updated = instanceA.GetNetworkConfiguration();
|
||||||
|
updated.EnableRemoteAccess = false;
|
||||||
|
instanceA.SaveConfiguration(NetworkConfigurationStore.StoreKey, updated);
|
||||||
|
|
||||||
|
Assert.False(instanceB.GetNetworkConfiguration().EnableRemoteAccess);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// A replica that cannot reach the bus keeps serving: the admin's save still lands on the shared
|
||||||
|
/// directory, and the only loss is that the other replicas are not told about it.
|
||||||
|
/// </summary>
|
||||||
|
[Fact]
|
||||||
|
public void SaveConfiguration_WithUnreachableBus_DoesNotThrow()
|
||||||
|
{
|
||||||
|
using var multiplexer = ConnectionMultiplexer.Connect("127.0.0.1:1,abortConnect=false,connectTimeout=200,connectRetry=1,syncTimeout=200");
|
||||||
|
var bus = new RedisConfigurationInvalidationBus(multiplexer, NullLogger<RedisConfigurationInvalidationBus>.Instance);
|
||||||
|
|
||||||
|
bus.Subscribe(_ => throw new InvalidOperationException("Nothing can be delivered by an unreachable bus."));
|
||||||
|
|
||||||
|
var instance = CreateInstance(new FakeInvalidationBusFabric(), "pod-a");
|
||||||
|
instance.InvalidationBus = bus;
|
||||||
|
|
||||||
|
instance.Configuration.QuickConnectAvailable = false;
|
||||||
|
instance.SaveConfiguration();
|
||||||
|
instance.SaveConfiguration(NetworkConfigurationStore.StoreKey, instance.GetNetworkConfiguration());
|
||||||
|
|
||||||
|
Assert.False(instance.Configuration.QuickConnectAvailable);
|
||||||
|
}
|
||||||
|
|
||||||
|
private async Task<ServerConfigurationManager> CreateSubscribedInstanceAsync(FakeInvalidationBusFabric fabric, string originId)
|
||||||
|
{
|
||||||
|
var instance = CreateInstance(fabric, originId);
|
||||||
|
var subscriber = new ConfigurationInvalidationSubscriber(
|
||||||
|
instance.InvalidationBus,
|
||||||
|
instance,
|
||||||
|
NullLogger<ConfigurationInvalidationSubscriber>.Instance);
|
||||||
|
|
||||||
|
await subscriber.StartAsync(CancellationToken.None);
|
||||||
|
return instance;
|
||||||
|
}
|
||||||
|
|
||||||
|
private ServerConfigurationManager CreateInstance(FakeInvalidationBusFabric fabric, string originId)
|
||||||
|
{
|
||||||
|
// Every instance has its own paths object, all of them pointing at the one shared directory.
|
||||||
|
var paths = new ServerApplicationPaths(
|
||||||
|
Ensure("data"),
|
||||||
|
Ensure("log"),
|
||||||
|
Ensure("config"),
|
||||||
|
Ensure("cache"),
|
||||||
|
Ensure("web"));
|
||||||
|
|
||||||
|
var manager = new ServerConfigurationManager(paths, NullLoggerFactory.Instance, new MyXmlSerializer())
|
||||||
|
{
|
||||||
|
InvalidationBus = fabric.Connect(originId)
|
||||||
|
};
|
||||||
|
|
||||||
|
manager.AddParts([new NetworkConfigurationFactory()]);
|
||||||
|
return manager;
|
||||||
|
}
|
||||||
|
|
||||||
|
private string Ensure(string name)
|
||||||
|
{
|
||||||
|
var path = Path.Combine(_root, name);
|
||||||
|
Directory.CreateDirectory(path);
|
||||||
|
return path;
|
||||||
|
}
|
||||||
|
}
|
||||||
+2
@@ -31,6 +31,8 @@
|
|||||||
<ItemGroup>
|
<ItemGroup>
|
||||||
<ProjectReference Include="..\..\Emby.Server.Implementations\Emby.Server.Implementations.csproj" />
|
<ProjectReference Include="..\..\Emby.Server.Implementations\Emby.Server.Implementations.csproj" />
|
||||||
<ProjectReference Include="..\..\Jellyfin.Server.Implementations\Jellyfin.Server.Implementations.csproj" />
|
<ProjectReference Include="..\..\Jellyfin.Server.Implementations\Jellyfin.Server.Implementations.csproj" />
|
||||||
|
<ProjectReference Include="..\..\MediaBrowser.Providers\MediaBrowser.Providers.csproj" />
|
||||||
|
<ProjectReference Include="..\..\src\Jellyfin.LiveTv\Jellyfin.LiveTv.csproj" />
|
||||||
<ProjectReference Include="..\Jellyfin.Server.Integration.Tests\Jellyfin.Server.Integration.Tests.csproj" />
|
<ProjectReference Include="..\Jellyfin.Server.Integration.Tests\Jellyfin.Server.Integration.Tests.csproj" />
|
||||||
<ProjectReference Include="..\..\src\Jellyfin.Database\Jellyfin.Database.Implementations\Jellyfin.Database.Implementations.csproj" />
|
<ProjectReference Include="..\..\src\Jellyfin.Database\Jellyfin.Database.Implementations\Jellyfin.Database.Implementations.csproj" />
|
||||||
</ItemGroup>
|
</ItemGroup>
|
||||||
|
|||||||
+57
-17
@@ -8,6 +8,7 @@ using MediaBrowser.Common.Extensions;
|
|||||||
using MediaBrowser.Controller.Authentication;
|
using MediaBrowser.Controller.Authentication;
|
||||||
using MediaBrowser.Controller.Configuration;
|
using MediaBrowser.Controller.Configuration;
|
||||||
using MediaBrowser.Controller.Net;
|
using MediaBrowser.Controller.Net;
|
||||||
|
using MediaBrowser.Controller.QuickConnect;
|
||||||
using MediaBrowser.Model.Configuration;
|
using MediaBrowser.Model.Configuration;
|
||||||
using Moq;
|
using Moq;
|
||||||
using Xunit;
|
using Xunit;
|
||||||
@@ -40,6 +41,8 @@ namespace Jellyfin.Server.Implementations.Tests.QuickConnect
|
|||||||
ConfigureMembers = true
|
ConfigureMembers = true
|
||||||
}).Inject(configManager.Object);
|
}).Inject(configManager.Object);
|
||||||
|
|
||||||
|
_fixture.Inject<IQuickConnectStore>(new InMemoryQuickConnectStore());
|
||||||
|
|
||||||
// User object contains circular references.
|
// User object contains circular references.
|
||||||
_fixture.Behaviors.OfType<ThrowingRecursionBehavior>().ToList()
|
_fixture.Behaviors.OfType<ThrowingRecursionBehavior>().ToList()
|
||||||
.ForEach(b => _fixture.Behaviors.Remove(b));
|
.ForEach(b => _fixture.Behaviors.Remove(b));
|
||||||
@@ -60,8 +63,8 @@ namespace Jellyfin.Server.Implementations.Tests.QuickConnect
|
|||||||
[InlineData("Device", "", "Client", "1.0.0")]
|
[InlineData("Device", "", "Client", "1.0.0")]
|
||||||
[InlineData("Device", "DeviceId", "", "1.0.0")]
|
[InlineData("Device", "DeviceId", "", "1.0.0")]
|
||||||
[InlineData("Device", "DeviceId", "Client", "")]
|
[InlineData("Device", "DeviceId", "Client", "")]
|
||||||
public void TryConnect_InvalidAuthorizationInfo_ThrowsArgumentException(string device, string deviceId, string client, string version)
|
public async Task TryConnect_InvalidAuthorizationInfo_ThrowsArgumentException(string device, string deviceId, string client, string version)
|
||||||
=> Assert.Throws<ArgumentException>(() => _quickConnectManager.TryConnect(
|
=> await Assert.ThrowsAsync<ArgumentException>(() => _quickConnectManager.TryConnect(
|
||||||
new AuthorizationInfo
|
new AuthorizationInfo
|
||||||
{
|
{
|
||||||
Device = device,
|
Device = device,
|
||||||
@@ -71,17 +74,17 @@ namespace Jellyfin.Server.Implementations.Tests.QuickConnect
|
|||||||
}));
|
}));
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public void TryConnect_QuickConnectUnavailable_ThrowsAuthenticationException()
|
public async Task TryConnect_QuickConnectUnavailable_ThrowsAuthenticationException()
|
||||||
{
|
{
|
||||||
_config.QuickConnectAvailable = false;
|
_config.QuickConnectAvailable = false;
|
||||||
Assert.Throws<AuthenticationException>(() => _quickConnectManager.TryConnect(_quickConnectAuthInfo));
|
await Assert.ThrowsAsync<AuthenticationException>(() => _quickConnectManager.TryConnect(_quickConnectAuthInfo));
|
||||||
}
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public void CheckRequestStatus_QuickConnectUnavailable_ThrowsAuthenticationException()
|
public async Task CheckRequestStatus_QuickConnectUnavailable_ThrowsAuthenticationException()
|
||||||
{
|
{
|
||||||
_config.QuickConnectAvailable = false;
|
_config.QuickConnectAvailable = false;
|
||||||
Assert.Throws<AuthenticationException>(() => _quickConnectManager.CheckRequestStatus(string.Empty));
|
await Assert.ThrowsAsync<AuthenticationException>(() => _quickConnectManager.CheckRequestStatus(string.Empty));
|
||||||
}
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
@@ -92,10 +95,10 @@ namespace Jellyfin.Server.Implementations.Tests.QuickConnect
|
|||||||
}
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public void GetAuthorizedRequest_QuickConnectUnavailable_ThrowsAuthenticationException()
|
public async Task GetAuthorizedRequest_QuickConnectUnavailable_ThrowsAuthenticationException()
|
||||||
{
|
{
|
||||||
_config.QuickConnectAvailable = false;
|
_config.QuickConnectAvailable = false;
|
||||||
Assert.Throws<AuthenticationException>(() => _quickConnectManager.GetAuthorizedRequest(string.Empty));
|
await Assert.ThrowsAsync<AuthenticationException>(() => _quickConnectManager.GetAuthorizedRequest(string.Empty));
|
||||||
}
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
@@ -106,34 +109,71 @@ namespace Jellyfin.Server.Implementations.Tests.QuickConnect
|
|||||||
}
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public void CheckRequestStatus_QuickConnectAvailable_Success()
|
public async Task CheckRequestStatus_QuickConnectAvailable_Success()
|
||||||
{
|
{
|
||||||
_config.QuickConnectAvailable = true;
|
_config.QuickConnectAvailable = true;
|
||||||
var res1 = _quickConnectManager.TryConnect(_quickConnectAuthInfo);
|
var res1 = await _quickConnectManager.TryConnect(_quickConnectAuthInfo);
|
||||||
var res2 = _quickConnectManager.CheckRequestStatus(res1.Secret);
|
var res2 = await _quickConnectManager.CheckRequestStatus(res1.Secret);
|
||||||
Assert.Equal(res1, res2);
|
Assert.Equal(res1.Secret, res2.Secret);
|
||||||
|
Assert.Equal(res1.Code, res2.Code);
|
||||||
}
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public void CheckRequestStatus_UnknownSecret_ThrowsResourceNotFoundException()
|
public async Task CheckRequestStatus_UnknownSecret_ThrowsResourceNotFoundException()
|
||||||
{
|
{
|
||||||
_config.QuickConnectAvailable = true;
|
_config.QuickConnectAvailable = true;
|
||||||
Assert.Throws<ResourceNotFoundException>(() => _quickConnectManager.CheckRequestStatus("Unknown secret"));
|
await Assert.ThrowsAsync<ResourceNotFoundException>(() => _quickConnectManager.CheckRequestStatus("Unknown secret"));
|
||||||
}
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public void GetAuthorizedRequest_UnknownSecret_ThrowsResourceNotFoundException()
|
public async Task GetAuthorizedRequest_UnknownSecret_ThrowsResourceNotFoundException()
|
||||||
{
|
{
|
||||||
_config.QuickConnectAvailable = true;
|
_config.QuickConnectAvailable = true;
|
||||||
Assert.Throws<ResourceNotFoundException>(() => _quickConnectManager.GetAuthorizedRequest("Unknown secret"));
|
await Assert.ThrowsAsync<ResourceNotFoundException>(() => _quickConnectManager.GetAuthorizedRequest("Unknown secret"));
|
||||||
}
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public async Task AuthorizeRequest_QuickConnectAvailable_Success()
|
public async Task AuthorizeRequest_QuickConnectAvailable_Success()
|
||||||
{
|
{
|
||||||
_config.QuickConnectAvailable = true;
|
_config.QuickConnectAvailable = true;
|
||||||
var res = _quickConnectManager.TryConnect(_quickConnectAuthInfo);
|
var res = await _quickConnectManager.TryConnect(_quickConnectAuthInfo);
|
||||||
Assert.True(await _quickConnectManager.AuthorizeRequest(Guid.Empty, res.Code));
|
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()
|
||||||
|
{
|
||||||
|
_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<ResourceNotFoundException>(() => _quickConnectManager.GetAuthorizedRequest(res.Secret));
|
||||||
|
}
|
||||||
|
|
||||||
|
private async Task<bool> AuthorizeAsync(string code)
|
||||||
|
{
|
||||||
|
try
|
||||||
|
{
|
||||||
|
return await _quickConnectManager.AuthorizeRequest(Guid.Empty, code).ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
catch (InvalidOperationException)
|
||||||
|
{
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+100
-9
@@ -1,6 +1,8 @@
|
|||||||
using System;
|
using System;
|
||||||
using System.Collections.Generic;
|
using System.Collections.Generic;
|
||||||
|
using System.IO;
|
||||||
using System.Linq;
|
using System.Linq;
|
||||||
|
using System.Reflection;
|
||||||
using System.Runtime.CompilerServices;
|
using System.Runtime.CompilerServices;
|
||||||
using Emby.Server.Implementations.ScheduledTasks.Tasks;
|
using Emby.Server.Implementations.ScheduledTasks.Tasks;
|
||||||
using MediaBrowser.Controller.ScheduledTasks;
|
using MediaBrowser.Controller.ScheduledTasks;
|
||||||
@@ -11,6 +13,14 @@ namespace Jellyfin.Server.Implementations.Tests.ScheduledTasks;
|
|||||||
|
|
||||||
public class ScanLeaderOptionsTests
|
public class ScanLeaderOptionsTests
|
||||||
{
|
{
|
||||||
|
private static readonly Assembly[] _taskAssemblies =
|
||||||
|
{
|
||||||
|
typeof(DeleteTranscodeFileTask).Assembly,
|
||||||
|
typeof(MediaBrowser.Providers.Lyric.LyricScheduledTask).Assembly,
|
||||||
|
typeof(Jellyfin.LiveTv.Guide.RefreshGuideScheduledTask).Assembly,
|
||||||
|
typeof(Jellyfin.MediaEncoding.Hls.ScheduledTasks.KeyframeExtractionScheduledTask).Assembly
|
||||||
|
};
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// A gated key that matches no registered task silently stops gating anything, so the default
|
/// A gated key that matches no registered task silently stops gating anything, so the default
|
||||||
/// set is pinned to the task keys that actually exist in the build.
|
/// set is pinned to the task keys that actually exist in the build.
|
||||||
@@ -18,28 +28,92 @@ public class ScanLeaderOptionsTests
|
|||||||
[Fact]
|
[Fact]
|
||||||
public void DefaultGatedTaskKeys_Should_MatchRegisteredScheduledTasks()
|
public void DefaultGatedTaskKeys_Should_MatchRegisteredScheduledTasks()
|
||||||
{
|
{
|
||||||
var registeredKeys = DiscoverScheduledTaskKeys();
|
var registeredKeys = DiscoverScheduledTaskKeys(_taskAssemblies);
|
||||||
|
|
||||||
Assert.NotEmpty(registeredKeys);
|
Assert.NotEmpty(registeredKeys);
|
||||||
Assert.Empty(new ScanLeaderOptions().GatedTaskKeys.Except(registeredKeys, StringComparer.Ordinal));
|
|
||||||
|
var unmatched = new ScanLeaderOptions().GatedTaskKeys.Except(registeredKeys, StringComparer.Ordinal).ToList();
|
||||||
|
Assert.True(
|
||||||
|
unmatched.Count == 0,
|
||||||
|
$"Gated keys match no scheduled task: {string.Join(", ", unmatched)}. Known keys: {string.Join(", ", registeredKeys.Order(StringComparer.Ordinal))}");
|
||||||
}
|
}
|
||||||
|
|
||||||
private static HashSet<string> DiscoverScheduledTaskKeys()
|
/// <summary>
|
||||||
|
/// A key dropped from the default set silently un-gates that task on every replica, so the whole
|
||||||
|
/// set is pinned against a hand-maintained expectation rather than read back from the options.
|
||||||
|
/// </summary>
|
||||||
|
[Fact]
|
||||||
|
public void DefaultGatedTaskKeys_Should_BeTheExpectedSet()
|
||||||
{
|
{
|
||||||
var keys = new HashSet<string>(StringComparer.Ordinal);
|
string[] expected =
|
||||||
var assemblies = new[]
|
|
||||||
{
|
{
|
||||||
typeof(DeleteTranscodeFileTask).Assembly,
|
"AudioNormalization",
|
||||||
typeof(Jellyfin.MediaEncoding.Hls.ScheduledTasks.KeyframeExtractionScheduledTask).Assembly
|
"CleanupUserDataTask",
|
||||||
|
"DownloadLyrics",
|
||||||
|
"DownloadSubtitles",
|
||||||
|
"KeyframeExtraction",
|
||||||
|
"MoveTrickplayImages",
|
||||||
|
"OptimizeDatabaseTask",
|
||||||
|
"PluginUpdates",
|
||||||
|
"RefreshChapterImages",
|
||||||
|
"RefreshGuide",
|
||||||
|
"RefreshInternetChannels",
|
||||||
|
"RefreshLibrary",
|
||||||
|
"RefreshPeople",
|
||||||
|
"RefreshTrickplayImages",
|
||||||
|
"TaskExtractMediaSegments",
|
||||||
|
"TmdbRefreshUpcomingEpisodes"
|
||||||
};
|
};
|
||||||
|
|
||||||
foreach (var type in assemblies.SelectMany(a => a.GetTypes()))
|
var actual = new ScanLeaderOptions().GatedTaskKeys;
|
||||||
|
var missing = expected.Except(actual, StringComparer.Ordinal).ToList();
|
||||||
|
var unexpected = actual.Except(expected, StringComparer.Ordinal).ToList();
|
||||||
|
|
||||||
|
Assert.True(
|
||||||
|
missing.Count == 0 && unexpected.Count == 0,
|
||||||
|
$"Default gated task keys drifted. Missing: {Describe(missing)}. Unexpected: {Describe(unexpected)}.");
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// The key universe is only as complete as the assemblies it is read from, so a task added to an
|
||||||
|
/// unscanned assembly must fail here rather than narrow what the previous test can catch.
|
||||||
|
/// </summary>
|
||||||
|
[Fact]
|
||||||
|
public void TaskAssemblies_Should_CoverEveryAssemblyDeclaringScheduledTasks()
|
||||||
|
{
|
||||||
|
var scanned = _taskAssemblies.Select(a => a.GetName().Name).ToHashSet(StringComparer.Ordinal);
|
||||||
|
var missing = new List<string>();
|
||||||
|
|
||||||
|
foreach (var path in Directory.EnumerateFiles(AppContext.BaseDirectory, "*.dll"))
|
||||||
{
|
{
|
||||||
if (type.IsAbstract || type.IsInterface || !typeof(IScheduledTask).IsAssignableFrom(type))
|
var name = Path.GetFileNameWithoutExtension(path);
|
||||||
|
if (scanned.Contains(name)
|
||||||
|
|| name.EndsWith(".Tests", StringComparison.Ordinal)
|
||||||
|
|| !(name.StartsWith("Jellyfin.", StringComparison.Ordinal)
|
||||||
|
|| name.StartsWith("Emby.", StringComparison.Ordinal)
|
||||||
|
|| name.StartsWith("MediaBrowser.", StringComparison.Ordinal)))
|
||||||
{
|
{
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if (GetScheduledTaskTypes(Assembly.LoadFrom(path)).Any())
|
||||||
|
{
|
||||||
|
missing.Add(name);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
Assert.True(missing.Count == 0, $"Assemblies declaring scheduled tasks but not scanned: {string.Join(", ", missing)}");
|
||||||
|
}
|
||||||
|
|
||||||
|
private static string Describe(IReadOnlyCollection<string> keys)
|
||||||
|
=> keys.Count == 0 ? "none" : string.Join(", ", keys.Order(StringComparer.Ordinal));
|
||||||
|
|
||||||
|
private static HashSet<string> DiscoverScheduledTaskKeys(IEnumerable<Assembly> assemblies)
|
||||||
|
{
|
||||||
|
var keys = new HashSet<string>(StringComparer.Ordinal);
|
||||||
|
|
||||||
|
foreach (var type in assemblies.SelectMany(GetScheduledTaskTypes))
|
||||||
|
{
|
||||||
// Task keys are constant expressions, so an uninitialised instance is enough to read
|
// Task keys are constant expressions, so an uninitialised instance is enough to read
|
||||||
// them without standing up each task's dependency graph.
|
// them without standing up each task's dependency graph.
|
||||||
var task = (IScheduledTask)RuntimeHelpers.GetUninitializedObject(type);
|
var task = (IScheduledTask)RuntimeHelpers.GetUninitializedObject(type);
|
||||||
@@ -48,4 +122,21 @@ public class ScanLeaderOptionsTests
|
|||||||
|
|
||||||
return keys;
|
return keys;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private static IEnumerable<Type> GetScheduledTaskTypes(Assembly assembly)
|
||||||
|
{
|
||||||
|
Type?[] types;
|
||||||
|
try
|
||||||
|
{
|
||||||
|
types = assembly.GetTypes();
|
||||||
|
}
|
||||||
|
catch (ReflectionTypeLoadException ex)
|
||||||
|
{
|
||||||
|
types = ex.Types;
|
||||||
|
}
|
||||||
|
|
||||||
|
return types
|
||||||
|
.Where(t => t is not null && !t.IsAbstract && !t.IsInterface && typeof(IScheduledTask).IsAssignableFrom(t))
|
||||||
|
.Select(t => t!);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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,
|
/// 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.
|
/// a correctly set variable is dropped and the feature it configures stays off without any error.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
|
[Collection("JellyfinSectionConfiguration")]
|
||||||
public sealed class JellyfinSectionConfigurationTests : IDisposable
|
public sealed class JellyfinSectionConfigurationTests : IDisposable
|
||||||
{
|
{
|
||||||
private const string RedisKey = "Jellyfin:TranscodeStore:RedisConnectionString";
|
private const string RedisKey = "Jellyfin:TranscodeStore:RedisConnectionString";
|
||||||
|
|||||||
@@ -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;
|
||||||
|
|
||||||
|
/// <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 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>
|
||||||
|
/// 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() => _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;
|
||||||
|
|
||||||
|
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)
|
||||||
|
{
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,91 @@
|
|||||||
|
using System;
|
||||||
|
using System.Threading.Tasks;
|
||||||
|
using StackExchange.Redis;
|
||||||
|
using Testcontainers.Redis;
|
||||||
|
|
||||||
|
namespace Jellyfin.Server.Tests.HighAvailability;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Hands out a Redis server for the tests that need one. A server named by <c>JELLYFIN_TEST_REDIS</c> 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.
|
||||||
|
/// </summary>
|
||||||
|
public sealed class RedisTestServer : IAsyncDisposable
|
||||||
|
{
|
||||||
|
/// <summary>
|
||||||
|
/// The connection string of an already running server.
|
||||||
|
/// </summary>
|
||||||
|
public const string ConnectionStringVariable = "JELLYFIN_TEST_REDIS";
|
||||||
|
|
||||||
|
private readonly RedisContainer? _container;
|
||||||
|
|
||||||
|
private RedisTestServer(RedisContainer? container, string connectionString)
|
||||||
|
{
|
||||||
|
_container = container;
|
||||||
|
ConnectionString = connectionString;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Gets the connection string of the running server.
|
||||||
|
/// </summary>
|
||||||
|
public string ConnectionString { get; }
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Starts or attaches to a Redis server and waits until it accepts connections.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>The running server.</returns>
|
||||||
|
public static async Task<RedisTestServer> 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;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// 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>A new multiplexer.</returns>
|
||||||
|
public async Task<IConnectionMultiplexer> ConnectAsync()
|
||||||
|
=> await ConnectionMultiplexer.ConnectAsync(ConnectionString).ConfigureAwait(false);
|
||||||
|
|
||||||
|
/// <inheritdoc/>
|
||||||
|
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);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,234 @@
|
|||||||
|
using System;
|
||||||
|
using System.Collections.Generic;
|
||||||
|
using System.Security.Claims;
|
||||||
|
using System.Threading;
|
||||||
|
using System.Threading.Tasks;
|
||||||
|
using Emby.Server.Implementations.Data;
|
||||||
|
using Jellyfin.Api.Constants;
|
||||||
|
using Jellyfin.Api.Controllers;
|
||||||
|
using Jellyfin.Database.Implementations;
|
||||||
|
using Jellyfin.Database.Implementations.DbConfiguration;
|
||||||
|
using Jellyfin.Database.Implementations.Entities;
|
||||||
|
using Jellyfin.Database.Implementations.Locking;
|
||||||
|
using Jellyfin.Database.Providers.PostgreSQL;
|
||||||
|
using Jellyfin.Server.Implementations.Item;
|
||||||
|
using Jellyfin.Server.Tests.Migrations;
|
||||||
|
using MediaBrowser.Controller.Dto;
|
||||||
|
using MediaBrowser.Controller.Entities;
|
||||||
|
using MediaBrowser.Controller.Library;
|
||||||
|
using MediaBrowser.Controller.Persistence;
|
||||||
|
using MediaBrowser.Controller.TV;
|
||||||
|
using MediaBrowser.Model.Querying;
|
||||||
|
using Microsoft.AspNetCore.Http;
|
||||||
|
using Microsoft.AspNetCore.Mvc;
|
||||||
|
using Microsoft.EntityFrameworkCore;
|
||||||
|
using Microsoft.Extensions.Logging.Abstractions;
|
||||||
|
using Moq;
|
||||||
|
using Npgsql;
|
||||||
|
using Xunit;
|
||||||
|
using BaseItemKind = Jellyfin.Data.Enums.BaseItemKind;
|
||||||
|
using User = Jellyfin.Database.Implementations.Entities.User;
|
||||||
|
|
||||||
|
namespace Jellyfin.Server.Tests.Item;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Drives the Next Up cutoff from the controller into a real PostgreSQL. The model binder hands a
|
||||||
|
/// query-string date over as <see cref="DateTimeKind.Unspecified"/>, and Npgsql refuses to write anything
|
||||||
|
/// but <see cref="DateTimeKind.Utc"/> to <c>timestamp with time zone</c>; SQLite takes every kind, so an
|
||||||
|
/// unnormalised cutoff only ever fails here.
|
||||||
|
/// </summary>
|
||||||
|
[Trait("Category", "RequiresDocker")]
|
||||||
|
public sealed class PostgreSqlNextUpServiceTests : IAsyncLifetime
|
||||||
|
{
|
||||||
|
private static readonly Guid _libraryId = Guid.Parse("aaaaaaaa-0000-0000-0000-000000000001");
|
||||||
|
private static readonly Guid _otherLibraryId = Guid.Parse("aaaaaaaa-0000-0000-0000-000000000002");
|
||||||
|
private static readonly Guid _userId = Guid.Parse("bbbbbbbb-0000-0000-0000-000000000001");
|
||||||
|
|
||||||
|
private static readonly Guid _recentWatchedId = Guid.Parse("cccccccc-0000-0000-0000-000000000001");
|
||||||
|
private static readonly Guid _recentOlderId = Guid.Parse("cccccccc-0000-0000-0000-000000000002");
|
||||||
|
private static readonly Guid _staleWatchedId = Guid.Parse("cccccccc-0000-0000-0000-000000000003");
|
||||||
|
private static readonly Guid _unwatchedId = Guid.Parse("cccccccc-0000-0000-0000-000000000004");
|
||||||
|
private static readonly Guid _foreignLibraryId = Guid.Parse("cccccccc-0000-0000-0000-000000000005");
|
||||||
|
|
||||||
|
private static readonly DateTime _recentPlayedAt = new DateTime(2026, 3, 1, 12, 0, 0, DateTimeKind.Utc);
|
||||||
|
private static readonly DateTime _stalePlayedAt = new DateTime(2020, 1, 1, 12, 0, 0, DateTimeKind.Utc);
|
||||||
|
|
||||||
|
private readonly ItemTypeLookup _itemTypeLookup = new();
|
||||||
|
private readonly User _user = new User("next-up", "auth", "reset") { Id = _userId };
|
||||||
|
|
||||||
|
private PostgreSqlTestServer _server = null!;
|
||||||
|
private NpgsqlDataSource _dataSource = null!;
|
||||||
|
private NextUpService _service = null!;
|
||||||
|
|
||||||
|
public async ValueTask InitializeAsync()
|
||||||
|
{
|
||||||
|
_server = await PostgreSqlTestServer.StartAsync().ConfigureAwait(false);
|
||||||
|
var connectionString = await _server.CreateDatabaseAsync("next_up_service", TestContext.Current.CancellationToken).ConfigureAwait(false);
|
||||||
|
_dataSource = new NpgsqlDataSourceBuilder(connectionString).Build();
|
||||||
|
|
||||||
|
var context = CreateDbContext();
|
||||||
|
await using (context.ConfigureAwait(false))
|
||||||
|
{
|
||||||
|
await context.Database.EnsureCreatedAsync(TestContext.Current.CancellationToken).ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
|
||||||
|
var factory = new Mock<IDbContextFactory<JellyfinDbContext>>();
|
||||||
|
factory.Setup(f => f.CreateDbContext()).Returns(CreateDbContext);
|
||||||
|
factory.Setup(f => f.CreateDbContextAsync(It.IsAny<CancellationToken>())).ReturnsAsync(CreateDbContext);
|
||||||
|
|
||||||
|
_service = new NextUpService(factory.Object, _itemTypeLookup, new Mock<IItemQueryHelpers>().Object);
|
||||||
|
|
||||||
|
await SeedAsync().ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
|
||||||
|
public async ValueTask DisposeAsync()
|
||||||
|
{
|
||||||
|
await _dataSource.DisposeAsync().ConfigureAwait(false);
|
||||||
|
await _server.DisposeAsync().ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// A cutoff on the query string, which the model binder leaves unspecified.
|
||||||
|
/// </summary>
|
||||||
|
[Fact]
|
||||||
|
public void GetNextUpSeriesKeys_WithSuppliedCutoff_DropsSeriesPlayedBeforeIt()
|
||||||
|
{
|
||||||
|
var cutoff = RunController(new DateTime(2025, 1, 1, 0, 0, 0, DateTimeKind.Unspecified)).NextUpDateCutoff;
|
||||||
|
|
||||||
|
var keys = _service.GetNextUpSeriesKeys(CreateFilter(), cutoff);
|
||||||
|
|
||||||
|
Assert.Equal(new[] { "series-recent" }, keys);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// The home-screen row, where the client sends no cutoff and the query default stands in.
|
||||||
|
/// </summary>
|
||||||
|
[Fact]
|
||||||
|
public void GetNextUpSeriesKeys_WithoutSuppliedCutoff_ReturnsWatchedSeriesNewestFirst()
|
||||||
|
{
|
||||||
|
var cutoff = RunController(null).NextUpDateCutoff;
|
||||||
|
|
||||||
|
var keys = _service.GetNextUpSeriesKeys(CreateFilter(), cutoff);
|
||||||
|
|
||||||
|
Assert.Equal(new[] { "series-recent", "series-stale" }, keys);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Calls <c>GET /Shows/NextUp</c> and hands back the query it built for the series lookup.
|
||||||
|
/// </summary>
|
||||||
|
private NextUpQuery RunController(DateTime? nextUpDateCutoff)
|
||||||
|
{
|
||||||
|
var userManager = new Mock<IUserManager>();
|
||||||
|
userManager.Setup(m => m.GetUserById(_userId)).Returns(_user);
|
||||||
|
|
||||||
|
var dtoService = new Mock<IDtoService>();
|
||||||
|
dtoService.Setup(s => s.GetBaseItemDtos(
|
||||||
|
It.IsAny<IReadOnlyList<BaseItem>>(),
|
||||||
|
It.IsAny<DtoOptions>(),
|
||||||
|
It.IsAny<User>(),
|
||||||
|
It.IsAny<BaseItem>(),
|
||||||
|
It.IsAny<bool>()))
|
||||||
|
.Returns([]);
|
||||||
|
|
||||||
|
NextUpQuery? captured = null;
|
||||||
|
var tvSeriesManager = new Mock<ITVSeriesManager>();
|
||||||
|
tvSeriesManager.Setup(m => m.GetNextUp(It.IsAny<NextUpQuery>(), It.IsAny<DtoOptions>()))
|
||||||
|
.Callback<NextUpQuery, DtoOptions>((query, _) => captured = query)
|
||||||
|
.Returns(new QueryResult<BaseItem>());
|
||||||
|
|
||||||
|
var controller = new TvShowsController(
|
||||||
|
userManager.Object,
|
||||||
|
new Mock<ILibraryManager>().Object,
|
||||||
|
dtoService.Object,
|
||||||
|
tvSeriesManager.Object)
|
||||||
|
{
|
||||||
|
ControllerContext = new ControllerContext
|
||||||
|
{
|
||||||
|
HttpContext = new DefaultHttpContext
|
||||||
|
{
|
||||||
|
User = new ClaimsPrincipal(new ClaimsIdentity([new Claim(InternalClaimTypes.UserId, _userId.ToString("D"))], "Test"))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
controller.GetNextUp(null, null, null, [], null, null, null, null, [], null, nextUpDateCutoff);
|
||||||
|
|
||||||
|
return captured!;
|
||||||
|
}
|
||||||
|
|
||||||
|
private InternalItemsQuery CreateFilter()
|
||||||
|
{
|
||||||
|
return new InternalItemsQuery(_user) { TopParentIds = [_libraryId] };
|
||||||
|
}
|
||||||
|
|
||||||
|
private JellyfinDbContext CreateDbContext()
|
||||||
|
{
|
||||||
|
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 SeedAsync()
|
||||||
|
{
|
||||||
|
var context = CreateDbContext();
|
||||||
|
await using (context.ConfigureAwait(false))
|
||||||
|
{
|
||||||
|
context.Users.Add(_user);
|
||||||
|
|
||||||
|
// The newest play of a series decides its place, so the older episode must not pull it down.
|
||||||
|
var recentWatched = AddEpisode(context, _recentWatchedId, "series-recent", _libraryId);
|
||||||
|
AddUserData(context, recentWatched, _recentPlayedAt);
|
||||||
|
var recentOlder = AddEpisode(context, _recentOlderId, "series-recent", _libraryId);
|
||||||
|
AddUserData(context, recentOlder, _stalePlayedAt);
|
||||||
|
|
||||||
|
var staleWatched = AddEpisode(context, _staleWatchedId, "series-stale", _libraryId);
|
||||||
|
AddUserData(context, staleWatched, _stalePlayedAt);
|
||||||
|
|
||||||
|
// Never played, and played but outside the requested libraries: both stay out.
|
||||||
|
AddEpisode(context, _unwatchedId, "series-unwatched", _libraryId);
|
||||||
|
var foreign = AddEpisode(context, _foreignLibraryId, "series-foreign", _otherLibraryId);
|
||||||
|
AddUserData(context, foreign, _recentPlayedAt);
|
||||||
|
|
||||||
|
await context.SaveChangesAsync(TestContext.Current.CancellationToken).ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private BaseItemEntity AddEpisode(JellyfinDbContext context, Guid id, string seriesKey, Guid topParentId)
|
||||||
|
{
|
||||||
|
var episode = new BaseItemEntity
|
||||||
|
{
|
||||||
|
Id = id,
|
||||||
|
Type = _itemTypeLookup.BaseItemKindNames[BaseItemKind.Episode],
|
||||||
|
Name = seriesKey + "-" + id.ToString("N"),
|
||||||
|
SeriesPresentationUniqueKey = seriesKey,
|
||||||
|
PresentationUniqueKey = id.ToString("N"),
|
||||||
|
TopParentId = topParentId,
|
||||||
|
MediaType = "Video",
|
||||||
|
IsFolder = false,
|
||||||
|
IsVirtualItem = false
|
||||||
|
};
|
||||||
|
|
||||||
|
context.BaseItems.Add(episode);
|
||||||
|
return episode;
|
||||||
|
}
|
||||||
|
|
||||||
|
private void AddUserData(JellyfinDbContext context, BaseItemEntity item, DateTime lastPlayedDate)
|
||||||
|
{
|
||||||
|
context.UserData.Add(new UserData
|
||||||
|
{
|
||||||
|
CustomDataKey = item.Id.ToString("N"),
|
||||||
|
ItemId = item.Id,
|
||||||
|
Item = item,
|
||||||
|
UserId = _userId,
|
||||||
|
User = _user,
|
||||||
|
LastPlayedDate = lastPlayedDate,
|
||||||
|
Played = true,
|
||||||
|
PlayCount = 1
|
||||||
|
});
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -12,6 +12,7 @@
|
|||||||
<PackageReference Include="Microsoft.NET.Test.Sdk" />
|
<PackageReference Include="Microsoft.NET.Test.Sdk" />
|
||||||
<PackageReference Include="Npgsql" />
|
<PackageReference Include="Npgsql" />
|
||||||
<PackageReference Include="Testcontainers.PostgreSql" />
|
<PackageReference Include="Testcontainers.PostgreSql" />
|
||||||
|
<PackageReference Include="Testcontainers.Redis" />
|
||||||
<PackageReference Include="xunit.v3" />
|
<PackageReference Include="xunit.v3" />
|
||||||
<PackageReference Include="xunit.runner.visualstudio">
|
<PackageReference Include="xunit.runner.visualstudio">
|
||||||
<PrivateAssets>all</PrivateAssets>
|
<PrivateAssets>all</PrivateAssets>
|
||||||
|
|||||||
@@ -0,0 +1,369 @@
|
|||||||
|
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>
|
||||||
|
/// 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.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||||
|
[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);
|
||||||
|
|
||||||
|
for (var attempt = 0; attempt < 25; attempt++)
|
||||||
|
{
|
||||||
|
var initiated = await replicaA.Manager.TryConnect(AuthorizationInfoFor(attempt));
|
||||||
|
await replicaB.Manager.AuthorizeRequest(user.Id, initiated.Code);
|
||||||
|
|
||||||
|
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);
|
||||||
|
|
||||||
|
// And it stays consumed for every later attempt, on any replica.
|
||||||
|
await Assert.ThrowsAsync<ResourceNotFoundException>(() => replicaB.Manager.GetAuthorizedRequest(initiated.Secret));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <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 (InvalidOperationException)
|
||||||
|
{
|
||||||
|
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);
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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;
|
||||||
|
|
||||||
|
/// <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 + ",abortConnect=false");
|
||||||
|
|
||||||
|
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.
|
||||||
|
/// </summary>
|
||||||
|
[Fact]
|
||||||
|
public void NoEnvironmentVariable_SelectsTheProcessLocalStore()
|
||||||
|
{
|
||||||
|
Environment.SetEnvironmentVariable(RedisConnectionStringVariable, null);
|
||||||
|
|
||||||
|
using var provider = BuildProvider();
|
||||||
|
|
||||||
|
Assert.IsType<InMemoryQuickConnectStore>(provider.GetRequiredService<IQuickConnectStore>());
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// 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.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||||
|
[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<IQuickConnectStore>();
|
||||||
|
Assert.IsType<InMemoryQuickConnectStore>(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<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();
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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;
|
||||||
|
|
||||||
|
/// <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>
|
||||||
|
/// 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.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||||
|
[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);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// 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.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||||
|
[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));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// 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.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||||
|
[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<JsonException>(() => instance.Store.GetRequestBySecretAsync(request.Secret, CancellationToken));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// 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.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||||
|
[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));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// 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.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||||
|
[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));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Exchanging during an outage fails loudly and spends nothing, so the token is still there to be
|
||||||
|
/// handed out once when Redis comes back.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||||
|
[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));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// 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.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A <see cref="Task"/> representing the asynchronous operation.</returns>
|
||||||
|
[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));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <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));
|
||||||
|
}
|
||||||
|
|
||||||
|
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<Task> 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<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);
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user