elsa-core/test/unit/Elsa.Workflows.Runtime.UnitTests/Quiescence/ExecutionCycleRegistryTests.cs

using Elsa.Common;
using Elsa.Workflows.Runtime.Services;
using NSubstitute;

namespace Elsa.Workflows.Runtime.UnitTests.Quiescence;

public class ExecutionCycleRegistryTests
{
    private readonly ISystemClock _clock;
    private readonly IIngressSourceRegistry _sources;

    public ExecutionCycleRegistryTests()
    {
        _clock = Substitute.For<ISystemClock>();
        _clock.UtcNow.Returns(DateTimeOffset.Parse("2026-04-24T10:00:00Z"));
        _sources = Substitute.For<IIngressSourceRegistry>();
        _sources.Snapshot().Returns(Array.Empty<IngressSourceSnapshot>());
    }

    [Fact(DisplayName = "Active count increases and decreases with begin/dispose")]
    public void ActiveCountFollowsExecutionCycleLifecycle()
    {
        var sut = new ExecutionCycleRegistry(_sources, _clock);

        Assert.Equal(0, sut.ActiveCount);

        var a = sut.BeginCycle("instance-1", ingressSourceName: null, CancellationToken.None);
        Assert.Equal(1, sut.ActiveCount);

        var b = sut.BeginCycle("instance-2", ingressSourceName: null, CancellationToken.None);
        Assert.Equal(2, sut.ActiveCount);

        a.Dispose();
        Assert.Equal(1, sut.ActiveCount);

        b.Dispose();
        Assert.Equal(0, sut.ActiveCount);
    }

    [Fact(DisplayName = "Begin with null ingress name does NOT flip any source")]
    public void NullIngressNameDoesNotFlip()
    {
        var sut = new ExecutionCycleRegistry(_sources, _clock);

        using var _ = sut.BeginCycle("instance-1", ingressSourceName: null, CancellationToken.None);

        _sources.DidNotReceive().MarkPauseFailedAsync(Arg.Any<string>(), Arg.Any<string>(), Arg.Any<Exception?>());
    }

    [Fact(DisplayName = "Begin from a Paused source flips it to PauseFailed (FR-018)")]
    public void PausedSourceDeliveringIsFlipped()
    {
        var now = _clock.UtcNow;
        var snapshot = new[] { new IngressSourceSnapshot("http.trigger", IngressSourceState.Paused, null, now) };
        _sources.Snapshot().Returns(snapshot);
        var sut = new ExecutionCycleRegistry(_sources, _clock);

        using var _ = sut.BeginCycle("instance-1", ingressSourceName: "http.trigger", CancellationToken.None);

        _sources.Received(1).MarkPauseFailedAsync("http.trigger", "delivered-while-paused", Arg.Any<Exception?>());
    }

    [Fact(DisplayName = "Begin from a Running source does NOT flip")]
    public void RunningSourceIsNotFlipped()
    {
        var now = _clock.UtcNow;
        var snapshot = new[] { new IngressSourceSnapshot("http.trigger", IngressSourceState.Running, null, now) };
        _sources.Snapshot().Returns(snapshot);
        var sut = new ExecutionCycleRegistry(_sources, _clock);

        using var _ = sut.BeginCycle("instance-1", ingressSourceName: "http.trigger", CancellationToken.None);

        _sources.DidNotReceive().MarkPauseFailedAsync(Arg.Any<string>(), Arg.Any<string>(), Arg.Any<Exception?>());
    }

    [Fact(DisplayName = "ListActiveCycles returns a snapshot of live handles")]
    public void ListActiveCyclesReturnsSnapshot()
    {
        var sut = new ExecutionCycleRegistry(_sources, _clock);
        var a = sut.BeginCycle("instance-1", null, CancellationToken.None);
        var b = sut.BeginCycle("instance-2", null, CancellationToken.None);

        var snapshot = sut.ListActiveCycles();

        Assert.Equal(2, snapshot.Count);
        Assert.Contains(a, snapshot);
        Assert.Contains(b, snapshot);

        a.Dispose();
        b.Dispose();
    }

    [Fact(DisplayName = "ExecutionCycleHandle.Cancel triggers the cancellation token")]
    public void ExecutionCycleHandleCancelFiresToken()
    {
        var sut = new ExecutionCycleRegistry(_sources, _clock);
        using var handle = sut.BeginCycle("instance-1", null, CancellationToken.None);

        Assert.False(handle.CancellationToken.IsCancellationRequested);
        handle.Cancel();
        Assert.True(handle.CancellationToken.IsCancellationRequested);
    }

    [Fact(DisplayName = "ExecutionCycleHandle.TryCancel reports whether this call transitioned the handle")]
    public void ExecutionCycleHandleTryCancelReportsTransition()
    {
        var sut = new ExecutionCycleRegistry(_sources, _clock);
        var handle = sut.BeginCycle("instance-1", null, CancellationToken.None);

        Assert.True(handle.TryCancel());
        Assert.False(handle.TryCancel());

        handle.Dispose();
        Assert.False(handle.TryCancel());

        var disposed = sut.BeginCycle("instance-2", null, CancellationToken.None);
        disposed.Dispose();
        Assert.False(disposed.TryCancel());
    }

    [Fact(DisplayName = "ExecutionCycleHandle.TryCancel reports false when disposed during the cancellation callback")]
    public async Task TryCancelReportsFalseWhenDisposedDuringCancellationCallback()
    {
        var sut = new ExecutionCycleRegistry(_sources, _clock);
        var callbackEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        var releaseCallback = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        var handle = sut.BeginCycle(
            "instance-1",
            ingressSourceName: null,
            linkedToken: CancellationToken.None,
            cancelCallback: () =>
            {
                callbackEntered.SetResult();
                releaseCallback.Task.GetAwaiter().GetResult();
            });

        var cancelTask = Task.Run(handle.TryCancel);
        await callbackEntered.Task.WaitAsync(TimeSpan.FromSeconds(5));

        handle.Dispose();
        releaseCallback.SetResult();

        Assert.False(await cancelTask.WaitAsync(TimeSpan.FromSeconds(5)));
    }

    [Fact(DisplayName = "ExecutionCycleHandle.Dispose completes while a CTS callback waits for it")]
    public async Task DisposeCompletesWhileCtsCallbackWaitsForIt()
    {
        var cancellationCallbackEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        var disposalCompleted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        var callbackObservedDisposal = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
        var handle = new ExecutionCycleHandle(
            Guid.NewGuid(),
            "instance-1",
            ingressSourceName: null,
            startedAt: DateTimeOffset.UtcNow,
            linkedToken: CancellationToken.None,
            onDisposed: null);
        using var registration = handle.CancellationToken.Register(() =>
        {
            cancellationCallbackEntered.SetResult();
            callbackObservedDisposal.SetResult(disposalCompleted.Task.Wait(TimeSpan.FromSeconds(5)));
        });

        var cancelTask = Task.Run(handle.TryCancel);
        await cancellationCallbackEntered.Task.WaitAsync(TimeSpan.FromSeconds(5));

        var disposeTask = Task.Run(() =>
        {
            handle.Dispose();
            disposalCompleted.TrySetResult();
        });

        Assert.True(await callbackObservedDisposal.Task.WaitAsync(TimeSpan.FromSeconds(5)));
        await disposeTask.WaitAsync(TimeSpan.FromSeconds(5));
        Assert.False(await cancelTask.WaitAsync(TimeSpan.FromSeconds(5)));
        Assert.True(handle.Disposed.IsCompletedSuccessfully);
    }

    [Fact(DisplayName = "ExecutionCycleHandle defers CTS disposal while linked-token cancellation is in progress")]
    public async Task DisposeCompletesWhileLinkedTokenCancellationWaitsForIt()
    {
        using var linkedCts = new CancellationTokenSource();
        var cancellationCallbackEntered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        var disposalCompleted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
        var callbackObservedDisposal = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
        using var handle = new ExecutionCycleHandle(
            Guid.NewGuid(),
            "instance-linked-cancellation",
            ingressSourceName: null,
            startedAt: DateTimeOffset.UtcNow,
            linkedToken: linkedCts.Token);
        var cycleToken = handle.CancellationToken;
        using var registration = cycleToken.Register(() =>
        {
            cancellationCallbackEntered.SetResult();
            callbackObservedDisposal.SetResult(disposalCompleted.Task.Wait(TimeSpan.FromSeconds(5)));
        });

        var cancelTask = Task.Run(() => linkedCts.Cancel());
        await cancellationCallbackEntered.Task.WaitAsync(TimeSpan.FromSeconds(5));

        var disposeTask = Task.Run(() =>
        {
            handle.Dispose();
            disposalCompleted.TrySetResult();
        });

        Assert.True(await callbackObservedDisposal.Task.WaitAsync(TimeSpan.FromSeconds(5)));
        await disposeTask.WaitAsync(TimeSpan.FromSeconds(5));
        await cancelTask.WaitAsync(TimeSpan.FromSeconds(5));
        Assert.True(handle.Disposed.IsCompletedSuccessfully);
        Assert.True(cycleToken.IsCancellationRequested);
    }

    [Fact(DisplayName = "ExecutionCycleHandle defers CTS disposal from a cancellation callback")]
    public void DisposeDefersCtsDisposalUntilCancellationPropagationExits()
    {
        var disposedDuringCancellation = false;
        var handle = new ExecutionCycleHandle(
            Guid.NewGuid(),
            "instance-1",
            ingressSourceName: null,
            startedAt: DateTimeOffset.UtcNow,
            linkedToken: CancellationToken.None);

        using var registration = handle.CancellationToken.Register(() =>
        {
            handle.Dispose();
            disposedDuringCancellation = handle.Disposed.IsCompleted;
        });

        Assert.False(handle.TryCancel());
        Assert.False(disposedDuringCancellation);
        Assert.True(handle.Disposed.IsCompletedSuccessfully);
    }

    [Fact(DisplayName = "ExecutionCycleHandle.Cancel invokes the cancel callback supplied at registration")]
    public void CancelCallbackIsInvoked()
    {
        var sut = new ExecutionCycleRegistry(_sources, _clock);
        var callbackInvocations = 0;
        using var handle = sut.BeginCycle(
            "instance-1",
            ingressSourceName: null,
            linkedToken: CancellationToken.None,
            cancelCallback: () => Interlocked.Increment(ref callbackInvocations));

        handle.Cancel();
        Assert.Equal(1, callbackInvocations);

        // Truly idempotent: a second Cancel() before Dispose() must NOT re-invoke the callback. The handle uses an
        // Interlocked lifecycle state so callers can't accidentally trigger non-idempotent cancellation side effects.
        handle.Cancel();
        Assert.Equal(1, callbackInvocations);

        // Once disposed, further Cancel() invocations remain silent no-ops.
        handle.Dispose();
        handle.Cancel();
        Assert.Equal(1, callbackInvocations);
    }

    [Fact(DisplayName = "ExecutionCycleHandle.Cancel swallows callback exceptions so drain is not interrupted")]
    public void CancelCallbackExceptionsAreSwallowed()
    {
        var sut = new ExecutionCycleRegistry(_sources, _clock);
        using var handle = sut.BeginCycle(
            "instance-1",
            ingressSourceName: null,
            linkedToken: CancellationToken.None,
            cancelCallback: () => throw new InvalidOperationException("activity refused to cancel"));

        // Should not throw — Cancel() must remain best-effort so a single misbehaving workflow does not crash drain.
        handle.Cancel();
        Assert.True(handle.CancellationToken.IsCancellationRequested);
    }

    [Fact(DisplayName = "ExecutionCycleHandle.TryCancel swallows non-fatal CTS callback exceptions")]
    public void TryCancelSwallowsNonFatalCtsCallbackExceptions()
    {
        var handle = new ExecutionCycleHandle(
            Guid.NewGuid(),
            "instance-1",
            ingressSourceName: null,
            startedAt: DateTimeOffset.UtcNow,
            linkedToken: CancellationToken.None);
        using var registration = handle.CancellationToken.Register(() => throw new InvalidOperationException("callback refused to cancel"));

        Assert.True(handle.TryCancel());
        Assert.False(handle.TryCancel());

        handle.Dispose();
        Assert.True(handle.Disposed.IsCompletedSuccessfully);
    }

    [Fact(DisplayName = "ExecutionCycleHandle.TryCancel propagates fatal CTS callback exceptions wrapped in an aggregate")]
    public void TryCancelPropagatesFatalCtsCallbackExceptions()
    {
        var handle = new ExecutionCycleHandle(
            Guid.NewGuid(),
            "instance-1",
            ingressSourceName: null,
            startedAt: DateTimeOffset.UtcNow,
            linkedToken: CancellationToken.None);
        using var registration = handle.CancellationToken.Register(() => throw new OutOfMemoryException("fatal callback failure"));

        var exception = Assert.Throws<AggregateException>(() => handle.TryCancel());

        Assert.Contains(exception.Flatten().InnerExceptions, inner => inner is OutOfMemoryException);
        handle.Dispose();
        Assert.True(handle.Disposed.IsCompletedSuccessfully);
    }

    [Fact(DisplayName = "ExecutionCycleHandle.Disposed completes when the handle is disposed")]
    public async Task DisposedTaskCompletesOnDispose()
    {
        var sut = new ExecutionCycleRegistry(_sources, _clock);
        var handle = sut.BeginCycle("instance-1", null, CancellationToken.None);

        Assert.False(handle.Disposed.IsCompleted);
        handle.Dispose();
        await handle.Disposed.WaitAsync(TimeSpan.FromSeconds(1));
        Assert.True(handle.Disposed.IsCompletedSuccessfully);
    }
}