propagate shared-config and library-option changes between instances
ci/woodpecker/push/ci Pipeline was successful
ci/woodpecker/pr/ci Pipeline was successful

Add a Redis pub/sub invalidation bus, publish from the configuration and library-option write paths, and drop the matching local cache entry on the instances that did not write. No-op without a Redis connection string, and fails open when it is unreachable.
This commit is contained in:
2026-09-20 23:47:21 +10:00
parent ec581b5e5f
commit b662ffa48f
16 changed files with 919 additions and 1 deletions
@@ -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.Publish(ConfigurationInvalidationScope.SystemConfiguration, null);
}
/// <summary>
@@ -350,6 +358,29 @@ namespace Emby.Server.Implementations.AppBase
}
OnNamedConfigurationUpdated(key, configuration);
InvalidationBus.Publish(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>();
@@ -0,0 +1,77 @@
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
{
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.");
}
}
}
}
+4
View File
@@ -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,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,13 @@ 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>
/// <param name="key">The named configuration key, or <c>null</c> for the system configuration.</param>
void InvalidateCachedConfiguration(string? key);
}
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.Publish(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.Publish(ConfigurationInvalidationScope.AllLibraryOptions, null);
}
public override bool IsSaveLocalMetadataEnabled()
{
return true;
@@ -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);
}
}
}
}
@@ -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;
}
}
@@ -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);
}
}
}
@@ -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;
}
}