diff --git a/src/ModularPipelines/Engine/ModuleExecutionPipeline.cs b/src/ModularPipelines/Engine/ModuleExecutionPipeline.cs index 061871c3fb..efc22589b6 100644 --- a/src/ModularPipelines/Engine/ModuleExecutionPipeline.cs +++ b/src/ModularPipelines/Engine/ModuleExecutionPipeline.cs @@ -678,29 +678,10 @@ private async Task> HandleException( executionContext.Exception = exception; - // Check for timeout - use the enhanced exception type for detailed logging - if (exception is ModuleTimeoutException timeoutException) - { - executionContext.Status = Status.TimedOut; + executionContext.Status = ClassifyException(config, executionContext, exception); - // Log additional timeout details - if (!timeoutException.WasCancellationTokenRespected) - { - logger.LogWarning( - "Module {ModuleName} did not complete within the cancellation grace period; timeout enforcement stopped waiting after {ElapsedTime}", - executionContext.ModuleType.Name, - timeoutException.ElapsedTime.ToDisplayString()); - } - } - else if (IsTimeout(config, executionContext, exception)) - { - executionContext.Status = Status.TimedOut; - } - - // Check for pipeline cancellation - else if (IsPipelineCancelled(exception)) + if (executionContext.Status == Status.PipelineTerminated) { - executionContext.Status = Status.PipelineTerminated; logger.LogInformation("Pipeline has been canceled"); var cancelledResult = ModuleResult.CreateFailure(exception, executionContext); @@ -708,9 +689,17 @@ private async Task> HandleException( executionContext.SetTypedResult(cancelledResult); return cancelledResult; } - else + + // Use the enhanced exception type for detailed timeout logging. + if (exception is ModuleTimeoutException timeoutException) { - executionContext.Status = Status.Failed; + if (!timeoutException.WasCancellationTokenRespected) + { + logger.LogWarning( + "Module {ModuleName} did not complete within the cancellation grace period; timeout enforcement stopped waiting after {ElapsedTime}", + executionContext.ModuleType.Name, + timeoutException.ElapsedTime.ToDisplayString()); + } } // Check if we should ignore failures @@ -751,6 +740,23 @@ await SaveResults( throw exception; } + private Status ClassifyException( + ModuleConfiguration config, + ModuleExecutionContext executionContext, + Exception exception) + { + if (!config.AlwaysRun + && _engineCancellationToken.IsCancelled + && exception is OperationCanceledException or ModuleTimeoutException) + { + return Status.PipelineTerminated; + } + + return exception is ModuleTimeoutException || IsTimeout(config, executionContext, exception) + ? Status.TimedOut + : Status.Failed; + } + private bool IsTimeout(ModuleConfiguration config, ModuleExecutionContext executionContext, Exception exception) { var timeout = GetTimeout(config); @@ -760,13 +766,7 @@ private bool IsTimeout(ModuleConfiguration config, ModuleExecutionContext execut } var isTimeoutExceeded = executionContext.Stopwatch.Elapsed >= timeout; - return isTimeoutExceeded && exception is ModuleTimeoutException or TaskCanceledException or OperationCanceledException; - } - - private bool IsPipelineCancelled(Exception exception) - { - return exception is TaskCanceledException or OperationCanceledException or ModuleTimeoutException - && _engineCancellationToken.IsCancelled; + return isTimeoutExceeded && exception is OperationCanceledException; } private void CancelPipelineAndThrow( diff --git a/test/ModularPipelines.UnitTests/Engine/ModuleExecutionPipelineTests.cs b/test/ModularPipelines.UnitTests/Engine/ModuleExecutionPipelineTests.cs index c5a5b72a0f..1cff922f5b 100644 --- a/test/ModularPipelines.UnitTests/Engine/ModuleExecutionPipelineTests.cs +++ b/test/ModularPipelines.UnitTests/Engine/ModuleExecutionPipelineTests.cs @@ -2,9 +2,11 @@ using ModularPipelines.Context; using ModularPipelines.Context.Domains; using ModularPipelines.Caching; +using ModularPipelines.Configuration; using ModularPipelines.Engine; using ModularPipelines.Engine.Execution; using ModularPipelines.Enums; +using ModularPipelines.Exceptions; using ModularPipelines.Helpers; using ModularPipelines.Logging; using ModularPipelines.Models; @@ -61,6 +63,112 @@ protected internal override Task ExecuteAsync( } } + private sealed class TimeoutExceptionModule : Module + { + protected internal override Task ExecuteAsync( + IModuleContext context, + CancellationToken cancellationToken) + { + return Task.FromException(new ModuleTimeoutException( + GetType(), + TimeSpan.FromSeconds(1))); + } + } + + private sealed class ElapsedCancellationModule : Module + { + protected override ModuleConfiguration Configure() => ModuleConfiguration.Create() + .WithTimeout(TimeSpan.FromMilliseconds(5)) + .Build(); + + protected internal override Task ExecuteAsync( + IModuleContext context, + CancellationToken cancellationToken) + { + return Task.FromException(new OperationCanceledException()); + } + } + + private sealed class AlwaysRunTimeoutExceptionModule : Module + { + protected override ModuleConfiguration Configure() => ModuleConfiguration.Create() + .WithAlwaysRun() + .Build(); + + protected internal override Task ExecuteAsync( + IModuleContext context, + CancellationToken cancellationToken) + { + return Task.FromException(new ModuleTimeoutException( + GetType(), + TimeSpan.FromSeconds(1))); + } + } + + private sealed class AlwaysRunElapsedCancellationModule : Module + { + protected override ModuleConfiguration Configure() => ModuleConfiguration.Create() + .WithAlwaysRun() + .WithTimeout(TimeSpan.FromMilliseconds(5)) + .Build(); + + protected internal override Task ExecuteAsync( + IModuleContext context, + CancellationToken cancellationToken) + { + return Task.FromException(new OperationCanceledException()); + } + } + + [Test] + public async Task ExecuteAsync_ClassifiesLateTimeoutAsPipelineTerminated() + { + var result = await ExecuteAfterPipelineCancellation(new TimeoutExceptionModule()); + + await Assert.That(result.ModuleStatus).IsEqualTo(Status.PipelineTerminated); + } + + [Test] + public async Task ExecuteAsync_ClassifiesElapsedCancellationAsPipelineTerminated() + { + var module = new ElapsedCancellationModule(); + var executionContext = new ModuleExecutionContext(module, module.GetType()); + executionContext.Stopwatch.Start(); + await Task.Delay(TimeSpan.FromMilliseconds(25)); + executionContext.Stopwatch.Stop(); + + var result = await ExecuteAfterPipelineCancellation(module, executionContext); + + await Assert.That(result.ModuleStatus).IsEqualTo(Status.PipelineTerminated); + } + + [Test] + public async Task ExecuteAsync_ClassifiesAlwaysRunTimeoutIndependentlyOfPipelineCancellation() + { + var module = new AlwaysRunTimeoutExceptionModule(); + var executionContext = new ModuleExecutionContext(module, module.GetType()); + + await Assert.That(async () => await ExecuteAfterPipelineCancellation(module, executionContext)) + .Throws(); + + await Assert.That(executionContext.Status).IsEqualTo(Status.TimedOut); + } + + [Test] + public async Task ExecuteAsync_ClassifiesAlwaysRunElapsedCancellationAsTimeout() + { + var module = new AlwaysRunElapsedCancellationModule(); + var executionContext = new ModuleExecutionContext(module, module.GetType()); + executionContext.Stopwatch.Start(); + await Task.Delay(TimeSpan.FromMilliseconds(25)); + executionContext.Stopwatch.Stop(); + + await Assert.That(async () => await ExecuteAfterPipelineCancellation(module, executionContext)) + .Throws(); + + await Assert.That(executionContext.Status).IsEqualTo(Status.TimedOut); + } + [Test] public async Task ExecuteAsync_DisposesOriginalAndLinkedCancellationTokenSources() { @@ -295,4 +403,63 @@ await Assert.That(cacheRepository.WriteCancellationToken) .IsEqualTo(cacheRepository.ReadCancellationToken); } } + + private static async Task> ExecuteAfterPipelineCancellation( + Module module, + ModuleExecutionContext? executionContext = null) + { + executionContext ??= new ModuleExecutionContext(module, module.GetType()); + + var logger = new Mock(); + var services = new Mock(); + services.SetupGet(x => x.Options).Returns(new PipelineOptions()); + var moduleContext = new Mock(); + moduleContext.SetupGet(x => x.Logger).Returns(logger.Object); + moduleContext.SetupGet(x => x.Services).Returns(services.Object); + + var resultRepository = new Mock(); + resultRepository.SetupGet(x => x.IsEnabled).Returns(false); + var directHookInvoker = new Mock(); + directHookInvoker + .Setup(x => x.InvokeBeforeExecuteAsync( + module, + moduleContext.Object, + It.IsAny())) + .Returns(Task.CompletedTask); + directHookInvoker + .Setup(x => x.InvokeFailedAsync( + module, + moduleContext.Object, + It.IsAny(), + It.IsAny())) + .Returns(Task.CompletedTask); + directHookInvoker + .Setup(x => x.InvokeAfterExecuteAsync( + module, + moduleContext.Object, + It.IsAny>(), + It.IsAny())) + .ReturnsAsync((ModuleResult?) null); + + using var engineCancellationToken = + new PipelineEngineCancellationToken(new PrimaryExceptionContainer()); + engineCancellationToken.CancelWithException(new InvalidOperationException("Prior module failure")); + + var moduleConditionHandler = new Mock(); + moduleConditionHandler + .Setup(x => x.ShouldIgnore(module, It.IsAny())) + .ReturnsAsync((false, null)); + var pipeline = new ModuleExecutionPipeline( + resultRepository.Object, + engineCancellationToken, + directHookInvoker.Object, + moduleConditionHandler.Object, + OptionsFactory.Create(new PipelineOptions())); + + return await pipeline.ExecuteAsync( + module, + executionContext, + moduleContext.Object, + CancellationToken.None); + } }