SemianiakaVY/elsa-core
1
0
Fork 0
Code Issues Pull requests Actions 1 Packages Projects Releases Wiki Activity

Refactor QuiescenceSignal to inject IServiceScopeFactory, enhance tenant ID handling, and expand unit tests with DI capabilities.

Browse source
This commit is contained in:
Sipke Schoorstra 2026-05-03 17:56:37 +02:00
parent 7c01fe8dd4
commit 26b17e35e2
No known key found for this signature in database
GPG key ID: 5C10502B28A4268F
6 changed files with 129 additions and 21 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
src/modules/Elsa.Common/Multitenancy/Implementations/TenantTaskLifecycleCoordinator.cs
View file

@ -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
{

8
src/modules/Elsa.Workflows.Runtime/Features/WorkflowRuntimeFeature.cs
View file

@ -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<IQuiescenceSignal, Elsa.Workflows.Runtime.Services.QuiescenceSignal>()
.AddSingleton<IQuiescenceSignal>(sp => new Elsa.Workflows.Runtime.Services.QuiescenceSignal(
sp.GetRequiredService<IOptions<GracefulShutdownOptions>>(),
sp.GetRequiredService<ISystemClock>(),
sp.GetRequiredService<IExecutionCycleRegistry>(),
sp.GetRequiredService<IServiceScopeFactory>()))
.AddSingleton<IIngressSourceRegistry, Elsa.Workflows.Runtime.Services.IngressSourceRegistry>()
.AddSingleton<IExecutionCycleRegistry, Elsa.Workflows.Runtime.Services.ExecutionCycleRegistry>()
// Lazy collection breaks the otherwise-circular DI chain QuiescenceSignal → IExecutionCycleRegistry →

77
src/modules/Elsa.Workflows.Runtime/Services/QuiescenceSignal.cs
View file

@ -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<GracefulShutdownOptions> _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 <c>RecoverInterruptedWorkflowsStartupTask</c>.
/// </summary>
[ActivatorUtilitiesConstructor]
public QuiescenceSignal(
IOptions<GracefulShutdownOptions> 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<GracefulShutdownOptions> options,
ISystemClock clock,
IExecutionCycleRegistry cycleRegistry,
string? shellName = null,
string? generationId = null) : this(options, clock, cycleRegistry, keyValueStore: null, serviceScopeFactory: null, shellName, generationId)
{
}
/// <summary>
/// Creates the signal with a fixed key-value store. Intended for tests and non-container usage.
/// </summary>
public static QuiescenceSignal Create(
IOptions<GracefulShutdownOptions> 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<GracefulShutdownOptions> 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
/// </remarks>
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<TResult> UseKeyValueStoreAsync<TResult>(Func<IKeyValueStore, Task<TResult>> 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<IKeyValueStore>();
return store is null ? defaultValue : await action(store);
}
private async ValueTask UseKeyValueStoreAsync(Func<IKeyValueStore, Task> 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<IKeyValueStore>();
if (store is not null)
await action(store);
}
}

2
src/modules/Elsa.Workflows.Runtime/ShellFeatures/WorkflowRuntimeFeature.cs
View file

@ -165,7 +165,7 @@ public class WorkflowRuntimeFeature : IShellFeature
sp.GetRequiredService<IOptions<GracefulShutdownOptions>>(),
sp.GetRequiredService<ISystemClock>(),
sp.GetRequiredService<IExecutionCycleRegistry>(),
sp.GetService<KeyValues.Contracts.IKeyValueStore>(),
sp.GetRequiredService<IServiceScopeFactory>(),
shellName: sp.GetService<CShells.ShellSettings>()?.Id))
.AddSingleton<IIngressSourceRegistry, IngressSourceRegistry>()
.AddSingleton<IExecutionCycleRegistry, ExecutionCycleRegistry>()

23
test/unit/Elsa.Common.UnitTests/Multitenancy/TenantTaskLifecycleCoordinatorTests.cs
View file

@ -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 ===

38
test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/QuiescenceSignalPersistenceTests.cs
View file

@ -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<GracefulShutdownOptions>().Configure(options => options.PausePersistence = PausePersistencePolicy.AcrossReactivations);
services.AddSingleton(_clock);
services.AddSingleton(_cycleRegistry);
services.AddScoped<IKeyValueStore>(_ => _kv);
services.AddSingleton<IQuiescenceSignal, QuiescenceSignal>();
await using var provider = services.BuildServiceProvider(new ServiceProviderOptions { ValidateScopes = true, ValidateOnBuild = true });
var sut = provider.GetRequiredService<IQuiescenceSignal>();
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<string, SerializedKeyValuePair> Pairs = new(StringComparer.Ordinal);