Merge pull request #7913 from elsa-workflows/claude/cranky-boyd-269f7e
feat(core): let a container activity handle a child's fault via FaultSignal
This commit is contained in:
commit
0412555b6e
|
|
@ -221,6 +221,18 @@ public static partial class ActivityExecutionContextExtensions
|
|||
/// Send a signal up the current hierarchy of ancestors.
|
||||
/// </summary>
|
||||
public async ValueTask SendSignalAsync(object signal)
|
||||
{
|
||||
await activityExecutionContext.TrySendSignalAsync(signal);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Send a signal up the current hierarchy of ancestors and report whether a receiver claimed it.
|
||||
/// </summary>
|
||||
/// <param name="signal">The signal to send.</param>
|
||||
/// <returns>
|
||||
/// <c>true</c> if a receiver called <see cref="SignalContext.StopPropagation"/>; otherwise, <c>false</c>.
|
||||
/// </returns>
|
||||
internal async ValueTask<bool> TrySendSignalAsync(object signal)
|
||||
{
|
||||
var receivingContexts = new[]
|
||||
{
|
||||
|
|
@ -244,9 +256,11 @@ public static partial class ActivityExecutionContextExtensions
|
|||
if (signalContext.StopPropagationRequested)
|
||||
{
|
||||
logger.LogDebug("Propagation of signal {SignalType} on activity {ActivityId} of type {ActivityType} was stopped", signalTypeName, ancestorContext.Activity.Id, ancestorContext.Activity.Type);
|
||||
return;
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
|
|
@ -335,6 +349,12 @@ public static partial class ActivityExecutionContextExtensions
|
|||
/// <summary>
|
||||
/// Recovers the current activity from a faulted state by resetting fault counts and transitioning the activity to the running status.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// This method is asymmetric with <c>Fault</c>: it <i>sets</i> the current context's fault count to zero, which is idempotent, but
|
||||
/// <i>decrements</i> the count on every ancestor, which is not. Calling it more than once per fault drives ancestor counts negative.
|
||||
/// The status transition only applies to an activity that is still <see cref="ActivityStatus.Faulted"/>, so that a
|
||||
/// <see cref="Elsa.Workflows.Signals.FaultSignal"/> handler that already terminalized the activity does not see its decision undone.
|
||||
/// </remarks>
|
||||
public void RecoverFromFault()
|
||||
{
|
||||
activityExecutionContext.AggregateFaultCount = 0;
|
||||
|
|
@ -344,7 +364,8 @@ public static partial class ActivityExecutionContextExtensions
|
|||
foreach (var ancestor in ancestors)
|
||||
ancestor.AggregateFaultCount--;
|
||||
|
||||
activityExecutionContext.TransitionTo(ActivityStatus.Running);
|
||||
if (activityExecutionContext.Status == ActivityStatus.Faulted)
|
||||
activityExecutionContext.TransitionTo(ActivityStatus.Running);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
using Elsa.Extensions;
|
||||
using Elsa.Workflows.Pipelines.ActivityExecution;
|
||||
using Elsa.Workflows.Signals;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace Elsa.Workflows.Middleware.Activities;
|
||||
|
|
@ -16,7 +17,8 @@ public static class ExceptionHandlingMiddlewareExtensions
|
|||
}
|
||||
|
||||
/// <summary>
|
||||
/// Catches any exceptions thrown by downstream components and transitions the workflow into the faulted state.
|
||||
/// Catches any exceptions thrown by downstream components and transitions the workflow into the faulted state,
|
||||
/// unless an enclosing container claims the fault by handling the <see cref="FaultSignal"/>.
|
||||
/// </summary>
|
||||
public class ExceptionHandlingMiddleware(ActivityMiddlewareDelegate next, IIncidentStrategyResolver incidentStrategyResolver, ILogger<ExceptionHandlingMiddleware> logger)
|
||||
: IActivityExecutionMiddleware
|
||||
|
|
@ -32,6 +34,15 @@ public class ExceptionHandlingMiddleware(ActivityMiddlewareDelegate next, IIncid
|
|||
{
|
||||
logger.LogWarning(e, "An exception was caught from a downstream middleware component");
|
||||
context.Fault(e);
|
||||
|
||||
// Give the ancestor chain a chance to handle the fault. See FaultSignal for the contract: the handler owns
|
||||
// terminalizing the faulted activity, and recovering the fault bookkeeping is ours alone to do, exactly once.
|
||||
if (await context.TrySendSignalAsync(new FaultSignal(e, context)))
|
||||
{
|
||||
context.RecoverFromFault();
|
||||
return;
|
||||
}
|
||||
|
||||
await HandleIncidentAsync(context);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
105
src/modules/Elsa.Workflows.Core/Signals/FaultSignal.cs
Normal file
105
src/modules/Elsa.Workflows.Core/Signals/FaultSignal.cs
Normal file
|
|
@ -0,0 +1,105 @@
|
|||
namespace Elsa.Workflows.Signals;
|
||||
|
||||
/// <summary>
|
||||
/// Sent up the ancestor chain when an activity faulted, giving an enclosing container the chance to handle the failure
|
||||
/// itself instead of letting it fall through to the workflow-global <see cref="IIncidentStrategy"/>.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// A container opts in exactly the way it opts into cancellation, by registering a handler for this signal:
|
||||
/// </para>
|
||||
/// <code>
|
||||
/// public MyContainer()
|
||||
/// {
|
||||
/// OnSignalReceived<FaultSignal>(OnChildFaultedAsync);
|
||||
/// }
|
||||
///
|
||||
/// private async ValueTask OnChildFaultedAsync(FaultSignal signal, SignalContext context)
|
||||
/// {
|
||||
/// if (!IsMine(signal.FaultedContext))
|
||||
/// return; // Not my child: let it keep bubbling.
|
||||
///
|
||||
/// context.StopPropagation(); // I am handling this.
|
||||
/// await signal.FaultedContext.CancelActivityAsync();
|
||||
/// await ScheduleFallbackAsync(context.ReceiverActivityExecutionContext);
|
||||
/// }
|
||||
/// </code>
|
||||
/// <para>
|
||||
/// Bubbling is the default. A handler that does not recognize <see cref="FaultedContext"/> simply returns without
|
||||
/// stopping propagation, and the signal continues to the next ancestor, so nested containers compose without any of
|
||||
/// them knowing about the others. If no ancestor stops propagation, the incident strategy runs exactly as it does
|
||||
/// when this signal does not exist at all.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>The faulting activity is itself a receiver.</b> <c>SendSignalAsync</c> delivers to the sender before walking its
|
||||
/// ancestors, and this signal reuses that dispatch rather than introducing a variant of it. An activity that throws and
|
||||
/// also handles <see cref="FaultSignal"/> therefore sees its own fault first, and may claim it, which is what a
|
||||
/// self-retrying or self-compensating activity wants. This grants no ability to hide a failure that an activity did not
|
||||
/// already have, since one that simply catches its own exception never faults at all. A handler that wants
|
||||
/// ancestors-only semantics should check <see cref="SignalContext.IsSelf"/>, or compare <see cref="FaultedContext"/>
|
||||
/// against its own receiver context, the same way it already checks that the faulting context is one of its children.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>The contract.</b> Responsibilities are split, and the split is deliberate:
|
||||
/// </para>
|
||||
/// <list type="bullet">
|
||||
/// <item>
|
||||
/// <description>
|
||||
/// <b>The middleware</b> calls <c>Fault()</c>, sends this signal, and — only when propagation was stopped — calls
|
||||
/// <c>RecoverFromFault()</c> exactly once. Nothing else.
|
||||
/// </description>
|
||||
/// </item>
|
||||
/// <item>
|
||||
/// <description>
|
||||
/// <b>The handler</b> decides the fault is its own, calls <see cref="SignalContext.StopPropagation"/>, and then
|
||||
/// terminalizes the faulted activity: cancel it, complete it, or reschedule it. There is no single right answer
|
||||
/// for the middleware to pick here — an interrupting error boundary wants the activity cancelled, a retry handler
|
||||
/// wants it rescheduled, a fallback handler may want it completed with a substitute result — so a handler that
|
||||
/// claims the failure also owns deciding what becomes of the failed work.
|
||||
/// </description>
|
||||
/// </item>
|
||||
/// <item>
|
||||
/// <description>
|
||||
/// <b>The handler must not</b> call <c>RecoverFromFault()</c>.
|
||||
/// </description>
|
||||
/// </item>
|
||||
/// </list>
|
||||
/// <para>
|
||||
/// <b>Completing the faulted activity takes one extra step.</b> <c>CompleteActivityAsync</c> returns immediately unless
|
||||
/// the activity is <see cref="ActivityStatus.Running"/>, and throughout the handler it is still
|
||||
/// <see cref="ActivityStatus.Faulted"/>, because recovery runs only once the handler has returned. Completing it inline
|
||||
/// therefore does nothing at all, silently. A handler that wants to complete the activity — substituting a result for
|
||||
/// the work that failed, say — has to move it out of the faulted state first:
|
||||
/// </para>
|
||||
/// <code>
|
||||
/// context.StopPropagation();
|
||||
/// signal.FaultedContext.TransitionTo(ActivityStatus.Running);
|
||||
/// await signal.FaultedContext.CompleteActivityAsync(substituteResult);
|
||||
/// </code>
|
||||
/// <para>
|
||||
/// This is not licence to call <c>RecoverFromFault()</c>, which also rewrites the fault counts and remains the
|
||||
/// middleware's job alone. Completing this way fires the enclosing container's completion callback, so the container's
|
||||
/// normal sequencing resumes. Cancelling needs no equivalent step, because <c>CancelActivityAsync</c> already accepts a
|
||||
/// faulted activity, and rescheduling needs none either.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// That last rule is not stylistic. <c>RecoverFromFault()</c> is asymmetric: it <i>sets</i> the faulting context's
|
||||
/// <see cref="ActivityExecutionContext.AggregateFaultCount"/> to zero, which is idempotent, but <i>decrements</i> the
|
||||
/// count on every ancestor, which is not. A second call is therefore harmless for the faulting context and harmful for
|
||||
/// every ancestor, driving their counts to <c>-1</c>. Those counts are persisted and surface through
|
||||
/// <c>ActivityExecutionRecord.AggregateFaultCount</c> and <c>ActivityExecutionStats</c>, so the failure mode is
|
||||
/// silently wrong fault numbers in the UI rather than an exception anyone would notice.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>On the faulted activity's status.</b> <c>RecoverFromFault()</c> transitions the activity out of
|
||||
/// <see cref="ActivityStatus.Faulted"/> and back to <see cref="ActivityStatus.Running"/>, so a handler that stops
|
||||
/// propagation and terminalizes nothing leaves the activity <see cref="ActivityStatus.Running"/> having already thrown.
|
||||
/// That is the handler's bug. A completing container happens to sweep such an activity via
|
||||
/// <see cref="ActivityExecutionContext.CompleteActivityAsync"/>, which cancels its non-completed children, so the
|
||||
/// mistake degrades rather than hangs — but that is a backstop, not the mechanism, and a container that suspends
|
||||
/// instead of completing will persist the activity as <see cref="ActivityStatus.Running"/>.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
/// <param name="Exception">The exception that caused the fault.</param>
|
||||
/// <param name="FaultedContext">The <see cref="ActivityExecutionContext"/> of the activity that faulted.</param>
|
||||
public record FaultSignal(Exception Exception, ActivityExecutionContext FaultedContext);
|
||||
|
|
@ -0,0 +1,36 @@
|
|||
using Elsa.Testing.Shared.Activities;
|
||||
using Elsa.Workflows.Signals;
|
||||
|
||||
namespace Elsa.Workflows.IntegrationTests.Scenarios.FaultSignals.Activities;
|
||||
|
||||
/// <summary>
|
||||
/// A sequential container that delegates <see cref="FaultSignal"/> handling to a caller-supplied strategy, so that a
|
||||
/// single activity type covers every case under test: declining a fault, claiming it, and claiming it incorrectly.
|
||||
/// </summary>
|
||||
public class FaultHandlingContainer : TestContainer
|
||||
{
|
||||
private readonly Func<FaultSignal, SignalContext, ValueTask>? _onChildFaulted;
|
||||
|
||||
/// <param name="onChildFaulted">
|
||||
/// Invoked for every fault bubbling through this container. Passing <c>null</c> declines every fault, which lets it
|
||||
/// keep bubbling to the next ancestor.
|
||||
/// </param>
|
||||
public FaultHandlingContainer(Func<FaultSignal, SignalContext, ValueTask>? onChildFaulted = null)
|
||||
{
|
||||
_onChildFaulted = onChildFaulted;
|
||||
OnSignalReceived<FaultSignal>(OnChildFaultedAsync);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The number of faults this container was notified of.
|
||||
/// </summary>
|
||||
public int FaultsSeen { get; private set; }
|
||||
|
||||
private async ValueTask OnChildFaultedAsync(FaultSignal signal, SignalContext context)
|
||||
{
|
||||
FaultsSeen++;
|
||||
|
||||
if (_onChildFaulted != null)
|
||||
await _onChildFaulted(signal, context);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,17 @@
|
|||
using Elsa.Workflows.Signals;
|
||||
|
||||
namespace Elsa.Workflows.IntegrationTests.Scenarios.FaultSignals.Activities;
|
||||
|
||||
/// <summary>
|
||||
/// An activity that throws and also claims its own <see cref="FaultSignal"/>, used to pin the channel's self-receipt
|
||||
/// behavior: a signal is delivered to its sender before the walk up the ancestor chain begins.
|
||||
/// </summary>
|
||||
public class SelfHandlingFaultingActivity : CodeActivity
|
||||
{
|
||||
public SelfHandlingFaultingActivity()
|
||||
{
|
||||
OnSignalReceived<FaultSignal>((_, context) => context.StopPropagation());
|
||||
}
|
||||
|
||||
protected override void Execute(ActivityExecutionContext context) => throw new InvalidOperationException("Whoops!");
|
||||
}
|
||||
|
|
@ -0,0 +1,276 @@
|
|||
using Elsa.Extensions;
|
||||
using Elsa.Testing.Shared;
|
||||
using Elsa.Testing.Shared.Activities;
|
||||
using Elsa.Workflows.Activities;
|
||||
using Elsa.Workflows.IncidentStrategies;
|
||||
using Elsa.Workflows.IntegrationTests.Scenarios.FaultSignals.Activities;
|
||||
using Elsa.Workflows.Models;
|
||||
using Elsa.Workflows.Signals;
|
||||
using Xunit.Abstractions;
|
||||
|
||||
namespace Elsa.Workflows.IntegrationTests.Scenarios.FaultSignals;
|
||||
|
||||
/// <summary>
|
||||
/// Integration tests for <see cref="FaultSignal"/>: a container's opportunity to claim a child's fault before it
|
||||
/// becomes a workflow-global incident.
|
||||
/// </summary>
|
||||
public class FaultSignalTests(ITestOutputHelper testOutputHelper)
|
||||
{
|
||||
private readonly WorkflowTestFixture _fixture = new(testOutputHelper);
|
||||
private readonly Fault _faultingActivity = Fault.Create("Whoops!", "Test", "Test");
|
||||
|
||||
[Fact(DisplayName = "A container that handles the signal suppresses the incident strategy")]
|
||||
public async Task HandledFault_DoesNotFaultTheWorkflow()
|
||||
{
|
||||
// Arrange
|
||||
var container = ContainerAround(_faultingActivity, ClaimAndCancelChildAsync);
|
||||
|
||||
// Act
|
||||
var result = await RunAsync(container);
|
||||
|
||||
// Assert
|
||||
Assert.Equal(1, container.FaultsSeen);
|
||||
Assert.Equal(WorkflowStatus.Finished, result.WorkflowState.Status);
|
||||
Assert.Equal(WorkflowSubStatus.Finished, result.WorkflowState.SubStatus);
|
||||
|
||||
// The incident stays on record even though the fault was handled: caught failures remain visible to operators.
|
||||
Assert.Single(result.WorkflowState.Incidents);
|
||||
}
|
||||
|
||||
[Theory(DisplayName = "A fault nobody handles is left to the incident strategy, exactly as before")]
|
||||
[InlineData(typeof(FaultStrategy), WorkflowSubStatus.Faulted)]
|
||||
[InlineData(typeof(ContinueWithIncidentsStrategy), WorkflowSubStatus.Suspended)]
|
||||
public async Task UnhandledFault_LeavesIncidentStrategyInCharge(Type incidentStrategyType, WorkflowSubStatus expectedSubStatus)
|
||||
{
|
||||
// Arrange: a plain container, with no FaultSignal handler anywhere in the chain.
|
||||
var container = new TestContainer
|
||||
{
|
||||
Activities =
|
||||
{
|
||||
_faultingActivity
|
||||
}
|
||||
};
|
||||
|
||||
// Act
|
||||
var result = await RunAsync(container, incidentStrategyType);
|
||||
|
||||
// Assert
|
||||
Assert.Equal(expectedSubStatus, result.WorkflowState.SubStatus);
|
||||
Assert.Single(result.WorkflowState.Incidents);
|
||||
Assert.Equal(ActivityStatus.Faulted, result.GetActivityStatus(_faultingActivity));
|
||||
AssertFaultCounts(result, expected: 1);
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "A fault the inner container declines keeps bubbling to the outer one")]
|
||||
public async Task DeclinedFault_ReachesTheNextAncestor()
|
||||
{
|
||||
// Arrange
|
||||
var inner = ContainerAround(_faultingActivity);
|
||||
var outer = ContainerAround(inner, ClaimAndCancelChildAsync);
|
||||
|
||||
// Act
|
||||
var result = await RunAsync(outer);
|
||||
|
||||
// Assert
|
||||
Assert.Equal(1, inner.FaultsSeen);
|
||||
Assert.Equal(1, outer.FaultsSeen);
|
||||
Assert.Equal(WorkflowSubStatus.Finished, result.WorkflowState.SubStatus);
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "The faulting activity receives its own fault before any ancestor does")]
|
||||
public async Task FaultingActivity_ReceivesItsOwnSignalFirst()
|
||||
{
|
||||
// Pinning the channel's self-plus-ancestors dispatch, which this signal reuses rather than varying. It lets a
|
||||
// self-retrying or self-compensating activity claim its own failure, and grants no ability to hide a failure
|
||||
// that an activity did not already have: one that simply catches its own exception never faults at all.
|
||||
|
||||
// Arrange
|
||||
var faultingActivity = new SelfHandlingFaultingActivity();
|
||||
var container = ContainerAround(faultingActivity);
|
||||
|
||||
// Act
|
||||
var result = await RunAsync(container, typeof(FaultStrategy));
|
||||
|
||||
// Assert: the activity claimed its own fault, so the walk never reached the container.
|
||||
Assert.Equal(0, container.FaultsSeen);
|
||||
Assert.NotEqual(WorkflowSubStatus.Faulted, result.WorkflowState.SubStatus);
|
||||
|
||||
// The incident is still on record, and the fault bookkeeping was still recovered exactly once.
|
||||
Assert.Single(result.WorkflowState.Incidents);
|
||||
|
||||
var faultedContext = result.GetActivityContext(faultingActivity);
|
||||
Assert.NotNull(faultedContext);
|
||||
Assert.Equal(0, faultedContext.AggregateFaultCount);
|
||||
Assert.All(faultedContext.GetAncestors(), x => Assert.Equal(0, x.AggregateFaultCount));
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "A handled fault restores the fault count on the faulting context and every ancestor")]
|
||||
public async Task HandledFault_RestoresFaultCounts()
|
||||
{
|
||||
// Arrange
|
||||
var container = ContainerAround(_faultingActivity, ClaimAndCancelChildAsync);
|
||||
|
||||
// Act
|
||||
var result = await RunAsync(container);
|
||||
|
||||
// Assert
|
||||
AssertFaultCounts(result, expected: 0);
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "A handler that also recovers from the fault drives ancestor fault counts negative")]
|
||||
public async Task HandlerThatAlsoRecoversFromFault_CorruptsAncestorFaultCounts()
|
||||
{
|
||||
// Asserting the documented consequence of violating the contract rather than leaving it accidental: recovery
|
||||
// *sets* the faulting context's count to zero, which is idempotent, but *decrements* every ancestor, which is
|
||||
// not. The middleware recovers too, so the ancestors end up one below where they started.
|
||||
|
||||
// Arrange
|
||||
var container = ContainerAround(_faultingActivity, async (signal, context) =>
|
||||
{
|
||||
context.StopPropagation();
|
||||
signal.FaultedContext.RecoverFromFault();
|
||||
await context.ReceiverActivityExecutionContext.CompleteActivityAsync();
|
||||
});
|
||||
|
||||
// Act
|
||||
var result = await RunAsync(container);
|
||||
|
||||
// Assert
|
||||
var faultedContext = result.GetActivityContext(_faultingActivity);
|
||||
Assert.NotNull(faultedContext);
|
||||
|
||||
var ancestors = faultedContext.GetAncestors().ToList();
|
||||
Assert.NotEmpty(ancestors);
|
||||
Assert.Equal(0, faultedContext.AggregateFaultCount);
|
||||
Assert.All(ancestors, x => Assert.Equal(-1, x.AggregateFaultCount));
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "A handler can complete the faulted child with a substitute result")]
|
||||
public async Task HandlerThatCompletesChildWithSubstituteResult_ResumesTheContainer()
|
||||
{
|
||||
// Arrange
|
||||
var container = new FaultHandlingContainer(async (signal, context) =>
|
||||
{
|
||||
context.StopPropagation();
|
||||
|
||||
// CompleteActivityAsync no-ops on a non-Running activity, and the middleware recovers only once this
|
||||
// handler has returned, so the faulted child has to be moved out of Faulted first. This is not the same as
|
||||
// RecoverFromFault, which would also rewrite the fault counts.
|
||||
signal.FaultedContext.TransitionTo(ActivityStatus.Running);
|
||||
await signal.FaultedContext.CompleteActivityAsync("substitute");
|
||||
})
|
||||
{
|
||||
Activities =
|
||||
{
|
||||
_faultingActivity,
|
||||
new WriteLine("after")
|
||||
}
|
||||
};
|
||||
|
||||
// Act
|
||||
var result = await RunAsync(container);
|
||||
|
||||
// Assert: completing the child fires the container's completion callback, so sequencing resumes.
|
||||
Assert.Equal(ActivityStatus.Completed, result.GetActivityStatus(_faultingActivity));
|
||||
Assert.Equal(new[] { "after" }, _fixture.CapturingTextWriter.Lines);
|
||||
Assert.Equal(WorkflowSubStatus.Finished, result.WorkflowState.SubStatus);
|
||||
AssertFaultCounts(result, expected: 0);
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "A handler that cancels the faulted child leaves it Canceled, not Running")]
|
||||
public async Task HandlerThatCancelsChild_LeavesItCanceled()
|
||||
{
|
||||
// Arrange
|
||||
var container = ContainerAround(_faultingActivity, ClaimAndCancelChildAsync);
|
||||
|
||||
// Act
|
||||
var result = await RunAsync(container);
|
||||
|
||||
// Assert
|
||||
Assert.Equal(ActivityStatus.Canceled, result.GetActivityStatus(_faultingActivity));
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "A handler that terminalizes nothing leaves the child Running and the workflow suspended")]
|
||||
public async Task HandlerThatTerminalizesNothing_DegradesRatherThanHangs()
|
||||
{
|
||||
// The handler's bug, documented: recovery transitions the faulted child back to Running, and a handler that
|
||||
// schedules no follow-up work moves it no further. The run returns rather than hanging, but the workflow
|
||||
// suspends with nothing left to resume it, which is why terminalizing is the handler's responsibility.
|
||||
|
||||
// Arrange
|
||||
var container = ContainerAround(_faultingActivity, (_, context) =>
|
||||
{
|
||||
context.StopPropagation();
|
||||
return default;
|
||||
});
|
||||
|
||||
// Act
|
||||
var result = await RunAsync(container);
|
||||
|
||||
// Assert
|
||||
Assert.Equal(ActivityStatus.Running, result.GetActivityStatus(_faultingActivity));
|
||||
Assert.Equal(WorkflowStatus.Running, result.WorkflowState.Status);
|
||||
Assert.Equal(WorkflowSubStatus.Suspended, result.WorkflowState.SubStatus);
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "A completing container sweeps a child the handler failed to terminalize")]
|
||||
public async Task HandlerThatTerminalizesNothing_IsBackstoppedByTheCompletingContainer()
|
||||
{
|
||||
// The backstop, not the mechanism: CompleteActivityAsync cancels non-completed children on the way out.
|
||||
|
||||
// Arrange
|
||||
var container = ContainerAround(_faultingActivity, async (_, context) =>
|
||||
{
|
||||
context.StopPropagation();
|
||||
await context.ReceiverActivityExecutionContext.CompleteActivityAsync();
|
||||
});
|
||||
|
||||
// Act
|
||||
var result = await RunAsync(container);
|
||||
|
||||
// Assert
|
||||
Assert.Equal(ActivityStatus.Completed, result.GetActivityStatus(container));
|
||||
Assert.Equal(ActivityStatus.Canceled, result.GetActivityStatus(_faultingActivity));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The canonical handler: claim the fault, terminalize the faulted child, and wind the container up.
|
||||
/// </summary>
|
||||
private static async ValueTask ClaimAndCancelChildAsync(FaultSignal signal, SignalContext context)
|
||||
{
|
||||
context.StopPropagation();
|
||||
await signal.FaultedContext.CancelActivityAsync();
|
||||
await context.ReceiverActivityExecutionContext.CompleteActivityAsync();
|
||||
}
|
||||
|
||||
private static FaultHandlingContainer ContainerAround(IActivity child, Func<FaultSignal, SignalContext, ValueTask>? onChildFaulted = null)
|
||||
{
|
||||
return new(onChildFaulted)
|
||||
{
|
||||
Activities =
|
||||
{
|
||||
child
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
private Task<RunWorkflowResult> RunAsync(IActivity root, Type? incidentStrategyType = null)
|
||||
{
|
||||
return _fixture.RunWorkflowAsync(new TestWorkflow(builder =>
|
||||
{
|
||||
builder.WorkflowOptions.IncidentStrategyType = incidentStrategyType;
|
||||
builder.Root = root;
|
||||
}));
|
||||
}
|
||||
|
||||
private void AssertFaultCounts(RunWorkflowResult result, int expected)
|
||||
{
|
||||
var faultedContext = result.GetActivityContext(_faultingActivity);
|
||||
Assert.NotNull(faultedContext);
|
||||
|
||||
var ancestors = faultedContext.GetAncestors().ToList();
|
||||
Assert.NotEmpty(ancestors);
|
||||
Assert.Equal(expected, faultedContext.AggregateFaultCount);
|
||||
Assert.All(ancestors, x => Assert.Equal(expected, x.AggregateFaultCount));
|
||||
}
|
||||
}
|
||||
|
|
@ -36,4 +36,74 @@ public class FaultHandlingTests
|
|||
Assert.Equal(0, context.AggregateFaultCount);
|
||||
Assert.Equal(ActivityStatus.Running, context.Status);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Fault_IncrementsFaultCountOnEveryAncestor()
|
||||
{
|
||||
// Arrange
|
||||
var chain = await CreateContextChainAsync();
|
||||
|
||||
// Act
|
||||
chain[^1].Fault(new InvalidOperationException("Test error"));
|
||||
|
||||
// Assert
|
||||
Assert.Equal(new[] { 1, 1, 1 }, chain.Select(x => x.AggregateFaultCount));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task RecoverFromFault_ResetsFaultCountOnEveryAncestor()
|
||||
{
|
||||
// Arrange
|
||||
var chain = await CreateContextChainAsync();
|
||||
var faultedContext = chain[^1];
|
||||
faultedContext.Fault(new InvalidOperationException("Test error"));
|
||||
|
||||
// Act
|
||||
faultedContext.RecoverFromFault();
|
||||
|
||||
// Assert
|
||||
Assert.Equal(new[] { 0, 0, 0 }, chain.Select(x => x.AggregateFaultCount));
|
||||
Assert.Equal(ActivityStatus.Running, faultedContext.Status);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task RecoverFromFault_CalledTwice_DrivesAncestorFaultCountsNegative()
|
||||
{
|
||||
// This is why a FaultSignal handler must not call RecoverFromFault: recovery *sets* the faulting context's
|
||||
// count to zero, which is idempotent, but *decrements* every ancestor, which is not. The failure mode is
|
||||
// silently wrong fault numbers rather than an exception, so it is asserted here rather than left accidental.
|
||||
|
||||
// Arrange
|
||||
var chain = await CreateContextChainAsync();
|
||||
var faultedContext = chain[^1];
|
||||
faultedContext.Fault(new InvalidOperationException("Test error"));
|
||||
|
||||
// Act
|
||||
faultedContext.RecoverFromFault();
|
||||
faultedContext.RecoverFromFault();
|
||||
|
||||
// Assert
|
||||
Assert.Equal(0, faultedContext.AggregateFaultCount);
|
||||
Assert.Equal(new[] { -1, -1 }, chain.Take(chain.Count - 1).Select(x => x.AggregateFaultCount));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task RecoverFromFault_LeavesAlreadyTerminalizedStatusAlone()
|
||||
{
|
||||
// A FaultSignal handler terminalizes the faulted activity, and the middleware recovers the fault bookkeeping
|
||||
// afterwards. Recovery must restore the counts without resurrecting the status the handler chose.
|
||||
|
||||
// Arrange
|
||||
var chain = await CreateContextChainAsync();
|
||||
var faultedContext = chain[^1];
|
||||
faultedContext.Fault(new InvalidOperationException("Test error"));
|
||||
faultedContext.TransitionTo(ActivityStatus.Canceled);
|
||||
|
||||
// Act
|
||||
faultedContext.RecoverFromFault();
|
||||
|
||||
// Assert
|
||||
Assert.Equal(ActivityStatus.Canceled, faultedContext.Status);
|
||||
Assert.Equal(new[] { 0, 0, 0 }, chain.Select(x => x.AggregateFaultCount));
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -19,4 +19,28 @@ public static class TestHelpers
|
|||
var fixture = new ActivityTestFixture(activity);
|
||||
return fixture.BuildAsync();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Creates a chain of nested activity execution contexts for testing ancestor-sensitive behavior.
|
||||
/// </summary>
|
||||
/// <param name="depth">The number of contexts to create. Must be at least two.</param>
|
||||
/// <returns>The contexts, ordered from the outermost ancestor to the innermost descendant.</returns>
|
||||
public static async Task<IReadOnlyList<ActivityExecutionContext>> CreateContextChainAsync(int depth = 3)
|
||||
{
|
||||
// Every level is a Sequence so that a single registered activity type covers the whole chain.
|
||||
var activities = Enumerable.Range(0, depth).Select(_ => new Sequence()).ToList();
|
||||
|
||||
for (var i = 0; i < depth - 1; i++)
|
||||
activities[i].Activities.Add(activities[i + 1]);
|
||||
|
||||
var contexts = new List<ActivityExecutionContext>
|
||||
{
|
||||
await CreateContextAsync(activities[0])
|
||||
};
|
||||
|
||||
for (var i = 1; i < depth; i++)
|
||||
contexts.Add(await contexts[0].WorkflowExecutionContext.CreateActivityExecutionContextAsync(activities[i], new() { Owner = contexts[i - 1] }));
|
||||
|
||||
return contexts;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue