Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 483c739fb1 | |||
| a9d6c749fb | |||
| b662ffa48f |
@@ -87,6 +87,12 @@ namespace Emby.Server.Implementations.AppBase
|
||||
/// <value>The application paths.</value>
|
||||
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>
|
||||
/// Gets or sets the system configuration.
|
||||
/// </summary>
|
||||
@@ -169,6 +175,8 @@ namespace Emby.Server.Implementations.AppBase
|
||||
}
|
||||
|
||||
OnConfigurationUpdated();
|
||||
|
||||
InvalidationBus.PublishLocalWrite(ConfigurationInvalidationScope.SystemConfiguration, null);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -350,6 +358,29 @@ namespace Emby.Server.Implementations.AppBase
|
||||
}
|
||||
|
||||
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>
|
||||
|
||||
@@ -706,6 +706,8 @@ namespace Emby.Server.Implementations
|
||||
BaseItem.UserDataManager = Resolve<IUserDataManager>();
|
||||
CollectionFolder.XmlSerializer = _xmlSerializer;
|
||||
CollectionFolder.ApplicationHost = this;
|
||||
CollectionFolder.InvalidationBus = Resolve<IConfigurationInvalidationBus>();
|
||||
ConfigurationManager.InvalidationBus = CollectionFolder.InvalidationBus;
|
||||
Folder.UserViewManager = Resolve<IUserViewManager>();
|
||||
Folder.CollectionManager = Resolve<ICollectionManager>();
|
||||
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>
|
||||
/// Called when [configuration updated].
|
||||
/// </summary>
|
||||
@@ -797,26 +836,24 @@ namespace Emby.Server.Implementations
|
||||
/// <param name="e">The <see cref="EventArgs"/> instance containing the event data.</param>
|
||||
private void OnConfigurationUpdated(object sender, EventArgs e)
|
||||
{
|
||||
var requiresRestart = false;
|
||||
var networkConfiguration = ConfigurationManager.GetNetworkConfiguration();
|
||||
|
||||
// Don't do anything if these haven't been set yet
|
||||
if (HttpPort != 0 && HttpsPort != 0)
|
||||
{
|
||||
// Need to restart if ports have changed
|
||||
if (networkConfiguration.InternalHttpPort != HttpPort
|
||||
|| networkConfiguration.InternalHttpsPort != HttpsPort)
|
||||
{
|
||||
if (ConfigurationManager.Configuration.IsPortAuthorized)
|
||||
{
|
||||
ConfigurationManager.Configuration.IsPortAuthorized = false;
|
||||
ConfigurationManager.SaveConfiguration();
|
||||
var portChange = EvaluatePortChange(
|
||||
HttpPort,
|
||||
HttpsPort,
|
||||
networkConfiguration.InternalHttpPort,
|
||||
networkConfiguration.InternalHttpsPort,
|
||||
ConfigurationManager.Configuration.IsPortAuthorized,
|
||||
ConfigurationInvalidationContext.IsApplyingRemoteInvalidation);
|
||||
|
||||
requiresRestart = true;
|
||||
}
|
||||
}
|
||||
if (portChange.ClearsPortAuthorization)
|
||||
{
|
||||
ConfigurationManager.Configuration.IsPortAuthorized = false;
|
||||
ConfigurationManager.SaveConfiguration();
|
||||
}
|
||||
|
||||
var requiresRestart = portChange.RequiresRestart;
|
||||
|
||||
if (ValidateSslCertificate(networkConfiguration))
|
||||
{
|
||||
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);
|
||||
}
|
||||
@@ -112,6 +112,10 @@ namespace Jellyfin.Server
|
||||
// instance. Active by default once a Redis connection is configured, no-op otherwise.
|
||||
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);
|
||||
|
||||
foreach (var type in GetExportTypes<ILyricProvider>())
|
||||
{
|
||||
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,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>
|
||||
/// <param name="factories">The factories.</param>
|
||||
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
|
||||
|
||||
@@ -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.Database.Implementations.Entities;
|
||||
using Jellyfin.Extensions.Json;
|
||||
using MediaBrowser.Common.Configuration;
|
||||
using MediaBrowser.Controller.IO;
|
||||
using MediaBrowser.Controller.Library;
|
||||
using MediaBrowser.Controller.Providers;
|
||||
@@ -70,6 +71,12 @@ namespace MediaBrowser.Controller.Entities
|
||||
|
||||
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]
|
||||
public override bool SupportsPlayedStatus => false;
|
||||
|
||||
@@ -188,11 +195,36 @@ namespace MediaBrowser.Controller.Entities
|
||||
XmlSerializer.SerializeToFile(clone, GetLibraryOptionsPath(path));
|
||||
|
||||
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();
|
||||
|
||||
public static void OnCollectionFolderChange()
|
||||
{
|
||||
InvalidateAllLibraryOptions();
|
||||
|
||||
InvalidationBus.PublishLocalWrite(ConfigurationInvalidationScope.AllLibraryOptions, null);
|
||||
}
|
||||
|
||||
public override bool IsSaveLocalMetadataEnabled()
|
||||
{
|
||||
return true;
|
||||
|
||||
@@ -444,6 +444,13 @@ public sealed class RecordingsManager : IRecordingsManager, IDisposable
|
||||
|
||||
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))
|
||||
{
|
||||
await CreateRecordingFolders().ConfigureAwait(false);
|
||||
|
||||
+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;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user