using Elsa.Common;
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;
///
/// Default thread-safe implementation of . Uses a single lock for transitions
/// and a volatile reference read for lock-free state queries. See FR-001..FR-005 and research R8 for
/// pause-persistence semantics.
///
public sealed class QuiescenceSignal : IQuiescenceSignal
{
private const string PersistenceKeyPrefix = "elsa.quiescence.pause.";
private readonly object _sync = new();
// Serializes persistence I/O so racing Pause/Resume can't reorder writes in the store. Held only across
// the I/O — the in-memory transition still uses the fast _sync lock, and pause/resume aren't hot paths.
private readonly SemaphoreSlim _persistenceMutex = new(1, 1);
private readonly IOptions _options;
private readonly ISystemClock _clock;
private readonly IKeyValueStore? _keyValueStore;
private readonly IServiceScopeFactory? _serviceScopeFactory;
private readonly IExecutionCycleRegistry _cycleRegistry;
private readonly string _persistenceKey;
private QuiescenceState _state;
///
/// Creates the signal. The generation id defaults to a new GUID per construction — when the container is torn
/// 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) => 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"));
}
///
public QuiescenceState CurrentState
{
get
{
// Volatile read — the reference is always overwritten atomically under the lock.
return Volatile.Read(ref _state);
}
}
///
public bool IsAcceptingNewWork => CurrentState.IsAcceptingNewWork;
///
public int ActiveExecutionCycleCount => _cycleRegistry.ActiveCount;
///
/// Loads any persisted administrative pause state. Called once per runtime generation by a startup task when
/// is .
/// No-op otherwise, or when the key-value store is not registered.
///
public async ValueTask InitializePersistedStateAsync(CancellationToken cancellationToken)
{
if (_options.Value.PausePersistence != PausePersistencePolicy.AcrossReactivations) return;
var pair = await UseKeyValueStoreAsync(store => store.FindAsync(new KeyValueFilter { Key = _persistenceKey }, cancellationToken), defaultValue: (SerializedKeyValuePair?)null);
if (pair is null) return;
lock (_sync)
{
if ((_state.Reason & QuiescenceReason.AdministrativePause) != 0) return; // someone already paused us
var next = _state with
{
Reason = _state.Reason | QuiescenceReason.AdministrativePause,
PausedAt = _clock.UtcNow,
PauseReasonText = pair.SerializedValue,
PauseRequestedBy = "persisted",
};
Volatile.Write(ref _state, next);
}
}
///
public ValueTask BeginDrainAsync(CancellationToken cancellationToken = default)
{
QuiescenceState next;
lock (_sync)
{
if ((_state.Reason & QuiescenceReason.Drain) != 0)
{
return new ValueTask(_state);
}
next = _state with
{
Reason = _state.Reason | QuiescenceReason.Drain,
DrainStartedAt = _clock.UtcNow,
};
Volatile.Write(ref _state, next);
}
return new ValueTask(next);
}
///
public async ValueTask PauseAsync(string? reasonText, string? requestedBy, CancellationToken cancellationToken)
{
QuiescenceState next;
bool transitioned = false;
lock (_sync)
{
if ((_state.Reason & QuiescenceReason.AdministrativePause) != 0)
{
next = _state;
}
else
{
next = _state with
{
Reason = _state.Reason | QuiescenceReason.AdministrativePause,
PausedAt = _clock.UtcNow,
PauseReasonText = reasonText,
PauseRequestedBy = requestedBy,
};
Volatile.Write(ref _state, next);
transitioned = true;
}
}
if (transitioned)
await PersistAsync();
return next;
}
///
public async ValueTask ResumeAsync(string? requestedBy, CancellationToken cancellationToken)
{
QuiescenceState next;
bool transitioned = false;
lock (_sync)
{
// Resume is a no-op while drain is active — the runtime cannot return to normal operation within the same generation.
if ((_state.Reason & QuiescenceReason.Drain) != 0) { return _state; }
if ((_state.Reason & QuiescenceReason.AdministrativePause) == 0) { return _state; }
next = _state with
{
Reason = _state.Reason & ~QuiescenceReason.AdministrativePause,
PausedAt = null,
PauseReasonText = null,
PauseRequestedBy = requestedBy,
};
Volatile.Write(ref _state, next);
transitioned = true;
}
if (transitioned)
await PersistAsync();
return next;
}
///
/// Persists the current administrative-pause state. Serialized via so racing
/// Pause/Resume can't reorder writes in the store. The live state is re-read inside the semaphore — each I/O
/// writes whatever the latest in-memory transition was, so N racing transitions produce N serialized writes
/// and the final persisted state always matches the final in-memory state.
///
///
/// Uses deliberately for both the semaphore wait and the store I/O.
/// By the time this runs the in-memory transition has already committed; if a cancelled HTTP request token
/// caused the persistence to skip, in-memory state would diverge from the store — and the idempotent
/// fast-path in / (transitioned == false) means a later call
/// would not retry the write. So persistence must complete regardless of caller cancellation.
///
private async ValueTask PersistAsync()
{
if (_options.Value.PausePersistence != PausePersistencePolicy.AcrossReactivations)
return;
await _persistenceMutex.WaitAsync(CancellationToken.None);
try
{
var live = Volatile.Read(ref _state);
if ((live.Reason & QuiescenceReason.AdministrativePause) != 0)
await UseKeyValueStoreAsync(store => store.SaveAsync(new SerializedKeyValuePair { Key = _persistenceKey, SerializedValue = live.PauseReasonText ?? string.Empty }, CancellationToken.None));
else
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);
}
}