fix(runtime): bound concurrent pre-cancel snapshot Finds during drain (#8113)
* fix(runtime): bound concurrent pre-cancel snapshot Finds during drain Unbounded Task.WhenAll of per-cycle Finds self-contends under large live-cycle N: more 250ms timeouts, more drainInduced excludes, more Interrupted misses. Cap snapshot Finds at 16. The 250ms budget still starts only after a slot is acquired so queued Finds are not fail-open excluded by waiting. Phase C stays sequential. No store-contract change. Closes #8083 Co-authored-by: Sipke Schoorstra <sipkeschoorstra@outlook.com> * fix(runtime): exclude null pre-cancel snapshots from drainInduced A successful Find that returns no row is unknown pre-state, not a confirmed non-Cancelled snapshot. Joining drainInduced let Phase C rewrite a later Finished/Cancelled as Interrupted. Timeout/error already excluded; null now does too. Closes nothing extra; keeps #8083 fail-open exclude. Co-authored-by: Sipke Schoorstra <sipkeschoorstra@outlook.com> * fix(runtime): promote null snapshot after Phase A force-cancel A successful pre-cancel Find that returns no row is not a persisted user-cancel, but excluding it from drainInduced skipped Interrupted persist after deadline-breach force-cancel of a live cycle (DeadlineBreachPersistsInterrupted). Join drainInduced only after we ourselves cancel that handle. Timeout/error and confirmed Cancelled snapshots stay excluded. Co-authored-by: Sipke Schoorstra <sipkeschoorstra@outlook.com> * fix(runtime): promote null snapshot only when TryCancel transitions Cancel() is a no-op on an already-disposed handle, so treating every Cancel() call as drain-induced could rewrite a Finished/Cancelled row the runner committed while snapshot was in flight. TryCancel reports a real transition; only those ids join drainInduced after a null Find. Co-authored-by: Sipke Schoorstra <sipkeschoorstra@outlook.com> --------- Co-authored-by: Cursor Agent <cursoragent@cursor.com>
This commit is contained in:
Sipke Schoorstra
GitHub
parent
4b15b0166d
commit
f477fc8b07
|
|
@ -74,13 +74,27 @@ public sealed class ExecutionCycleHandle : IDisposable
|
|||
/// Invokes the cancel callback (when supplied at construction) to propagate cancellation into the workflow
|
||||
/// execution, then cancels the cycle's own linked CTS. Safe to call multiple times; idempotent.
|
||||
/// </summary>
|
||||
public void Cancel()
|
||||
public void Cancel() => TryCancel();
|
||||
|
||||
/// <summary>
|
||||
/// Attempts to cancel the cycle. Returns <c>true</c> only when this call transitioned the handle from
|
||||
/// not-cancelled to cancelled. Returns <c>false</c> when the handle was already disposed or already cancelled,
|
||||
/// so drain can avoid treating a finished cycle as a force-cancel.
|
||||
/// </summary>
|
||||
public bool TryCancel()
|
||||
{
|
||||
if (_disposed != 0) return;
|
||||
if (Volatile.Read(ref _disposed) != 0)
|
||||
return false;
|
||||
|
||||
// Idempotent guard: ensures the cancel callback and CTS cancellation run AT MOST once even if Cancel() is
|
||||
// called repeatedly before disposal. Without this the drain orchestrator (or any other future caller) could
|
||||
// accidentally trigger a non-idempotent cancellation side effect multiple times.
|
||||
if (Interlocked.Exchange(ref _cancelled, 1) != 0) return;
|
||||
if (Interlocked.Exchange(ref _cancelled, 1) != 0)
|
||||
return false;
|
||||
|
||||
// Disposed after we claimed cancel: the cycle already finished; do not treat as our force-cancel.
|
||||
if (Volatile.Read(ref _disposed) != 0)
|
||||
return false;
|
||||
|
||||
// Propagate to the workflow execution first (this typically marks the workflow as Cancelled and clears its
|
||||
// schedule, so the runner stops scheduling new activities). The orchestrator's subsequent Interrupted
|
||||
|
|
@ -90,6 +104,8 @@ public sealed class ExecutionCycleHandle : IDisposable
|
|||
|
||||
try { _cycleCts.Cancel(); }
|
||||
catch (ObjectDisposedException) { /* Race with Dispose — acceptable. */ }
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
/// <summary>Releases the linked CTS, notifies the registry, and signals <see cref="Disposed"/>.</summary>
|
||||
|
|
|
|||
|
|
@ -1,6 +1,8 @@
|
|||
using System.Collections.Concurrent;
|
||||
using System.Diagnostics;
|
||||
using Elsa.Common;
|
||||
using Elsa.Workflows.Management;
|
||||
using Elsa.Workflows.Management.Entities;
|
||||
using Elsa.Workflows.Management.Filters;
|
||||
using Elsa.Workflows.Runtime.Options;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
|
|
@ -282,6 +284,22 @@ public sealed class DrainOrchestrator : IDrainOrchestrator
|
|||
/// </summary>
|
||||
private static readonly TimeSpan PersistInterruptedTimeout = TimeSpan.FromSeconds(5);
|
||||
|
||||
/// <summary>
|
||||
/// Short shutdown budget for the pre-cancel instance snapshot. Force-cancel runs after the drain
|
||||
/// deadline, so remaining deadline is already zero. WaitAsync unblocks even if the store ignores
|
||||
/// its cancellation token. The budget starts when a snapshot slot is acquired — not while
|
||||
/// waiting for <see cref="MaxConcurrentPreCancelSnapshotFinds"/>.
|
||||
/// </summary>
|
||||
private static readonly TimeSpan PreCancelSnapshotTimeout = TimeSpan.FromMilliseconds(250);
|
||||
|
||||
/// <summary>
|
||||
/// Cap on concurrent pre-cancel snapshot Finds. Unbounded <c>Task.WhenAll</c> self-contends
|
||||
/// under large live-cycle N: more 250ms timeouts, more <c>drainInduced</c> excludes, more
|
||||
/// Interrupted misses. Queue wait uses the drain token only; each Find still gets its own
|
||||
/// 250ms after it acquires a slot.
|
||||
/// </summary>
|
||||
internal const int MaxConcurrentPreCancelSnapshotFinds = 16;
|
||||
|
||||
private async Task<(int Count, IReadOnlyList<string> Ids)> ForceCancelActiveCyclesAsync(DrainTrigger trigger, string generationId, CancellationToken cancellationToken)
|
||||
{
|
||||
var cap = _options.Value.MaxForceCancelledInstanceIdsReported;
|
||||
|
|
@ -299,6 +317,64 @@ public sealed class DrainOrchestrator : IDrainOrchestrator
|
|||
? WorkflowInterruptedPayload.ReasonOperatorForce
|
||||
: WorkflowInterruptedPayload.ReasonDeadlineBreach;
|
||||
|
||||
// Snapshot before Phase A. Each Find has its own 250ms budget and its own
|
||||
// DI scope so concurrent reads do not share an EF DbContext (Phase C stays
|
||||
// sequential on the outer scope for the same reason). Finds run concurrently
|
||||
// under a semaphore so a large live-cycle set cannot self-contend into
|
||||
// 250ms timeouts. Timeout/error: exclude that id (prefer preserving
|
||||
// user-cancel / #8052). A successful null Find is not drain-induced yet —
|
||||
// no persisted user-cancel exists, but we only promote if Phase A actually
|
||||
// cancels that live handle (DeadlineBreachPersistsInterrupted).
|
||||
var drainInducedInstanceIds = new HashSet<string>(StringComparer.Ordinal);
|
||||
var missingPersistedRowIds = new ConcurrentBag<string>();
|
||||
using var snapshotGate = new SemaphoreSlim(MaxConcurrentPreCancelSnapshotFinds);
|
||||
var snapshotTasks = live.Select(async handle =>
|
||||
{
|
||||
try
|
||||
{
|
||||
await snapshotGate.WaitAsync(cancellationToken).ConfigureAwait(false);
|
||||
try
|
||||
{
|
||||
using var findScope = _scopeFactory.CreateScope();
|
||||
var findStore = findScope.ServiceProvider.GetRequiredService<IWorkflowInstanceStore>();
|
||||
using var perFindCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
|
||||
perFindCts.CancelAfter(PreCancelSnapshotTimeout);
|
||||
var snapshot = await findStore
|
||||
.FindAsync(new WorkflowInstanceFilter { Id = handle.WorkflowInstanceId }, perFindCts.Token)
|
||||
.AsTask()
|
||||
.WaitAsync(perFindCts.Token)
|
||||
.ConfigureAwait(false);
|
||||
if (snapshot is null)
|
||||
{
|
||||
missingPersistedRowIds.Add(handle.WorkflowInstanceId);
|
||||
return null;
|
||||
}
|
||||
|
||||
if (snapshot.SubStatus != WorkflowSubStatus.Cancelled)
|
||||
return handle.WorkflowInstanceId;
|
||||
}
|
||||
finally
|
||||
{
|
||||
// Release after the await budget, not after the store call returns.
|
||||
// A cancel-ignoring Find can outlive this slot (documented secondary;
|
||||
// holding the slot until it completes would stall WhenAll / Phase A).
|
||||
snapshotGate.Release();
|
||||
}
|
||||
}
|
||||
catch (Exception ex) when (!ex.IsFatal())
|
||||
{
|
||||
_logger.LogWarning(ex, "Pre-cancel snapshot for instance {InstanceId} timed out or failed; excluding from drain-induced promote.", handle.WorkflowInstanceId);
|
||||
}
|
||||
|
||||
return null;
|
||||
});
|
||||
|
||||
foreach (var instanceId in await Task.WhenAll(snapshotTasks).ConfigureAwait(false))
|
||||
{
|
||||
if (instanceId is not null)
|
||||
drainInducedInstanceIds.Add(instanceId);
|
||||
}
|
||||
|
||||
// Force-cancel proceeds in three phases. The split exists because cancelling and
|
||||
// awaiting in the same loop made every execution cycle after the first run at full speed
|
||||
// through the prior execution cycle's settle window — total wall time was O(N × settle
|
||||
|
|
@ -307,12 +383,19 @@ public sealed class DrainOrchestrator : IDrainOrchestrator
|
|||
// Phase A — cancel every handle synchronously. CancellationTokenSource.Cancel is
|
||||
// cheap and we want every runner to observe cancellation simultaneously rather
|
||||
// than serialized behind preceding settle waits.
|
||||
var cancelledInstanceIds = new HashSet<string>(StringComparer.Ordinal);
|
||||
foreach (var handle in live)
|
||||
{
|
||||
try
|
||||
{
|
||||
handle.Cancel();
|
||||
// Only a real transition counts. Cancel() is a no-op on an already-disposed handle
|
||||
// (cycle finished during snapshot); treating that as drain-induced would rewrite a
|
||||
// later Finished/Cancelled row the runner already committed.
|
||||
if (!handle.TryCancel())
|
||||
continue;
|
||||
|
||||
totalCancelled++;
|
||||
cancelledInstanceIds.Add(handle.WorkflowInstanceId);
|
||||
if (reportedIds.Count < cap) reportedIds.Add(handle.WorkflowInstanceId);
|
||||
}
|
||||
catch (Exception ex) when (!ex.IsFatal())
|
||||
|
|
@ -321,6 +404,15 @@ public sealed class DrainOrchestrator : IDrainOrchestrator
|
|||
}
|
||||
}
|
||||
|
||||
// A live handle we ourselves cancelled whose snapshot found no row is drain-induced:
|
||||
// there was no persisted user-cancel to preserve. Timeout/error and disposed no-ops stay excluded.
|
||||
// Do not use reportedIds here — that list is capped by MaxForceCancelledInstanceIdsReported.
|
||||
foreach (var instanceId in missingPersistedRowIds)
|
||||
{
|
||||
if (cancelledInstanceIds.Contains(instanceId))
|
||||
drainInducedInstanceIds.Add(instanceId);
|
||||
}
|
||||
|
||||
// Phase B — wait for every runner to settle in parallel under a shared deadline.
|
||||
// The handle disposes when ExecutionCycleTrackingMiddleware exits its `using` block, which
|
||||
// is after the workflow runner has finished its commit. We want runners' terminal
|
||||
|
|
@ -358,7 +450,7 @@ public sealed class DrainOrchestrator : IDrainOrchestrator
|
|||
try
|
||||
{
|
||||
using var persistCts = new CancellationTokenSource(PersistInterruptedTimeout);
|
||||
await PersistInterruptedAsync(instanceStore, logStore, handle, generationId, reason, persistCts.Token);
|
||||
await PersistInterruptedAsync(instanceStore, logStore, handle, generationId, reason, drainInducedInstanceIds, persistCts.Token);
|
||||
}
|
||||
catch (Exception ex) when (!ex.IsFatal())
|
||||
{
|
||||
|
|
@ -375,6 +467,7 @@ public sealed class DrainOrchestrator : IDrainOrchestrator
|
|||
ExecutionCycleHandle handle,
|
||||
string generationId,
|
||||
string reason,
|
||||
HashSet<string> drainInducedInstanceIds,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var instance = await instanceStore.FindAsync(new WorkflowInstanceFilter { Id = handle.WorkflowInstanceId }, cancellationToken);
|
||||
|
|
@ -420,6 +513,14 @@ public sealed class DrainOrchestrator : IDrainOrchestrator
|
|||
return;
|
||||
}
|
||||
|
||||
if (ShouldSkipInterruptedPersist(instance, drainInducedInstanceIds))
|
||||
{
|
||||
_logger.LogInformation(
|
||||
"Skipping Interrupted persist for instance {InstanceId}: already in terminal status {Status}/{SubStatus}.",
|
||||
instance.Id, instance.Status, instance.SubStatus);
|
||||
return;
|
||||
}
|
||||
|
||||
var actualReason = reason;
|
||||
|
||||
try
|
||||
|
|
@ -459,4 +560,21 @@ public sealed class DrainOrchestrator : IDrainOrchestrator
|
|||
.Select(s => new IngressSourceFinalState(s.Name, s.State, s.LastError, WasForceStopped: s.State == IngressSourceState.PauseFailed && s.LastError is not null))
|
||||
.ToArray();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Skip persist when the row is already a real terminal outcome: natural completion,
|
||||
/// fault, already Cancelled, or unknown pre-state (snapshot timeout/error). Ids that
|
||||
/// snapshot showed were not Cancelled, or that had no row and that we ourselves
|
||||
/// force-cancelled, may be promoted.
|
||||
/// </summary>
|
||||
private static bool ShouldSkipInterruptedPersist(WorkflowInstance instance, HashSet<string> drainInducedInstanceIds)
|
||||
{
|
||||
if (instance.Status != WorkflowStatus.Finished)
|
||||
return false;
|
||||
|
||||
if (instance.SubStatus == WorkflowSubStatus.Cancelled)
|
||||
return !drainInducedInstanceIds.Contains(instance.Id);
|
||||
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
using Elsa.Workflows.Management.Entities;
|
||||
using Elsa.Workflows.Management.Filters;
|
||||
using Elsa.Workflows.Runtime.HostedServices;
|
||||
using Elsa.Workflows.Runtime.Services;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using NSubstitute;
|
||||
|
||||
|
|
@ -78,6 +79,225 @@ public class DrainOrchestratorWaitTests : DrainOrchestratorTestsBase
|
|||
Arg.Any<CancellationToken>());
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "Pre-cancel snapshot Finds never exceed MaxConcurrentPreCancelSnapshotFinds in flight")]
|
||||
public async Task SnapshotFindsAreCapped()
|
||||
{
|
||||
const int extra = 8;
|
||||
var cap = DrainOrchestrator.MaxConcurrentPreCancelSnapshotFinds;
|
||||
var count = cap + extra;
|
||||
var handles = Enumerable.Range(0, count)
|
||||
.Select(i => new ExecutionCycleHandle(Guid.NewGuid(), $"instance-capped-{i}", ingressSourceName: "http.trigger", startedAt: DateTimeOffset.UtcNow, linkedToken: CancellationToken.None))
|
||||
.ToArray();
|
||||
ExecutionCycleRegistry.ActiveCount.Returns(count);
|
||||
ExecutionCycleRegistry.ListActiveCycles().Returns(handles);
|
||||
|
||||
var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
var capReached = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
var seen = new HashSet<string>(StringComparer.Ordinal);
|
||||
var inFlight = 0;
|
||||
var maxInFlight = 0;
|
||||
var maxLock = new object();
|
||||
|
||||
InstanceStore.FindAsync(Arg.Any<WorkflowInstanceFilter>(), Arg.Any<CancellationToken>())
|
||||
.Returns(ci =>
|
||||
{
|
||||
var id = ci.Arg<WorkflowInstanceFilter>().Id!;
|
||||
lock (seen)
|
||||
{
|
||||
if (!seen.Add(id))
|
||||
return new ValueTask<WorkflowInstance?>(RunningInstance(id));
|
||||
}
|
||||
|
||||
return new ValueTask<WorkflowInstance?>(WaitForSnapshotSlotAsync(id));
|
||||
});
|
||||
|
||||
async Task<WorkflowInstance?> WaitForSnapshotSlotAsync(string id)
|
||||
{
|
||||
var current = Interlocked.Increment(ref inFlight);
|
||||
lock (maxLock)
|
||||
{
|
||||
if (current > maxInFlight)
|
||||
maxInFlight = current;
|
||||
if (maxInFlight >= cap)
|
||||
capReached.TrySetResult();
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
await release.Task;
|
||||
return RunningInstance(id);
|
||||
}
|
||||
finally
|
||||
{
|
||||
Interlocked.Decrement(ref inFlight);
|
||||
}
|
||||
}
|
||||
|
||||
var sut = BuildSut();
|
||||
var drainTask = sut.DrainAsync(DrainTrigger.OperatorForce).AsTask();
|
||||
|
||||
await capReached.Task.WaitAsync(TimeSpan.FromSeconds(5));
|
||||
await Task.Delay(50);
|
||||
Assert.Equal(cap, maxInFlight);
|
||||
|
||||
release.TrySetResult();
|
||||
|
||||
var outcome = await drainTask.WaitAsync(TimeSpan.FromSeconds(8));
|
||||
Assert.Equal(DrainResult.Forced, outcome.OverallResult);
|
||||
Assert.Equal(count, outcome.ExecutionCyclesForceCancelledCount);
|
||||
foreach (var handle in handles)
|
||||
await InstanceStore.Received().SaveAsync(Arg.Is<WorkflowInstance>(i => i.Id == handle.WorkflowInstanceId && i.SubStatus == WorkflowSubStatus.Interrupted), Arg.Any<CancellationToken>());
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "A null pre-cancel snapshot still promotes a later drain-induced Finished/Cancelled row")]
|
||||
public async Task NullSnapshotPromotesLaterCancelledInstance()
|
||||
{
|
||||
var handle = new ExecutionCycleHandle(Guid.NewGuid(), "instance-missing", ingressSourceName: "http.trigger", startedAt: DateTimeOffset.UtcNow, linkedToken: CancellationToken.None);
|
||||
ExecutionCycleRegistry.ActiveCount.Returns(1);
|
||||
ExecutionCycleRegistry.ListActiveCycles().Returns(new[] { handle });
|
||||
|
||||
var finds = 0;
|
||||
InstanceStore.FindAsync(Arg.Any<WorkflowInstanceFilter>(), Arg.Any<CancellationToken>())
|
||||
.Returns(_ =>
|
||||
{
|
||||
if (Interlocked.Increment(ref finds) == 1)
|
||||
return new ValueTask<WorkflowInstance?>((WorkflowInstance?)null);
|
||||
|
||||
return new ValueTask<WorkflowInstance?>(CancelledInstance("instance-missing"));
|
||||
});
|
||||
|
||||
var sut = BuildSut();
|
||||
var outcome = await sut.DrainAsync(DrainTrigger.OperatorForce);
|
||||
|
||||
Assert.Equal(DrainResult.Forced, outcome.OverallResult);
|
||||
Assert.Equal(1, outcome.ExecutionCyclesForceCancelledCount);
|
||||
await InstanceStore.Received().SaveAsync(Arg.Is<WorkflowInstance>(i => i.Id == "instance-missing" && i.SubStatus == WorkflowSubStatus.Interrupted), Arg.Any<CancellationToken>());
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "A confirmed Cancelled pre-cancel snapshot does not promote a later Finished/Cancelled row")]
|
||||
public async Task CancelledSnapshotDoesNotPromoteLaterCancelledInstance()
|
||||
{
|
||||
var handle = new ExecutionCycleHandle(Guid.NewGuid(), "instance-user-cancel", ingressSourceName: "http.trigger", startedAt: DateTimeOffset.UtcNow, linkedToken: CancellationToken.None);
|
||||
ExecutionCycleRegistry.ActiveCount.Returns(1);
|
||||
ExecutionCycleRegistry.ListActiveCycles().Returns(new[] { handle });
|
||||
InstanceStore.FindAsync(Arg.Any<WorkflowInstanceFilter>(), Arg.Any<CancellationToken>())
|
||||
.Returns(_ => new ValueTask<WorkflowInstance?>(CancelledInstance("instance-user-cancel")));
|
||||
|
||||
var sut = BuildSut();
|
||||
var outcome = await sut.DrainAsync(DrainTrigger.OperatorForce);
|
||||
|
||||
Assert.Equal(DrainResult.Forced, outcome.OverallResult);
|
||||
Assert.Equal(1, outcome.ExecutionCyclesForceCancelledCount);
|
||||
await InstanceStore.DidNotReceive().SaveAsync(Arg.Any<WorkflowInstance>(), Arg.Any<CancellationToken>());
|
||||
await LogStore.DidNotReceive().AddAsync(Arg.Any<Entities.WorkflowExecutionLogRecord>(), Arg.Any<CancellationToken>());
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "A null snapshot of an already-disposed handle does not promote a later Finished/Cancelled row")]
|
||||
public async Task DisposedHandleNullSnapshotDoesNotPromoteLaterCancelledInstance()
|
||||
{
|
||||
var handle = new ExecutionCycleHandle(Guid.NewGuid(), "instance-already-done", ingressSourceName: "http.trigger", startedAt: DateTimeOffset.UtcNow, linkedToken: CancellationToken.None);
|
||||
handle.Dispose();
|
||||
ExecutionCycleRegistry.ActiveCount.Returns(1);
|
||||
ExecutionCycleRegistry.ListActiveCycles().Returns(new[] { handle });
|
||||
|
||||
var finds = 0;
|
||||
InstanceStore.FindAsync(Arg.Any<WorkflowInstanceFilter>(), Arg.Any<CancellationToken>())
|
||||
.Returns(_ =>
|
||||
{
|
||||
if (Interlocked.Increment(ref finds) == 1)
|
||||
return new ValueTask<WorkflowInstance?>((WorkflowInstance?)null);
|
||||
|
||||
return new ValueTask<WorkflowInstance?>(CancelledInstance("instance-already-done"));
|
||||
});
|
||||
|
||||
var sut = BuildSut();
|
||||
var outcome = await sut.DrainAsync(DrainTrigger.OperatorForce);
|
||||
|
||||
Assert.Equal(DrainResult.Forced, outcome.OverallResult);
|
||||
Assert.Equal(0, outcome.ExecutionCyclesForceCancelledCount);
|
||||
await InstanceStore.DidNotReceive().SaveAsync(Arg.Any<WorkflowInstance>(), Arg.Any<CancellationToken>());
|
||||
await LogStore.DidNotReceive().AddAsync(Arg.Any<Entities.WorkflowExecutionLogRecord>(), Arg.Any<CancellationToken>());
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "Waiting for a snapshot slot does not burn the per-Find 250ms budget")]
|
||||
public async Task SnapshotQueueWaitDoesNotExcludeLaterFinds()
|
||||
{
|
||||
var cap = DrainOrchestrator.MaxConcurrentPreCancelSnapshotFinds;
|
||||
const int extra = 4;
|
||||
var count = cap + extra;
|
||||
var handles = Enumerable.Range(0, count)
|
||||
.Select(i => new ExecutionCycleHandle(Guid.NewGuid(), $"instance-queued-{i}", ingressSourceName: "http.trigger", startedAt: DateTimeOffset.UtcNow, linkedToken: CancellationToken.None))
|
||||
.ToArray();
|
||||
ExecutionCycleRegistry.ActiveCount.Returns(count);
|
||||
ExecutionCycleRegistry.ListActiveCycles().Returns(handles);
|
||||
|
||||
var hang = new TaskCompletionSource<WorkflowInstance?>(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
var seen = new HashSet<string>(StringComparer.Ordinal);
|
||||
var stalledIds = new HashSet<string>(StringComparer.Ordinal);
|
||||
var promotedIds = new HashSet<string>(StringComparer.Ordinal);
|
||||
var snapshotFinds = 0;
|
||||
|
||||
InstanceStore.FindAsync(Arg.Any<WorkflowInstanceFilter>(), Arg.Any<CancellationToken>())
|
||||
.Returns(ci =>
|
||||
{
|
||||
var id = ci.Arg<WorkflowInstanceFilter>().Id!;
|
||||
lock (seen)
|
||||
{
|
||||
// Phase C: runner committed Finished/Cancelled after force-cancel.
|
||||
// Only ids that snapshot successfully join drainInduced and may promote.
|
||||
if (!seen.Add(id))
|
||||
return new ValueTask<WorkflowInstance?>(CancelledInstance(id));
|
||||
}
|
||||
|
||||
if (Interlocked.Increment(ref snapshotFinds) <= cap)
|
||||
{
|
||||
lock (stalledIds)
|
||||
stalledIds.Add(id);
|
||||
return new ValueTask<WorkflowInstance?>(hang.Task);
|
||||
}
|
||||
|
||||
lock (promotedIds)
|
||||
promotedIds.Add(id);
|
||||
return new ValueTask<WorkflowInstance?>(RunningInstance(id));
|
||||
});
|
||||
|
||||
var sut = BuildSut();
|
||||
var outcome = await sut.DrainAsync(DrainTrigger.OperatorForce).AsTask().WaitAsync(TimeSpan.FromSeconds(8));
|
||||
|
||||
Assert.Equal(DrainResult.Forced, outcome.OverallResult);
|
||||
Assert.Equal(count, outcome.ExecutionCyclesForceCancelledCount);
|
||||
Assert.Equal(cap, stalledIds.Count);
|
||||
Assert.Equal(extra, promotedIds.Count);
|
||||
Assert.False(hang.Task.IsCompleted);
|
||||
|
||||
foreach (var id in stalledIds)
|
||||
await InstanceStore.DidNotReceive().SaveAsync(Arg.Is<WorkflowInstance>(i => i.Id == id), Arg.Any<CancellationToken>());
|
||||
foreach (var id in promotedIds)
|
||||
await InstanceStore.Received().SaveAsync(Arg.Is<WorkflowInstance>(i => i.Id == id && i.SubStatus == WorkflowSubStatus.Interrupted), Arg.Any<CancellationToken>());
|
||||
}
|
||||
|
||||
private static WorkflowInstance RunningInstance(string id) => new()
|
||||
{
|
||||
Id = id,
|
||||
DefinitionId = "def-1",
|
||||
DefinitionVersionId = "ver-1",
|
||||
Version = 1,
|
||||
Status = WorkflowStatus.Running,
|
||||
SubStatus = WorkflowSubStatus.Executing,
|
||||
IsExecuting = true,
|
||||
};
|
||||
|
||||
private static WorkflowInstance CancelledInstance(string id) => new()
|
||||
{
|
||||
Id = id,
|
||||
DefinitionId = "def-1",
|
||||
DefinitionVersionId = "ver-1",
|
||||
Version = 1,
|
||||
Status = WorkflowStatus.Finished,
|
||||
SubStatus = WorkflowSubStatus.Cancelled,
|
||||
IsExecuting = false,
|
||||
};
|
||||
|
||||
[Fact(DisplayName = "Second non-force drain in same generation throws InvalidOperationException")]
|
||||
public async Task SecondNonForceDrainThrows()
|
||||
{
|
||||
|
|
|
|||
|
|
@ -101,6 +101,23 @@ public class ExecutionCycleRegistryTests
|
|||
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.Cancel invokes the cancel callback supplied at registration")]
|
||||
public void CancelCallbackIsInvoked()
|
||||
{
|
||||
|
|
|
|||
Loading…
Reference in a new issue