diff --git a/docs/docs/how-to/retry-policy.md b/docs/docs/how-to/retry-policy.md index 5785a383edd..4b5b5d8388c 100644 --- a/docs/docs/how-to/retry-policy.md +++ b/docs/docs/how-to/retry-policy.md @@ -97,6 +97,12 @@ public class ResilientModule : Module } ``` +`WithTimeout` applies to each execution attempt, not to the whole retry chain. Backoff delays run +outside that timeout. A timed-out attempt is passed to the resilience shield as a +`ModuleTimeoutException`, so exception filters still decide whether it should be retried. If the +attempt remains active after the cancellation grace period, the engine bypasses retries to avoid +running the same module instance concurrently with its abandoned attempt. + ## Default Retry Configuration Retries are off by default. You can set a default retry count on the `PipelineOptions`: diff --git a/docs/docs/how-to/timeouts.md b/docs/docs/how-to/timeouts.md index 094dd9f1d55..54f22daafb1 100644 --- a/docs/docs/how-to/timeouts.md +++ b/docs/docs/how-to/timeouts.md @@ -75,9 +75,16 @@ public class ResilientModule : Module ## Timeout Behavior +Timeouts apply to each execution attempt. With retries enabled, every attempt receives the full +timeout and retry backoff delays do not consume it. A module configured with a five-minute timeout +and three retries can therefore spend up to five minutes in each of its four attempts, plus retry +delays. + When a timeout occurs: - The `CancellationToken` passed to `ExecuteAsync` will be cancelled - The module will fail with a `ModuleTimeoutException` -- If retry policies are configured, the module may be retried +- If retry policies are configured and the attempt stops within the cancellation grace period, + the module may be retried. An attempt that remains active after the grace period is never retried, + preventing concurrent executions of the same module instance. - If `WithIgnoreFailures()` is configured, the pipeline will continue despite the timeout diff --git a/src/ModularPipelines/Configuration/ModuleConfiguration.cs b/src/ModularPipelines/Configuration/ModuleConfiguration.cs index f74183aeeb3..03dc66b076b 100644 --- a/src/ModularPipelines/Configuration/ModuleConfiguration.cs +++ b/src/ModularPipelines/Configuration/ModuleConfiguration.cs @@ -58,10 +58,10 @@ public sealed class ModuleConfiguration internal Func>? PlanningSkipCondition { get; init; } /// - /// Gets the timeout duration for module execution. + /// Gets the timeout duration for each module execution attempt. /// /// - /// A representing the maximum execution time, + /// A representing the maximum time for each attempt, /// or null if no timeout is configured. /// public TimeSpan? Timeout { get; init; } diff --git a/src/ModularPipelines/Configuration/ModuleConfigurationBuilder.cs b/src/ModularPipelines/Configuration/ModuleConfigurationBuilder.cs index 7d3b8708bbd..fd68aa35515 100644 --- a/src/ModularPipelines/Configuration/ModuleConfigurationBuilder.cs +++ b/src/ModularPipelines/Configuration/ModuleConfigurationBuilder.cs @@ -325,10 +325,13 @@ public ModuleConfigurationBuilder DependsOnIf(Type moduleType, bool condition) #region WithTimeout /// - /// Sets the timeout duration for module execution. + /// Sets the timeout duration for each module execution attempt. /// - /// The maximum duration allowed for module execution. + /// The maximum duration allowed for each execution attempt. /// This builder instance for method chaining. + /// + /// When retries are configured, the timeout restarts for every attempt and does not include retry delays. + /// public ModuleConfigurationBuilder WithTimeout(TimeSpan timeout) { _timeout = timeout; @@ -349,6 +352,7 @@ public ModuleConfigurationBuilder WithTimeout(TimeSpan timeout) /// /// Each delay uses equal jitter between half and all of its exponential-backoff ceiling. /// A null retries every exception handled by the retry engine. + /// A configured module timeout applies separately to each attempt and does not include these delays. /// public ModuleConfigurationBuilder WithRetry( int count, diff --git a/src/ModularPipelines/Context/Command.cs b/src/ModularPipelines/Context/Command.cs index c251ea54961..5587558bb05 100644 --- a/src/ModularPipelines/Context/Command.cs +++ b/src/ModularPipelines/Context/Command.cs @@ -553,10 +553,7 @@ await WaitForForcefulCancellationAsync( command.WorkingDirPath)); var failure = loggingFailures.CombineWith(e); - if (ShouldPreserveCallerCancellation(e, failure, callerCancellationToken)) - { - throw; - } + ThrowCallerCancellationIfRequired(e, failure, callerCancellationToken); throw CreateExecutionFailure( e, @@ -627,6 +624,38 @@ private static bool ShouldPreserveCallerCancellation( && ReferenceEquals(combinedFailure, executionFailure); } + private static void ThrowCallerCancellationIfRequired( + Exception executionFailure, + Exception combinedFailure, + CancellationToken callerCancellationToken) + { + if (!ShouldPreserveCallerCancellation( + executionFailure, + combinedFailure, + callerCancellationToken)) + { + return; + } + + if (executionFailure is OperationCanceledException cancellationException + && cancellationException.CancellationToken != callerCancellationToken) + { + throw cancellationException is TaskCanceledException + ? new TaskCanceledException( + cancellationException.Message, + cancellationException, + callerCancellationToken) + : new OperationCanceledException( + cancellationException.Message, + cancellationException, + callerCancellationToken); + } + + System.Runtime.ExceptionServices.ExceptionDispatchInfo + .Capture(executionFailure) + .Throw(); + } + private Exception CreateExecutionFailure( Exception executionFailure, Exception combinedFailure, diff --git a/src/ModularPipelines/Engine/ModuleExecutionPipeline.cs b/src/ModularPipelines/Engine/ModuleExecutionPipeline.cs index 8997f5e779a..7fa61e831fa 100644 --- a/src/ModularPipelines/Engine/ModuleExecutionPipeline.cs +++ b/src/ModularPipelines/Engine/ModuleExecutionPipeline.cs @@ -472,80 +472,126 @@ private async Task ExecuteWithPolicies( IModuleContext moduleContext) { var timeout = GetTimeout(config); - if (config.Timeout is null) - { - if (timeout == TimeSpan.Zero) - { - moduleContext.Logger.LogTrace("No module timeout configured. The pipeline default timeout is disabled"); - } - else - { - moduleContext.Logger.LogTrace("No module timeout configured. Using pipeline default timeout {Timeout}", timeout); - } - } + LogTimeoutConfiguration(config, timeout, moduleContext.Logger); var cancellationToken = executionContext.ModuleCancellationTokenSource.Token; // Get resilience shield if applicable var resilienceShield = GetResilienceShield(config, moduleContext); - var moduleAttemptCount = 0; - var moduleAttemptRespondedToCancellation = 0; + using var retryCancellationTokenSource = resilienceShield is null + ? null + : CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + var policyExecutionState = new PolicyExecutionState(); + + // Keep timeout enforcement inside the resilience shield so each attempt gets a fresh budget + // and shield-owned backoff delays are not mistaken for unresponsive module execution. + Task ExecuteModuleAttempt(CancellationToken ct) => ExecuteModuleAttemptAsync( + module, + executionContext, + moduleContext, + timeout, + retryCancellationTokenSource, + policyExecutionState, + ct); - async Task ExecuteModuleAttempt(CancellationToken ct) + T result; + try { - Interlocked.Increment(ref moduleAttemptCount); - try - { - return await module.ExecuteAsync(moduleContext, ct).ConfigureAwait(false); - } - finally - { - if (ct.IsCancellationRequested) - { - Volatile.Write(ref moduleAttemptRespondedToCancellation, 1); - } - } + result = resilienceShield != null + ? await resilienceShield.ExecuteAsync( + async shieldToken => await ExecuteModuleAttempt(shieldToken).ConfigureAwait(false), + retryCancellationTokenSource!.Token).ConfigureAwait(false) + : await ExecuteModuleAttempt(cancellationToken).ConfigureAwait(false); } - - // Create the execution function that optionally includes resilience strategies - Func> executeFunc = resilienceShield != null - ? async ct => await resilienceShield.ExecuteAsync( - async shieldToken => await ExecuteModuleAttempt(shieldToken).ConfigureAwait(false), - ct).ConfigureAwait(false) - : ExecuteModuleAttempt; - - // Use TimeoutHelper with detailed results to get information about token cooperation - TimeoutExecutionResult timeoutResult; - try + catch (OperationCanceledException) when (policyExecutionState.AbandonedAttemptTimeout is not null + && !cancellationToken.IsCancellationRequested) { - timeoutResult = await TimeoutHelper.ExecuteWithTimeoutAndDetailsAsync( - executeFunc, - timeout == TimeSpan.Zero ? null : timeout, - cancellationToken, - $"Module {executionContext.ModuleType.Name} timed out after {timeout}").ConfigureAwait(false); + return ThrowPreservingStack(policyExecutionState.AbandonedAttemptTimeout); } finally { ModuleActivityTracing.RecordModuleRetries( executionContext.ModuleType, - Math.Max(0, Volatile.Read(ref moduleAttemptCount) - 1)); + policyExecutionState.RetryCount); } - if (timeoutResult.TimedOut) + if (policyExecutionState.AbandonedAttemptTimeout is { } abandonedAttemptTimeout) { - var wasCancellationTokenRespected = timeoutResult.WasCancellationTokenRespected - && (resilienceShield is null - || Volatile.Read(ref moduleAttemptRespondedToCancellation) == 1); + return ThrowPreservingStack(abandonedAttemptTimeout); + } - // Create a detailed timeout exception with information about token cooperation - throw new ModuleTimeoutException( - executionContext.ModuleType, - timeout, - timeoutResult.ElapsedTime, - wasCancellationTokenRespected); + return result; + } + + private static void LogTimeoutConfiguration( + ModuleConfiguration config, + TimeSpan timeout, + IModuleLogger logger) + { + if (config.Timeout is not null) + { + return; + } + + if (timeout == TimeSpan.Zero) + { + logger.LogTrace("No module timeout configured. The pipeline default timeout is disabled"); + return; } - return timeoutResult.Value!; + logger.LogTrace("No module timeout configured. Using pipeline default timeout {Timeout}", timeout); + } + + private static async Task ExecuteModuleAttemptAsync( + Module module, + ModuleExecutionContext executionContext, + IModuleContext moduleContext, + TimeSpan timeout, + CancellationTokenSource? retryCancellationTokenSource, + PolicyExecutionState policyExecutionState, + CancellationToken cancellationToken) + { + if (policyExecutionState.AbandonedAttemptTimeout is { } abandonedAttemptTimeout) + { + return ThrowPreservingStack(abandonedAttemptTimeout); + } + + policyExecutionState.RecordAttempt(); + + var timeoutResult = await TimeoutHelper.ExecuteWithTimeoutAndDetailsAsync( + attemptToken => module.ExecuteAsync(moduleContext, attemptToken), + timeout == TimeSpan.Zero ? null : timeout, + cancellationToken, + $"Module {executionContext.ModuleType.Name} timed out after {timeout}").ConfigureAwait(false); + + if (!timeoutResult.TimedOut) + { + return timeoutResult.Value!; + } + + var timeoutException = new ModuleTimeoutException( + executionContext.ModuleType, + timeout, + timeoutResult.ElapsedTime, + timeoutResult.WasCancellationTokenRespected); + + if (!timeoutResult.WasCancellationTokenRespected) + { + policyExecutionState.AbandonedAttemptTimeout = timeoutException; + // Cancel the shield before its retry delay because re-entering this module while + // the abandoned attempt is active is unsafe. + retryCancellationTokenSource?.Cancel(); + } + + throw timeoutException; + } + + private static T ThrowPreservingStack(Exception exception) + { + System.Runtime.ExceptionServices.ExceptionDispatchInfo + .Capture(exception) + .Throw(); + throw new System.Diagnostics.UnreachableException(); } private TimeSpan GetTimeout(ModuleConfiguration config) @@ -680,7 +726,7 @@ private async Task> HandleException( executionContext.Exception = exception; - executionContext.Status = ClassifyException(config, executionContext, exception); + executionContext.Status = ClassifyException(config, exception); // Use the enhanced exception type for detailed timeout logging. if (exception is ModuleTimeoutException timeoutException) @@ -693,12 +739,10 @@ private async Task> HandleException( timeoutException.ElapsedTime.ToDisplayString()); } } - // Check if we should ignore failures if (config.IgnoreFailuresCondition != null && (executionContext.Status != Status.PipelineTerminated - || exception is ModuleTimeoutException - || IsTimeout(config, executionContext, exception))) + || exception is ModuleTimeoutException)) { if (await config.IgnoreFailuresCondition(moduleContext, exception).ConfigureAwait(false)) { @@ -747,31 +791,23 @@ await SaveResults( private Status ClassifyException( ModuleConfiguration config, - ModuleExecutionContext executionContext, Exception exception) { if (!config.AlwaysRun - && _engineCancellationToken.IsCancelled - && exception is OperationCanceledException or ModuleTimeoutException) + && IsPipelineCancelled(exception)) { return Status.PipelineTerminated; } - return exception is ModuleTimeoutException || IsTimeout(config, executionContext, exception) + return exception is ModuleTimeoutException ? Status.TimedOut : Status.Failed; } - private bool IsTimeout(ModuleConfiguration config, ModuleExecutionContext executionContext, Exception exception) + private bool IsPipelineCancelled(Exception exception) { - var timeout = GetTimeout(config); - if (timeout == TimeSpan.Zero) - { - return false; - } - - var isTimeoutExceeded = executionContext.Stopwatch.Elapsed >= timeout; - return isTimeoutExceeded && exception is OperationCanceledException; + return exception is OperationCanceledException or ModuleTimeoutException + && _engineCancellationToken.IsCancelled; } private void CancelPipelineAndThrow( @@ -841,6 +877,17 @@ public void Complete(ModuleResult result) } } + private sealed class PolicyExecutionState + { + private int _moduleAttemptCount; + + public ModuleTimeoutException? AbandonedAttemptTimeout { get; set; } + + public int RetryCount => Math.Max(0, Volatile.Read(ref _moduleAttemptCount) - 1); + + public void RecordAttempt() => Interlocked.Increment(ref _moduleAttemptCount); + } + private static void LogModuleStatus(ModuleExecutionContext executionContext, IModuleLogger logger) { var moduleName = executionContext.ModuleType.Name; diff --git a/src/ModularPipelines/Exceptions/ModuleTimeoutException.cs b/src/ModularPipelines/Exceptions/ModuleTimeoutException.cs index 8376be45421..558bb87044c 100644 --- a/src/ModularPipelines/Exceptions/ModuleTimeoutException.cs +++ b/src/ModularPipelines/Exceptions/ModuleTimeoutException.cs @@ -7,12 +7,13 @@ namespace ModularPipelines.Exceptions; /// /// /// -/// This exception is thrown when a module's execution time exceeds the configured timeout. +/// This exception is thrown when a module execution attempt exceeds the configured timeout. /// The timeout can be configured per-module using the Timeout property or module options. +/// When retries are configured, each attempt receives a fresh timeout and retry delays are excluded. /// /// When this is thrown: /// -/// When a module's ExecuteAsync takes longer than the configured timeout +/// When a module's ExecuteAsync attempt takes longer than the configured timeout /// When the module does not respond to cancellation token within the grace period /// /// Properties available: diff --git a/src/ModularPipelines/Helpers/TimeoutHelper.cs b/src/ModularPipelines/Helpers/TimeoutHelper.cs index fbe6524d143..d03eaa4862c 100644 --- a/src/ModularPipelines/Helpers/TimeoutHelper.cs +++ b/src/ModularPipelines/Helpers/TimeoutHelper.cs @@ -90,117 +90,174 @@ public static async Task> ExecuteWithTimeoutAndDetails // Fast path: no timeout specified if (!timeout.HasValue || timeout.Value == TimeSpan.Zero) { - var task = taskFactory(cancellationToken); + return await ExecuteWithoutTimeoutAsync(taskFactory, cancellationToken, stopwatch) + .ConfigureAwait(false); + } - // If the token can't be cancelled, just await directly (avoid allocations) - if (!cancellationToken.CanBeCanceled) - { - var result = await task.ConfigureAwait(false); - return TimeoutExecutionResult.Success(result, stopwatch.Elapsed); - } + // Keep the attempt token separate so signal ordering is recorded before + // cancellation is propagated to the executing task. + using var deadlineCts = new CancellationTokenSource(); + using var attemptCts = new CancellationTokenSource(); - // Race against cancellation - TrySetCanceled makes the TCS throw - // OperationCanceledException when awaited - var tcs = new TaskCompletionSource( - TaskCreationOptions.RunContinuationsAsynchronously); - using var reg = cancellationToken.Register( - static state => ((TaskCompletionSource) state!).TrySetCanceled(), - tcs); + var deadlineState = new CancellationSignalState(attemptCts); + var externalCancellationState = new CancellationSignalState(attemptCts); + using var deadlineRegistration = deadlineCts.Token.Register( + static state => ((CancellationSignalState) state!).SignalCancellation(), + deadlineState); + using var externalCancellationRegistration = cancellationToken.Register( + static state => ((CancellationSignalState) state!).SignalCancellation(), + externalCancellationState); - var fastPathWinner = await Task.WhenAny(task, tcs.Task).ConfigureAwait(false); - if (fastPathWinner != task) - { - TaskObservation.ObserveFault(task); - } + // Now schedule the timeout - registration is guaranteed to catch it + deadlineCts.CancelAfter(timeout.Value); + + var executionTask = taskFactory(attemptCts.Token); + deadlineState.PublishExecutionTask(executionTask); + externalCancellationState.PublishExecutionTask(executionTask); + + await Task.WhenAny( + executionTask, + deadlineState.Signal.Task, + externalCancellationState.Signal.Task) + .ConfigureAwait(false); + + if (externalCancellationState.Signal.Task.IsCompletedSuccessfully + && !await externalCancellationState.Signal.Task.ConfigureAwait(false)) + { + TaskObservation.ObserveFault(executionTask); + throw new OperationCanceledException(cancellationToken); + } - var winningResult = await fastPathWinner.ConfigureAwait(false); - return TimeoutExecutionResult.Success(winningResult, stopwatch.Elapsed); + if (deadlineState.Signal.Task.IsCompletedSuccessfully + && !await deadlineState.Signal.Task.ConfigureAwait(false)) + { + return await CreateTimeoutResultAsync(executionTask, cancellationToken, stopwatch) + .ConfigureAwait(false); } - // Timeout path: create linked token so task can observe both timeout - // and external cancellation. - using var timeoutCts = CancellationTokenSource - .CreateLinkedTokenSource(cancellationToken); + var value = await executionTask.ConfigureAwait(false); + return TimeoutExecutionResult.Success(value, stopwatch.Elapsed); + } - // Set up cancellation detection BEFORE scheduling timeout to avoid race - // condition where timeout fires before registration completes - // (with very small timeouts) - var cancelledTcs = new TaskCompletionSource( + private static async Task> ExecuteWithoutTimeoutAsync( + Func> taskFactory, + CancellationToken cancellationToken, + Stopwatch stopwatch) + { + var task = taskFactory(cancellationToken); + + if (!cancellationToken.CanBeCanceled) + { + var result = await task.ConfigureAwait(false); + return TimeoutExecutionResult.Success(result, stopwatch.Elapsed); + } + + var cancellationTaskSource = new TaskCompletionSource( TaskCreationOptions.RunContinuationsAsynchronously); - using var registration = timeoutCts.Token.Register( - static state => ((TaskCompletionSource) state!) - .TrySetCanceled(), - cancelledTcs); + using var registration = cancellationToken.Register( + static state => ((TaskCompletionSource) state!).TrySetCanceled(), + cancellationTaskSource); - // Now schedule the timeout - registration is guaranteed to catch it - timeoutCts.CancelAfter(timeout.Value); + var winner = await Task.WhenAny(task, cancellationTaskSource.Task).ConfigureAwait(false); + if (winner != task) + { + TaskObservation.ObserveFault(task); + } - var executionTask = taskFactory(timeoutCts.Token); + var resultValue = await winner.ConfigureAwait(false); + return TimeoutExecutionResult.Success(resultValue, stopwatch.Elapsed); + } - var winner = await Task.WhenAny(executionTask, cancelledTcs.Task) + private static async Task> CreateTimeoutResultAsync( + Task executionTask, + CancellationToken cancellationToken, + Stopwatch stopwatch) + { + if (cancellationToken.IsCancellationRequested) + { + TaskObservation.ObserveFault(executionTask); + throw new OperationCanceledException(cancellationToken); + } + + var taskRespondedDuringGrace = await DidTaskRespondDuringGracePeriodAsync(executionTask) .ConfigureAwait(false); + var elapsedTime = stopwatch.Elapsed; + + return taskRespondedDuringGrace + ? TimeoutExecutionResult.TimeoutWithTokenRespected(elapsedTime) + : TimeoutExecutionResult.TimeoutWithTokenIgnored(elapsedTime); + } - if (winner == cancelledTcs.Task) + private static async Task DidTaskRespondDuringGracePeriodAsync(Task executionTask) + { + try + { + await executionTask.WaitAsync(GracePeriod, CancellationToken.None).ConfigureAwait(false); + return true; + } + catch (TimeoutException) { - // Determine if it was external cancellation or timeout - if (cancellationToken.IsCancellationRequested) + var taskRespondedDuringGrace = executionTask.IsCompleted; + if (!taskRespondedDuringGrace) { TaskObservation.ObserveFault(executionTask); - throw new OperationCanceledException(cancellationToken); } - // Timeout occurred - the task did NOT respond to the cancellation token - // in time (otherwise executionTask would have completed first with a - // TaskCanceledException). Give it a brief grace period to clean up. - var taskRespondedDuringGrace = false; - try - { - await executionTask.WaitAsync(GracePeriod, CancellationToken.None).ConfigureAwait(false); + return taskRespondedDuringGrace; + } + catch (OperationCanceledException) + { + return true; + } + catch (Exception) + { + return true; + } + } - // Task completed during grace period - it did eventually respond - taskRespondedDuringGrace = true; - } - catch (TimeoutException) + internal sealed class CancellationSignalState(CancellationTokenSource attemptCts) + { + private readonly Lock _lock = new(); + private Task? _executionTask; + private bool _cancellationSignaled; + + public TaskCompletionSource Signal { get; } = new( + TaskCreationOptions.RunContinuationsAsynchronously); + + public void PublishExecutionTask(Task executionTask) + { + bool cancellationSignaled; + lock (_lock) { - // Task still didn't complete - definitely not respecting the token - taskRespondedDuringGrace = false; - TaskObservation.ObserveFault(executionTask); + _executionTask = executionTask; + cancellationSignaled = _cancellationSignaled; } - catch (OperationCanceledException) + + if (cancellationSignaled) { - // Task threw OperationCanceledException/TaskCanceledException from - // finally observing the cancellation token - consider it responsive - taskRespondedDuringGrace = true; + // A completed value or fault existed by the time publication caught up + // with the signal. A cancelled task still belongs to the signal that + // cancelled the attempt token. + Signal.TrySetResult(executionTask.IsCompleted && !executionTask.IsCanceled); } - catch (Exception) + } + + public void SignalCancellation() + { + Task? executionTask; + lock (_lock) { - // Task threw some other exception during grace period - it did respond - // (with an error), so consider it responsive to the cancellation - taskRespondedDuringGrace = true; + _cancellationSignaled = true; + executionTask = _executionTask; } - var elapsedTime = stopwatch.Elapsed; - - // If the task completed exactly when timeout fired (race condition), - // we still consider it a timeout since the deadline was reached - return taskRespondedDuringGrace - ? TimeoutExecutionResult.TimeoutWithTokenRespected(elapsedTime) - : TimeoutExecutionResult.TimeoutWithTokenIgnored(elapsedTime); - } + if (executionTask is not null) + { + // Record ordering before propagating cancellation to the attempt. + Signal.TrySetResult(executionTask.IsCompleted); + } - // Task completed before timeout - try - { - var value = await executionTask.ConfigureAwait(false); - return TimeoutExecutionResult.Success(value, stopwatch.Elapsed); - } - catch (OperationCanceledException) when (timeoutCts.IsCancellationRequested && !cancellationToken.IsCancellationRequested) - { - // The task threw OperationCanceledException/TaskCanceledException in response to - // our timeout cancellation - this means it DID respect the token. - // This can happen when executionTask and cancelledTcs.Task complete at nearly - // the same time, and executionTask wins the race. - return TimeoutExecutionResult.TimeoutWithTokenRespected(stopwatch.Elapsed); + attemptCts.Cancel(); } } } diff --git a/src/ModularPipelines/Options/PipelineOptions.cs b/src/ModularPipelines/Options/PipelineOptions.cs index ac730f6a621..d32bbece021 100644 --- a/src/ModularPipelines/Options/PipelineOptions.cs +++ b/src/ModularPipelines/Options/PipelineOptions.cs @@ -88,8 +88,9 @@ public record PipelineOptions public ExecutionMode ExecutionMode { get; init; } = ExecutionMode.StopOnFirstException; /// - /// Gets the default timeout for modules that do not configure their own timeout. + /// Gets the default per-attempt timeout for modules that do not configure their own timeout. /// Set to to disable the default module timeout. + /// Retry delays do not count toward this timeout. /// public TimeSpan DefaultModuleTimeout { get; init; } = TimeSpan.FromMinutes(30); diff --git a/test/ModularPipelines.UnitTests/Commands/CommandLoggerTests.cs b/test/ModularPipelines.UnitTests/Commands/CommandLoggerTests.cs index c4170482dfe..29813c2739d 100644 --- a/test/ModularPipelines.UnitTests/Commands/CommandLoggerTests.cs +++ b/test/ModularPipelines.UnitTests/Commands/CommandLoggerTests.cs @@ -548,6 +548,56 @@ await Assert.ThrowsAsync(() => await Assert.That(await File.ReadAllTextAsync(file)).DoesNotContain(marker); } + [Test] + public async Task CallerCancellationPreservesCallerTokenIdentity() + { + using var cancellationTokenSource = new CancellationTokenSource(); + var commandContext = await GetService(); + var readyFile = Path.Combine( + TestContext.WorkingDirectory, + $"command-cancellation-{Guid.NewGuid():N}.ready"); + var commandTask = commandContext.ExecuteCommandLineToolAsync( + new PowershellScriptOptions( + "[IO.File]::WriteAllText($env:MP_COMMAND_CANCELLATION_READY_FILE, 'ready'); " + + "Start-Sleep -Seconds 30"), + new CommandExecutionOptions + { + EnvironmentVariables = new Dictionary + { + ["MP_COMMAND_CANCELLATION_READY_FILE"] = readyFile, + }, + GracefulShutdownTimeout = TimeSpan.FromMilliseconds(50), + }, + cancellationToken: cancellationTokenSource.Token); + + try + { + await Assert.That(await WaitUntilAsync( + () => File.Exists(readyFile), + TimeSpan.FromSeconds(30))) + .IsTrue(); + await cancellationTokenSource.CancelAsync(); + var exception = await Assert.ThrowsAsync(() => commandTask); + + await Assert.That( + exception!.CancellationToken == cancellationTokenSource.Token) + .IsTrue(); + } + finally + { + await cancellationTokenSource.CancelAsync(); + try + { + await commandTask; + } + catch (OperationCanceledException) when (cancellationTokenSource.IsCancellationRequested) + { + } + + File.Delete(readyFile); + } + } + [Test] public async Task Deferred_Logging_Failure_After_NonZero_Exit_Preserves_Command_Failure() { diff --git a/test/ModularPipelines.UnitTests/Engine/ModuleExecutionPipelineTests.cs b/test/ModularPipelines.UnitTests/Engine/ModuleExecutionPipelineTests.cs index 56705329466..edf15ce79b0 100644 --- a/test/ModularPipelines.UnitTests/Engine/ModuleExecutionPipelineTests.cs +++ b/test/ModularPipelines.UnitTests/Engine/ModuleExecutionPipelineTests.cs @@ -209,7 +209,7 @@ await Assert.That(async () => await ExecuteAfterPipelineCancellation(module, exe } [Test] - public async Task ExecuteAsync_ClassifiesAlwaysRunElapsedCancellationAsTimeout() + public async Task ExecuteAsync_DoesNotClassifyAlwaysRunElapsedCancellationAsTimeout() { var module = new AlwaysRunElapsedCancellationModule(); var executionContext = new ModuleExecutionContext(module, module.GetType()); @@ -220,7 +220,7 @@ public async Task ExecuteAsync_ClassifiesAlwaysRunElapsedCancellationAsTimeout() await Assert.That(async () => await ExecuteAfterPipelineCancellation(module, executionContext)) .Throws(); - await Assert.That(executionContext.Status).IsEqualTo(Status.TimedOut); + await Assert.That(executionContext.Status).IsEqualTo(Status.Failed); } [Test] diff --git a/test/ModularPipelines.UnitTests/Execution/ModuleTimeoutTests.cs b/test/ModularPipelines.UnitTests/Execution/ModuleTimeoutTests.cs index 490bc2771f3..3b211af24b8 100644 --- a/test/ModularPipelines.UnitTests/Execution/ModuleTimeoutTests.cs +++ b/test/ModularPipelines.UnitTests/Execution/ModuleTimeoutTests.cs @@ -1,6 +1,7 @@ using ModularPipelines.Configuration; using ModularPipelines.Context; using ModularPipelines.Exceptions; +using ModularPipelines.Helpers; using ModularPipelines.Modules; using ModularPipelines.Options; using ModularPipelines.TestHelpers; @@ -196,4 +197,154 @@ public async Task Timeout_Exception_Message_Reports_Grace_Period_Expiry() await Assert.That(timeoutException).IsNotNull(); await Assert.That(timeoutException!.Message).Contains("did not complete within the cancellation grace period"); } + + [Test] + public async Task Timeout_Fault_During_Grace_Period_Counts_As_Response() + { + var result = await TimeoutHelper.ExecuteWithTimeoutAndDetailsAsync( + async cancellationToken => + { + try + { + await Task.Delay(Timeout.InfiniteTimeSpan, cancellationToken); + } + catch (OperationCanceledException) + { + throw new TimeoutException("Inner operation timed out."); + } + + return true; + }, + TimeSpan.FromMilliseconds(10), + CancellationToken.None); + + using (Assert.Multiple()) + { + await Assert.That(result.TimedOut).IsTrue(); + await Assert.That(result.WasCancellationTokenRespected).IsTrue(); + } + } + + [Test] + public async Task Timeout_Does_Not_Claim_Unrelated_Cancellation_Before_Deadline() + { + using var unrelatedCancellation = new CancellationTokenSource(); + unrelatedCancellation.Cancel(); + + var exception = await Assert.ThrowsAsync(async () => + await TimeoutHelper.ExecuteWithTimeoutAndDetailsAsync( + _ => Task.FromCanceled(unrelatedCancellation.Token), + TimeSpan.FromSeconds(1), + CancellationToken.None)); + + await Assert.That(exception!.CancellationToken).IsEqualTo(unrelatedCancellation.Token); + } + + [Test] + public async Task Timeout_Claims_Cancellation_Through_Linked_Attempt_Token() + { + using var unrelatedCancellation = new CancellationTokenSource(); + + var result = await TimeoutHelper.ExecuteWithTimeoutAndDetailsAsync( + timeoutToken => + { + using var linkedCancellation = CancellationTokenSource.CreateLinkedTokenSource( + timeoutToken, + unrelatedCancellation.Token); + timeoutToken.WaitHandle.WaitOne(TimeSpan.FromSeconds(1)); + return Task.FromCanceled(linkedCancellation.Token); + }, + TimeSpan.FromMilliseconds(10), + CancellationToken.None); + + using (Assert.Multiple()) + { + await Assert.That(result.TimedOut).IsTrue(); + await Assert.That(result.WasCancellationTokenRespected).IsTrue(); + } + } + + [Test] + public async Task Fault_Completed_Before_Deadline_Is_Not_Claimed_By_Timeout() + { + var exception = await Assert.ThrowsAsync(async () => + await TimeoutHelper.ExecuteWithTimeoutAndDetailsAsync( + _ => Task.FromException(new TimeoutException("Inner operation timed out.")), + TimeSpan.FromMilliseconds(10), + CancellationToken.None)); + + await Assert.That(exception!.Message).IsEqualTo("Inner operation timed out."); + } + + [Test] + public async Task Completion_With_Asynchronous_Continuations_Before_Deadline_Is_Not_Claimed_By_Timeout() + { + var completion = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + + var execution = TimeoutHelper.ExecuteWithTimeoutAndDetailsAsync( + _ => completion.Task, + TimeSpan.FromSeconds(1), + CancellationToken.None); + + completion.SetResult(true); + + var result = await execution; + + using (Assert.Multiple()) + { + await Assert.That(result.TimedOut).IsFalse(); + await Assert.That(result.Value).IsTrue(); + } + } + + [Test] + public async Task Completed_Execution_Published_After_Deadline_Signal_Wins() + { + using var attemptCancellation = new CancellationTokenSource(); + var signalState = new TimeoutHelper.CancellationSignalState(attemptCancellation); + + signalState.SignalCancellation(); + signalState.PublishExecutionTask(Task.FromResult(true)); + + using (Assert.Multiple()) + { + await Assert.That(await signalState.Signal.Task).IsTrue(); + await Assert.That(attemptCancellation.IsCancellationRequested).IsTrue(); + } + } + + [Test] + public async Task Cancelled_Execution_Published_After_Deadline_Signal_Belongs_To_Deadline() + { + using var attemptCancellation = new CancellationTokenSource(); + var signalState = new TimeoutHelper.CancellationSignalState(attemptCancellation); + + signalState.SignalCancellation(); + signalState.PublishExecutionTask(Task.FromCanceled(attemptCancellation.Token)); + + await Assert.That(await signalState.Signal.Task).IsFalse(); + } + + [Test] + public async Task Timeout_Claims_Tokenless_Cooperative_Cancellation() + { + var result = await TimeoutHelper.ExecuteWithTimeoutAndDetailsAsync( + timeoutToken => + { + timeoutToken.WaitHandle.WaitOne(TimeSpan.FromSeconds(1)); + var completion = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + completion.TrySetCanceled(); + return completion.Task; + }, + TimeSpan.FromMilliseconds(10), + CancellationToken.None); + + using (Assert.Multiple()) + { + await Assert.That(result.TimedOut).IsTrue(); + await Assert.That(result.WasCancellationTokenRespected).IsTrue(); + } + } } diff --git a/test/ModularPipelines.UnitTests/Execution/RetryTests.cs b/test/ModularPipelines.UnitTests/Execution/RetryTests.cs index 327caffa354..276c4543da5 100644 --- a/test/ModularPipelines.UnitTests/Execution/RetryTests.cs +++ b/test/ModularPipelines.UnitTests/Execution/RetryTests.cs @@ -3,6 +3,7 @@ using ModularPipelines.Configuration; using ModularPipelines.Context; using ModularPipelines.Engine; +using ModularPipelines.Enums; using ModularPipelines.Exceptions; using ModularPipelines.Extensions; using ModularPipelines.Models; @@ -157,12 +158,16 @@ protected internal override Task ExecuteAsync(IModuleContext context, Canc private class CancellableModuleWithTimeout : Module { + internal int ExecutionCount; + protected override ModuleConfiguration Configure() => ModuleConfiguration.Create() .WithTimeout(TimeSpan.FromMilliseconds(ModuleTimeoutMs)) + .WithRetry(DefaultRetryCount, TimeSpan.Zero) .Build(); protected internal override async Task ExecuteAsync(IModuleContext context, CancellationToken cancellationToken) { + ExecutionCount++; await Task.Delay(Timeout.InfiniteTimeSpan, cancellationToken); return true; } @@ -293,14 +298,90 @@ public async Task When_Error_And_Zero_Retry_Count_Then_Do_Not_Retry() } } + private class NonCancellableModuleWithTimeout : Module + { + private readonly TaskCompletionSource _completion = + new(TaskCreationOptions.RunContinuationsAsynchronously); + + internal int ExecutionCount; + + protected override ModuleConfiguration Configure() => ModuleConfiguration.Create() + .WithTimeout(TimeSpan.FromMilliseconds(50)) + .WithRetry(DefaultRetryCount, TimeSpan.FromMinutes(1)) + .Build(); + + protected internal override async Task ExecuteAsync(IModuleContext context, CancellationToken cancellationToken) + { + ExecutionCount++; + return await _completion.Task; + } + + internal void Complete() => _completion.TrySetResult(true); + } + + private class NonCancellableModuleWithAdvancedShield : Module + { + private readonly TaskCompletionSource _completion = + new(TaskCreationOptions.RunContinuationsAsynchronously); + + internal int ExecutionCount; + internal int ObservedTimeoutCount; + + protected override ModuleConfiguration Configure() => ModuleConfiguration.Create() + .WithTimeout(TimeSpan.FromMilliseconds(50)) + .Advanced + .WithShield(Shield + .When(_ => + { + ObservedTimeoutCount++; + return true; + }) + .Retry(DefaultRetryCount, Backoff.None)) + .Build(); + + protected internal override async Task ExecuteAsync( + IModuleContext context, + CancellationToken cancellationToken) + { + ExecutionCount++; + return await _completion.Task; + } + + internal void Complete() => _completion.TrySetResult(true); + } + + private class CancelledDuringRetryModule : Module + { + private readonly TaskCompletionSource _secondAttemptStarted = + new(TaskCreationOptions.RunContinuationsAsynchronously); + + internal Task SecondAttemptStarted => _secondAttemptStarted.Task; + + internal int ExecutionCount; + + protected override ModuleConfiguration Configure() => ModuleConfiguration.Create() + .WithTimeout(TimeSpan.FromSeconds(2)) + .WithRetry(1, TimeSpan.FromMilliseconds(250)) + .Build(); + + protected internal override async Task ExecuteAsync(IModuleContext context, CancellationToken cancellationToken) + { + ExecutionCount++; + if (ExecutionCount == 1) + { + throw new RetryableTestException(); + } + + _secondAttemptStarted.TrySetResult(); + await Task.Delay(Timeout.InfiniteTimeSpan, cancellationToken); + return true; + } + } + [Test] - public async Task When_Retry_With_Timeout_Then_Honour_Overall_Timeout() + public async Task When_Retry_Backoff_Exceeds_Timeout_Then_All_Attempts_Run() { var host = await TestPipelineBuilder.Create() - .ConfigurePipelineOptions((_, options) => options with - { - DefaultRetryCount = DefaultRetryCount, - }) .AddModule() .BuildAsync(); @@ -308,27 +389,111 @@ public async Task When_Retry_With_Timeout_Then_Honour_Overall_Timeout() var module = host.Services.GetServices().OfType().Single(); var timeoutException = moduleFailedException?.InnerException as ModuleTimeoutException; - await Assert.That(timeoutException).IsNotNull(); using (Assert.Multiple()) { - await Assert.That(module.ExecutionCount).IsEqualTo(ExpectedSingleExecutionCount); - await Assert.That(timeoutException!.WasCancellationTokenRespected).IsFalse(); + await Assert.That(module.ExecutionCount).IsEqualTo(ExpectedExecutionCountAfterRetries); + await Assert.That(timeoutException).IsNull(); } } [Test] public async Task When_Retry_Timeouts_During_Module_Then_Report_Token_Respected() { - var moduleFailedException = await Assert.ThrowsAsync(async () => await TestPipelineBuilder.Create() - .ConfigurePipelineOptions((_, options) => options with - { - DefaultRetryCount = DefaultRetryCount, - }) + var host = await TestPipelineBuilder.Create() .AddModule() - .ExecutePipelineAsync()); + .BuildAsync(); + var moduleFailedException = await Assert.ThrowsAsync(() => host.RunAsync()); + + var module = host.Services.GetServices().OfType().Single(); var timeoutException = moduleFailedException?.InnerException as ModuleTimeoutException; - await Assert.That(timeoutException).IsNotNull(); - await Assert.That(timeoutException!.WasCancellationTokenRespected).IsTrue(); + using (Assert.Multiple()) + { + await Assert.That(module.ExecutionCount).IsEqualTo(ExpectedExecutionCountAfterRetries); + await Assert.That(timeoutException).IsNotNull(); + await Assert.That(timeoutException!.WasCancellationTokenRespected).IsTrue(); + } + } + + [Test] + public async Task When_Timed_Out_Attempt_Remains_Active_Then_Do_Not_Retry() + { + var host = await TestPipelineBuilder.Create() + .AddModule() + .BuildAsync(); + var module = host.Services.GetServices().OfType().Single(); + + try + { + var moduleFailedException = await Assert.ThrowsAsync( + () => host.RunAsync().WaitAsync(TimeSpan.FromSeconds(10))); + var timeoutException = moduleFailedException?.InnerException as ModuleTimeoutException; + + using (Assert.Multiple()) + { + await Assert.That(module.ExecutionCount).IsEqualTo(ExpectedSingleExecutionCount); + await Assert.That(timeoutException).IsNotNull(); + await Assert.That(timeoutException!.WasCancellationTokenRespected).IsFalse(); + } + } + finally + { + module.Complete(); + } + } + + [Test] + public async Task When_Advanced_Shield_Sees_Abandoned_Attempt_Then_Do_Not_Reenter_Module() + { + var host = await TestPipelineBuilder.Create() + .AddModule() + .BuildAsync(); + var module = host.Services.GetServices() + .OfType() + .Single(); + + try + { + var moduleFailedException = await Assert.ThrowsAsync( + () => host.RunAsync().WaitAsync(TimeSpan.FromSeconds(10))); + var timeoutException = moduleFailedException?.InnerException as ModuleTimeoutException; + + using (Assert.Multiple()) + { + await Assert.That(module.ExecutionCount).IsEqualTo(ExpectedSingleExecutionCount); + await Assert.That(module.ObservedTimeoutCount).IsEqualTo(ExpectedSingleExecutionCount); + await Assert.That(timeoutException).IsNotNull(); + await Assert.That(timeoutException!.WasCancellationTokenRespected).IsFalse(); + } + } + finally + { + module.Complete(); + } + } + + [Test] + public async Task When_Cancelled_During_Later_Attempt_Then_Report_PipelineTerminated() + { + using var cancellationTokenSource = new CancellationTokenSource(); + var host = await TestPipelineBuilder.Create() + .AddModule() + .BuildAsync(); + var module = host.Services.GetServices().OfType().Single(); + var resultRegistry = host.Services.GetRequiredService(); + var pipelineTask = host.RunAsync(cancellationTokenSource.Token); + + await module.SecondAttemptStarted.WaitAsync(TimeSpan.FromSeconds(5)); + cancellationTokenSource.Cancel(); + + await pipelineTask; + + var result = resultRegistry.GetResult(typeof(CancelledDuringRetryModule)); + using (Assert.Multiple()) + { + await Assert.That(module.ExecutionCount).IsEqualTo(2); + await Assert.That(result).IsNotNull(); + await Assert.That(result!.ModuleStatus).IsEqualTo(Status.PipelineTerminated); + } } }