diff --git a/src/modules/Elsa.Common/Multitenancy/Implementations/TenantTaskLifecycleCoordinator.cs b/src/modules/Elsa.Common/Multitenancy/Implementations/TenantTaskLifecycleCoordinator.cs index 03a8670d1..3ec498510 100644 --- a/src/modules/Elsa.Common/Multitenancy/Implementations/TenantTaskLifecycleCoordinator.cs +++ b/src/modules/Elsa.Common/Multitenancy/Implementations/TenantTaskLifecycleCoordinator.cs @@ -282,7 +282,7 @@ public class TenantTaskLifecycleCoordinator(RecurringTaskScheduleManager schedul state.CancellationTokenSource = null; } - private static string GetTenantId(Tenant tenant) => tenant.Id; + private static string GetTenantId(Tenant tenant) => tenant.Id.NormalizeTenantId(); private class TenantRuntimeState { diff --git a/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs b/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs index 707130193..5dfd26fc1 100644 --- a/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs +++ b/src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs @@ -1,5 +1,6 @@ using System.Diagnostics.CodeAnalysis; using System.Reflection; +using Elsa.Common; using Elsa.Common.DistributedHosting; using Elsa.Common.Features; using Elsa.Common.RecurringTasks; @@ -23,6 +24,7 @@ using Elsa.Workflows.Runtime.UIHints; using Medallion.Threading; using Medallion.Threading.FileSystem; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; namespace Elsa.Workflows.Runtime.Features; @@ -252,7 +254,11 @@ public class WorkflowRuntimeFeature(IModule module) : FeatureBase(module) // Graceful-shutdown core (US1 — quiescence machinery). Services - .AddSingleton() + .AddSingleton(sp => new Elsa.Workflows.Runtime.Services.QuiescenceSignal( + sp.GetRequiredService>(), + sp.GetRequiredService(), + sp.GetRequiredService(), + sp.GetRequiredService())) .AddSingleton() .AddSingleton() // Lazy collection breaks the otherwise-circular DI chain QuiescenceSignal → IExecutionCycleRegistry → diff --git a/src/modules/Elsa.Workflows.Runtime/Services/QuiescenceSignal.cs b/src/modules/Elsa.Workflows.Runtime/Services/QuiescenceSignal.cs index c05e65866..5a53b95fb 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/QuiescenceSignal.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/QuiescenceSignal.cs @@ -3,6 +3,7 @@ using Elsa.KeyValues.Contracts; using Elsa.KeyValues.Entities; using Elsa.KeyValues.Models; using Elsa.Workflows.Runtime.Options; +using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Options; namespace Elsa.Workflows.Runtime.Services; @@ -23,6 +24,7 @@ public sealed class QuiescenceSignal : IQuiescenceSignal private readonly IOptions _options; private readonly ISystemClock _clock; private readonly IKeyValueStore? _keyValueStore; + private readonly IServiceScopeFactory? _serviceScopeFactory; private readonly IExecutionCycleRegistry _cycleRegistry; private readonly string _persistenceKey; @@ -33,18 +35,51 @@ public sealed class QuiescenceSignal : IQuiescenceSignal /// down and rebuilt (shell reactivation or host restart), a fresh id is minted, which is what scopes recovery /// in RecoverInterruptedWorkflowsStartupTask. /// + [ActivatorUtilitiesConstructor] public QuiescenceSignal( + IOptions options, + ISystemClock clock, + IExecutionCycleRegistry cycleRegistry, + IServiceScopeFactory serviceScopeFactory, + string? shellName = null, + string? generationId = null) : this(options, clock, cycleRegistry, keyValueStore: null, serviceScopeFactory, shellName, generationId) + { + } + + public QuiescenceSignal( + IOptions options, + ISystemClock clock, + IExecutionCycleRegistry cycleRegistry, + string? shellName = null, + string? generationId = null) : this(options, clock, cycleRegistry, keyValueStore: null, serviceScopeFactory: null, shellName, generationId) + { + } + + /// + /// Creates the signal with a fixed key-value store. Intended for tests and non-container usage. + /// + public static QuiescenceSignal Create( IOptions options, ISystemClock clock, IExecutionCycleRegistry cycleRegistry, IKeyValueStore? keyValueStore = null, string? shellName = null, - string? generationId = null) + string? generationId = null) => new(options, clock, cycleRegistry, keyValueStore, serviceScopeFactory: null, shellName, generationId); + + private QuiescenceSignal( + IOptions options, + ISystemClock clock, + IExecutionCycleRegistry cycleRegistry, + IKeyValueStore? keyValueStore, + IServiceScopeFactory? serviceScopeFactory, + string? shellName, + string? generationId) { _options = options; _clock = clock; _cycleRegistry = cycleRegistry; _keyValueStore = keyValueStore; + _serviceScopeFactory = serviceScopeFactory; _persistenceKey = PersistenceKeyPrefix + (shellName ?? "default"); _state = QuiescenceState.Initial(generationId ?? Guid.NewGuid().ToString("N")); } @@ -73,9 +108,8 @@ public sealed class QuiescenceSignal : IQuiescenceSignal public async ValueTask InitializePersistedStateAsync(CancellationToken cancellationToken) { if (_options.Value.PausePersistence != PausePersistencePolicy.AcrossReactivations) return; - if (_keyValueStore is null) return; - var pair = await _keyValueStore.FindAsync(new KeyValueFilter { Key = _persistenceKey }, cancellationToken); + var pair = await UseKeyValueStoreAsync(store => store.FindAsync(new KeyValueFilter { Key = _persistenceKey }, cancellationToken), defaultValue: (SerializedKeyValuePair?)null); if (pair is null) return; lock (_sync) @@ -188,7 +222,7 @@ public sealed class QuiescenceSignal : IQuiescenceSignal /// private async ValueTask PersistAsync() { - if (_options.Value.PausePersistence != PausePersistencePolicy.AcrossReactivations || _keyValueStore is null) + if (_options.Value.PausePersistence != PausePersistencePolicy.AcrossReactivations) return; await _persistenceMutex.WaitAsync(CancellationToken.None); @@ -196,13 +230,44 @@ public sealed class QuiescenceSignal : IQuiescenceSignal { var live = Volatile.Read(ref _state); if ((live.Reason & QuiescenceReason.AdministrativePause) != 0) - await _keyValueStore.SaveAsync(new SerializedKeyValuePair { Key = _persistenceKey, SerializedValue = live.PauseReasonText ?? string.Empty }, CancellationToken.None); + await UseKeyValueStoreAsync(store => store.SaveAsync(new SerializedKeyValuePair { Key = _persistenceKey, SerializedValue = live.PauseReasonText ?? string.Empty }, CancellationToken.None)); else - await _keyValueStore.DeleteAsync(_persistenceKey, CancellationToken.None); + await UseKeyValueStoreAsync(store => store.DeleteAsync(_persistenceKey, CancellationToken.None)); } finally { _persistenceMutex.Release(); } } + + private async ValueTask UseKeyValueStoreAsync(Func> action, TResult defaultValue) + { + if (_keyValueStore is not null) + return await action(_keyValueStore); + + if (_serviceScopeFactory is null) + return defaultValue; + + using var scope = _serviceScopeFactory.CreateScope(); + var store = scope.ServiceProvider.GetService(); + + return store is null ? defaultValue : await action(store); + } + + private async ValueTask UseKeyValueStoreAsync(Func action) + { + if (_keyValueStore is not null) + { + await action(_keyValueStore); + return; + } + + if (_serviceScopeFactory is null) + return; + + using var scope = _serviceScopeFactory.CreateScope(); + var store = scope.ServiceProvider.GetService(); + if (store is not null) + await action(store); + } } diff --git a/src/modules/Elsa.Workflows.Runtime/ShellFeatures/WorkflowRuntimeFeature.cs b/src/modules/Elsa.Workflows.Runtime/ShellFeatures/WorkflowRuntimeFeature.cs index 345a3c734..95ca01d94 100644 --- a/src/modules/Elsa.Workflows.Runtime/ShellFeatures/WorkflowRuntimeFeature.cs +++ b/src/modules/Elsa.Workflows.Runtime/ShellFeatures/WorkflowRuntimeFeature.cs @@ -165,7 +165,7 @@ public class WorkflowRuntimeFeature : IShellFeature sp.GetRequiredService>(), sp.GetRequiredService(), sp.GetRequiredService(), - sp.GetService(), + sp.GetRequiredService(), shellName: sp.GetService()?.Id)) .AddSingleton() .AddSingleton() diff --git a/test/unit/Elsa.Common.UnitTests/Multitenancy/TenantTaskLifecycleCoordinatorTests.cs b/test/unit/Elsa.Common.UnitTests/Multitenancy/TenantTaskLifecycleCoordinatorTests.cs index af4052c0a..2edfc3f6e 100644 --- a/test/unit/Elsa.Common.UnitTests/Multitenancy/TenantTaskLifecycleCoordinatorTests.cs +++ b/test/unit/Elsa.Common.UnitTests/Multitenancy/TenantTaskLifecycleCoordinatorTests.cs @@ -51,6 +51,17 @@ public class TenantTaskLifecycleCoordinatorTests : IAsyncDisposable Assert.True(_recurringTask.WasStopCalled); } + [Fact] + public async Task ActivateAndDeactivateAsync_WithNullTenantId_TreatsTenantAsDefaultTenant() + { + var tenant = new Tenant { Id = null! }; + + await ActivateAsync(tenant); + await _coordinator.DeactivateTenantAsync(DeactivationArgs(tenant)); + + Assert.True(_recurringTask.WasStopCalled); + } + [Fact] public async Task DisposeAsync_WithActiveTenant_StopsRecurringTasks() { @@ -61,11 +72,17 @@ public class TenantTaskLifecycleCoordinatorTests : IAsyncDisposable // === Instance helpers === - private Task ActivateAsync() => - _coordinator.ActivateTenantAsync(new TenantActivatedEventArgs(_tenant, CreateTenantScope(_tenant, _serviceProvider), CancellationToken.None)); + private Task ActivateAsync(Tenant? tenant = null) + { + tenant ??= _tenant; + return _coordinator.ActivateTenantAsync(new TenantActivatedEventArgs(tenant, CreateTenantScope(tenant, _serviceProvider), CancellationToken.None)); + } private TenantDeactivatedEventArgs DeactivationArgs(CancellationToken cancellationToken = default) => - new(_tenant, CreateTenantScope(_tenant, _serviceProvider), cancellationToken); + DeactivationArgs(_tenant, cancellationToken); + + private TenantDeactivatedEventArgs DeactivationArgs(Tenant tenant, CancellationToken cancellationToken = default) => + new(tenant, CreateTenantScope(tenant, _serviceProvider), cancellationToken); // === Static helpers === diff --git a/test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/QuiescenceSignalPersistenceTests.cs b/test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/QuiescenceSignalPersistenceTests.cs index 5f994b46d..a9baa11c3 100644 --- a/test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/QuiescenceSignalPersistenceTests.cs +++ b/test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/QuiescenceSignalPersistenceTests.cs @@ -4,6 +4,7 @@ using Elsa.KeyValues.Entities; using Elsa.KeyValues.Models; using Elsa.Workflows.Runtime.Options; using Elsa.Workflows.Runtime.Services; +using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Options; using NSubstitute; @@ -27,7 +28,7 @@ public class QuiescenceSignalPersistenceTests public async Task SessionScopedIgnoresKey() { _kv.Pairs["elsa.quiescence.pause.default"] = new SerializedKeyValuePair { Key = "elsa.quiescence.pause.default", SerializedValue = "prior" }; - var sut = new QuiescenceSignal(Microsoft.Extensions.Options.Options.Create(new GracefulShutdownOptions { PausePersistence = PausePersistencePolicy.SessionScoped }), _clock, _cycleRegistry, _kv); + var sut = QuiescenceSignal.Create(Microsoft.Extensions.Options.Options.Create(new GracefulShutdownOptions { PausePersistence = PausePersistencePolicy.SessionScoped }), _clock, _cycleRegistry, _kv); await sut.InitializePersistedStateAsync(CancellationToken.None); @@ -38,7 +39,7 @@ public class QuiescenceSignalPersistenceTests public async Task AcrossReactivationsRestoresPause() { _kv.Pairs["elsa.quiescence.pause.default"] = new SerializedKeyValuePair { Key = "elsa.quiescence.pause.default", SerializedValue = "maintenance" }; - var sut = new QuiescenceSignal(Microsoft.Extensions.Options.Options.Create(new GracefulShutdownOptions { PausePersistence = PausePersistencePolicy.AcrossReactivations }), _clock, _cycleRegistry, _kv); + var sut = QuiescenceSignal.Create(Microsoft.Extensions.Options.Options.Create(new GracefulShutdownOptions { PausePersistence = PausePersistencePolicy.AcrossReactivations }), _clock, _cycleRegistry, _kv); await sut.InitializePersistedStateAsync(CancellationToken.None); @@ -49,7 +50,7 @@ public class QuiescenceSignalPersistenceTests [Fact(DisplayName = "Pause writes the persisted key when policy is AcrossReactivations")] public async Task PauseWritesKey() { - var sut = new QuiescenceSignal(Microsoft.Extensions.Options.Options.Create(new GracefulShutdownOptions { PausePersistence = PausePersistencePolicy.AcrossReactivations }), _clock, _cycleRegistry, _kv); + var sut = QuiescenceSignal.Create(Microsoft.Extensions.Options.Options.Create(new GracefulShutdownOptions { PausePersistence = PausePersistencePolicy.AcrossReactivations }), _clock, _cycleRegistry, _kv); await sut.PauseAsync("migration", "op@ex.com", CancellationToken.None); @@ -60,7 +61,7 @@ public class QuiescenceSignalPersistenceTests [Fact(DisplayName = "Resume clears the persisted key when policy is AcrossReactivations")] public async Task ResumeClearsKey() { - var sut = new QuiescenceSignal(Microsoft.Extensions.Options.Options.Create(new GracefulShutdownOptions { PausePersistence = PausePersistencePolicy.AcrossReactivations }), _clock, _cycleRegistry, _kv); + var sut = QuiescenceSignal.Create(Microsoft.Extensions.Options.Options.Create(new GracefulShutdownOptions { PausePersistence = PausePersistencePolicy.AcrossReactivations }), _clock, _cycleRegistry, _kv); await sut.PauseAsync("migration", "op@ex.com", CancellationToken.None); await sut.ResumeAsync("op@ex.com", CancellationToken.None); @@ -75,8 +76,8 @@ public class QuiescenceSignalPersistenceTests // deployment shared "elsa.quiescence.pause.default" — pausing shell A would re-pause shell B on next // activation. The factory in ShellFeatures/WorkflowRuntimeFeature now injects ShellSettings.Id; this // test locks in the constructor-level contract that shellName is reflected in the persistence key. - var sutA = new QuiescenceSignal(Microsoft.Extensions.Options.Options.Create(new GracefulShutdownOptions { PausePersistence = PausePersistencePolicy.AcrossReactivations }), _clock, _cycleRegistry, _kv, shellName: "shell-a"); - var sutB = new QuiescenceSignal(Microsoft.Extensions.Options.Options.Create(new GracefulShutdownOptions { PausePersistence = PausePersistencePolicy.AcrossReactivations }), _clock, _cycleRegistry, _kv, shellName: "shell-b"); + var sutA = QuiescenceSignal.Create(Microsoft.Extensions.Options.Options.Create(new GracefulShutdownOptions { PausePersistence = PausePersistencePolicy.AcrossReactivations }), _clock, _cycleRegistry, _kv, shellName: "shell-a"); + var sutB = QuiescenceSignal.Create(Microsoft.Extensions.Options.Options.Create(new GracefulShutdownOptions { PausePersistence = PausePersistencePolicy.AcrossReactivations }), _clock, _cycleRegistry, _kv, shellName: "shell-b"); await sutA.PauseAsync("migration-a", "op@ex.com", CancellationToken.None); await sutB.PauseAsync("migration-b", "op@ex.com", CancellationToken.None); @@ -97,7 +98,7 @@ public class QuiescenceSignalPersistenceTests // paused state the operator had already cancelled. The fix serializes persistence on a dedicated semaphore // and re-reads live state inside it, so each I/O writes the most recent in-memory transition. var gatedStore = new GatedFakeKeyValueStore(); - var sut = new QuiescenceSignal(Microsoft.Extensions.Options.Options.Create(new GracefulShutdownOptions { PausePersistence = PausePersistencePolicy.AcrossReactivations }), _clock, _cycleRegistry, gatedStore); + var sut = QuiescenceSignal.Create(Microsoft.Extensions.Options.Options.Create(new GracefulShutdownOptions { PausePersistence = PausePersistencePolicy.AcrossReactivations }), _clock, _cycleRegistry, gatedStore); var pauseTask = sut.PauseAsync("migration", "op@ex.com", CancellationToken.None).AsTask(); await gatedStore.SaveStarted.Task; // Pause has won the persistence mutex; its SaveAsync is in flight (blocked). @@ -125,7 +126,7 @@ public class QuiescenceSignalPersistenceTests // AdministrativePause set in memory with no persisted record. The idempotent fast-path on subsequent // PauseAsync calls (transitioned == false) meant no retry; on the next host restart the runtime came // back unpaused, defeating PausePersistencePolicy.AcrossReactivations. - var sut = new QuiescenceSignal(Microsoft.Extensions.Options.Options.Create(new GracefulShutdownOptions { PausePersistence = PausePersistencePolicy.AcrossReactivations }), _clock, _cycleRegistry, _kv); + var sut = QuiescenceSignal.Create(Microsoft.Extensions.Options.Options.Create(new GracefulShutdownOptions { PausePersistence = PausePersistencePolicy.AcrossReactivations }), _clock, _cycleRegistry, _kv); var cancelled = new CancellationToken(canceled: true); var state = await sut.PauseAsync("migration", "op@ex.com", cancelled); @@ -138,7 +139,7 @@ public class QuiescenceSignalPersistenceTests [Fact(DisplayName = "Null key-value store is tolerated under AcrossReactivations")] public async Task NullKeyValueStoreTolerated() { - var sut = new QuiescenceSignal(Microsoft.Extensions.Options.Options.Create(new GracefulShutdownOptions { PausePersistence = PausePersistencePolicy.AcrossReactivations }), _clock, _cycleRegistry, keyValueStore: null); + var sut = new QuiescenceSignal(Microsoft.Extensions.Options.Options.Create(new GracefulShutdownOptions { PausePersistence = PausePersistencePolicy.AcrossReactivations }), _clock, _cycleRegistry); await sut.InitializePersistedStateAsync(CancellationToken.None); await sut.PauseAsync("migration", null, CancellationToken.None); @@ -148,6 +149,25 @@ public class QuiescenceSignalPersistenceTests Assert.Equal(QuiescenceReason.None, sut.CurrentState.Reason); } + [Fact(DisplayName = "DI construction tolerates scoped key-value store")] + public async Task DiConstructionToleratesScopedKeyValueStore() + { + var services = new ServiceCollection(); + services.AddOptions().Configure(options => options.PausePersistence = PausePersistencePolicy.AcrossReactivations); + services.AddSingleton(_clock); + services.AddSingleton(_cycleRegistry); + services.AddScoped(_ => _kv); + services.AddSingleton(); + + await using var provider = services.BuildServiceProvider(new ServiceProviderOptions { ValidateScopes = true, ValidateOnBuild = true }); + var sut = provider.GetRequiredService(); + + await sut.PauseAsync("migration", "op@ex.com", CancellationToken.None); + + Assert.True(_kv.Pairs.TryGetValue("elsa.quiescence.pause.default", out var pair)); + Assert.Equal("migration", pair.SerializedValue); + } + private sealed class FakeKeyValueStore : IKeyValueStore { public readonly Dictionary Pairs = new(StringComparer.Ordinal);