From a9d6c749fba3c9fa0bce93a5362674a9b0d3d0f9 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Mon, 21 Sep 2026 00:24:01 +1000 Subject: [PATCH] keep an applied invalidation from inducing a write that publishes back --- .../AppBase/BaseConfigurationManager.cs | 4 +- .../ApplicationHost.cs | 6 +- .../ConfigurationInvalidationSubscriber.cs | 5 + .../ConfigurationInvalidationBusExtensions.cs | 30 +++ .../ConfigurationInvalidationContext.cs | 57 +++++ .../Entities/CollectionFolder.cs | 4 +- .../Recordings/RecordingsManager.cs | 7 + .../RemoteInvalidationApplyTests.cs | 205 ++++++++++++++++++ 8 files changed, 312 insertions(+), 6 deletions(-) create mode 100644 MediaBrowser.Common/Configuration/ConfigurationInvalidationBusExtensions.cs create mode 100644 MediaBrowser.Common/Configuration/ConfigurationInvalidationContext.cs create mode 100644 tests/Jellyfin.Server.Implementations.Tests/Configuration/RemoteInvalidationApplyTests.cs diff --git a/Emby.Server.Implementations/AppBase/BaseConfigurationManager.cs b/Emby.Server.Implementations/AppBase/BaseConfigurationManager.cs index 3c2342406d..a5c4f3fa8b 100644 --- a/Emby.Server.Implementations/AppBase/BaseConfigurationManager.cs +++ b/Emby.Server.Implementations/AppBase/BaseConfigurationManager.cs @@ -176,7 +176,7 @@ namespace Emby.Server.Implementations.AppBase OnConfigurationUpdated(); - InvalidationBus.Publish(ConfigurationInvalidationScope.SystemConfiguration, null); + InvalidationBus.PublishLocalWrite(ConfigurationInvalidationScope.SystemConfiguration, null); } /// @@ -359,7 +359,7 @@ namespace Emby.Server.Implementations.AppBase OnNamedConfigurationUpdated(key, configuration); - InvalidationBus.Publish(ConfigurationInvalidationScope.NamedConfiguration, key); + InvalidationBus.PublishLocalWrite(ConfigurationInvalidationScope.NamedConfiguration, key); } /// diff --git a/Emby.Server.Implementations/ApplicationHost.cs b/Emby.Server.Implementations/ApplicationHost.cs index c4db55fa71..106c938f8a 100644 --- a/Emby.Server.Implementations/ApplicationHost.cs +++ b/Emby.Server.Implementations/ApplicationHost.cs @@ -802,8 +802,10 @@ namespace Emby.Server.Implementations var requiresRestart = false; var networkConfiguration = ConfigurationManager.GetNetworkConfiguration(); - // Don't do anything if these haven't been set yet - if (HttpPort != 0 && HttpsPort != 0) + // Don't do anything if these haven't been set yet, and don't clear the authorization flag on + // behalf of another instance: it already wrote that flag along with the port change, so + // repeating the save here only races it. + if (HttpPort != 0 && HttpsPort != 0 && !ConfigurationInvalidationContext.IsApplyingRemoteInvalidation) { // Need to restart if ports have changed if (networkConfiguration.InternalHttpPort != HttpPort diff --git a/Emby.Server.Implementations/Configuration/ConfigurationInvalidationSubscriber.cs b/Emby.Server.Implementations/Configuration/ConfigurationInvalidationSubscriber.cs index 80fe0160b1..c28fbb60a3 100644 --- a/Emby.Server.Implementations/Configuration/ConfigurationInvalidationSubscriber.cs +++ b/Emby.Server.Implementations/Configuration/ConfigurationInvalidationSubscriber.cs @@ -48,6 +48,11 @@ namespace Emby.Server.Implementations.Configuration { 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: diff --git a/MediaBrowser.Common/Configuration/ConfigurationInvalidationBusExtensions.cs b/MediaBrowser.Common/Configuration/ConfigurationInvalidationBusExtensions.cs new file mode 100644 index 0000000000..c556c33e5d --- /dev/null +++ b/MediaBrowser.Common/Configuration/ConfigurationInvalidationBusExtensions.cs @@ -0,0 +1,30 @@ +namespace MediaBrowser.Common.Configuration +{ + /// + /// Extensions for . + /// + public static class ConfigurationInvalidationBusExtensions + { + /// + /// Announces a write this instance originated, and stays silent for a write induced by an + /// invalidation another instance published. + /// + /// + /// 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. + /// + /// The bus. + /// The cache that was written. + /// The configuration key or library path that was written, if any. + public static void PublishLocalWrite(this IConfigurationInvalidationBus bus, ConfigurationInvalidationScope scope, string? target) + { + if (ConfigurationInvalidationContext.IsApplyingRemoteInvalidation) + { + return; + } + + bus.Publish(scope, target); + } + } +} diff --git a/MediaBrowser.Common/Configuration/ConfigurationInvalidationContext.cs b/MediaBrowser.Common/Configuration/ConfigurationInvalidationContext.cs new file mode 100644 index 0000000000..57f0de4dd1 --- /dev/null +++ b/MediaBrowser.Common/Configuration/ConfigurationInvalidationContext.cs @@ -0,0 +1,57 @@ +using System; +using System.Threading; + +namespace MediaBrowser.Common.Configuration +{ + /// + /// 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. + /// + /// + /// 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. + /// + public static class ConfigurationInvalidationContext + { + private static readonly AsyncLocal _applyingRemoteInvalidation = new(); + + /// + /// Gets a value indicating whether the current flow of control is applying an invalidation + /// published by another instance rather than handling a local save. + /// + public static bool IsApplyingRemoteInvalidation => _applyingRemoteInvalidation.Value; + + /// + /// Marks the current flow of control as applying a remote invalidation until the returned scope is + /// disposed. + /// + /// The scope to dispose once the invalidation has been applied. + 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; + } + } + } +} diff --git a/MediaBrowser.Controller/Entities/CollectionFolder.cs b/MediaBrowser.Controller/Entities/CollectionFolder.cs index 8ebbb72020..004ff8cf72 100644 --- a/MediaBrowser.Controller/Entities/CollectionFolder.cs +++ b/MediaBrowser.Controller/Entities/CollectionFolder.cs @@ -196,7 +196,7 @@ namespace MediaBrowser.Controller.Entities LibraryOptionsUpdated?.Invoke(null, new LibraryOptionsUpdatedEventArgs(path, options)); - InvalidationBus.Publish(ConfigurationInvalidationScope.LibraryOptions, path); + InvalidationBus.PublishLocalWrite(ConfigurationInvalidationScope.LibraryOptions, path); } /// @@ -222,7 +222,7 @@ namespace MediaBrowser.Controller.Entities { InvalidateAllLibraryOptions(); - InvalidationBus.Publish(ConfigurationInvalidationScope.AllLibraryOptions, null); + InvalidationBus.PublishLocalWrite(ConfigurationInvalidationScope.AllLibraryOptions, null); } public override bool IsSaveLocalMetadataEnabled() diff --git a/src/Jellyfin.LiveTv/Recordings/RecordingsManager.cs b/src/Jellyfin.LiveTv/Recordings/RecordingsManager.cs index 62a06370da..ce463ceb3d 100644 --- a/src/Jellyfin.LiveTv/Recordings/RecordingsManager.cs +++ b/src/Jellyfin.LiveTv/Recordings/RecordingsManager.cs @@ -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); diff --git a/tests/Jellyfin.Server.Implementations.Tests/Configuration/RemoteInvalidationApplyTests.cs b/tests/Jellyfin.Server.Implementations.Tests/Configuration/RemoteInvalidationApplyTests.cs new file mode 100644 index 0000000000..6766f05276 --- /dev/null +++ b/tests/Jellyfin.Server.Implementations.Tests/Configuration/RemoteInvalidationApplyTests.cs @@ -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; + +/// +/// Applying an invalidation re-raises the same update events a local save raises, and some consumers of +/// those events answer an update by writing - RecordingsManager creating the recording folders for +/// the livetv key, ApplicationHost clearing IsPortAuthorized 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. +/// +public sealed class RemoteInvalidationApplyTests : IDisposable +{ + private readonly string _root; + + /// + /// Initializes a new instance of the class. + /// + public RemoteInvalidationApplyTests() + { + _root = Path.Combine(Path.GetTempPath(), "jf-config-apply-" + Guid.NewGuid().ToString("N")); + Directory.CreateDirectory(_root); + } + + /// + public void Dispose() + { + try + { + Directory.Delete(_root, true); + } + catch (IOException) + { + } + } + + /// + /// 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. + /// + /// A representing the asynchronous operation. + [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); + } + + /// + /// The shape of RecordingsManager: 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. + /// + /// A representing the asynchronous operation. + [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); + } + + /// + /// The ApplicationHost.IsPortAuthorized 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. + /// + /// A representing the asynchronous operation. + [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 CreateSubscribedInstanceAsync(FakeInvalidationBusFabric fabric, string originId) + { + var instance = CreateInstance(fabric, originId); + var subscriber = new ConfigurationInvalidationSubscriber( + instance.InvalidationBus, + instance, + NullLogger.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; + } +}