From c555a40b1bbb7853c6737c27771115505c72c097 Mon Sep 17 00:00:00 2001 From: Tom Longhurst <30480171+thomhurst@users.noreply.github.com> Date: Mon, 10 Aug 2026 07:30:52 +0100 Subject: [PATCH 01/12] fix(engine): timeout each retry attempt Give every retry its own timeout budget while preventing abandoned attempts from re-entering the module. Cancel retry backoff after a non-cooperative timeout and classify only typed timeout failures as timed out. --- docs/docs/how-to/retry-policy.md | 6 + docs/docs/how-to/timeouts.md | 9 +- .../Configuration/ModuleConfiguration.cs | 4 +- .../ModuleConfigurationBuilder.cs | 8 +- .../Engine/ModuleExecutionPipeline.cs | 174 +++++++++++------- .../Exceptions/ModuleTimeoutException.cs | 5 +- src/ModularPipelines/Helpers/TimeoutHelper.cs | 149 ++++++++------- .../Options/PipelineOptions.cs | 3 +- .../Engine/ModuleExecutionPipelineTests.cs | 4 +- .../Execution/ModuleTimeoutTests.cs | 28 +++ .../Execution/RetryTests.cs | 145 +++++++++++++-- 11 files changed, 369 insertions(+), 166 deletions(-) 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/Engine/ModuleExecutionPipeline.cs b/src/ModularPipelines/Engine/ModuleExecutionPipeline.cs index 8997f5e779a..53350dc27e8 100644 --- a/src/ModularPipelines/Engine/ModuleExecutionPipeline.cs +++ b/src/ModularPipelines/Engine/ModuleExecutionPipeline.cs @@ -472,80 +472,111 @@ 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; + using var retryCancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); // Get resilience shield if applicable var resilienceShield = GetResilienceShield(config, moduleContext); - var moduleAttemptCount = 0; - var moduleAttemptRespondedToCancellation = 0; - - async Task ExecuteModuleAttempt(CancellationToken ct) - { - Interlocked.Increment(ref moduleAttemptCount); - try - { - return await module.ExecuteAsync(moduleContext, ct).ConfigureAwait(false); - } - finally - { - if (ct.IsCancellationRequested) - { - Volatile.Write(ref moduleAttemptRespondedToCancellation, 1); - } - } - } + var policyExecutionState = new PolicyExecutionState(); - // 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; + // 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); - // Use TimeoutHelper with detailed results to get information about token cooperation - TimeoutExecutionResult timeoutResult; + T result; try { - timeoutResult = await TimeoutHelper.ExecuteWithTimeoutAndDetailsAsync( - executeFunc, - timeout == TimeSpan.Zero ? null : timeout, - cancellationToken, - $"Module {executionContext.ModuleType.Name} timed out after {timeout}").ConfigureAwait(false); + result = resilienceShield != null + ? await resilienceShield.ExecuteAsync( + async shieldToken => await ExecuteModuleAttempt(shieldToken).ConfigureAwait(false), + retryCancellationTokenSource.Token).ConfigureAwait(false) + : await ExecuteModuleAttempt(cancellationToken).ConfigureAwait(false); + } + catch (OperationCanceledException) when (policyExecutionState.AbandonedAttemptTimeout is not null + && !cancellationToken.IsCancellationRequested) + { + throw 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); + throw 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; + } + + 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) + { + 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; + // Let wrapped policies observe the failure, but cancel their retry delay because + // re-entering this module while the abandoned attempt is active is unsafe. + retryCancellationTokenSource.Cancel(); } - return timeoutResult.Value!; + throw timeoutException; } private TimeSpan GetTimeout(ModuleConfiguration config) @@ -680,7 +711,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 +724,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 +776,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 TaskCanceledException or OperationCanceledException or ModuleTimeoutException + && _engineCancellationToken.IsCancelled; } private void CancelPipelineAndThrow( @@ -841,6 +862,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..565e2e873a4 100644 --- a/src/ModularPipelines/Helpers/TimeoutHelper.cs +++ b/src/ModularPipelines/Helpers/TimeoutHelper.cs @@ -90,31 +90,8 @@ public static async Task> ExecuteWithTimeoutAndDetails // Fast path: no timeout specified if (!timeout.HasValue || timeout.Value == TimeSpan.Zero) { - var task = taskFactory(cancellationToken); - - // 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); - } - - // 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 fastPathWinner = await Task.WhenAny(task, tcs.Task).ConfigureAwait(false); - if (fastPathWinner != task) - { - TaskObservation.ObserveFault(task); - } - - var winningResult = await fastPathWinner.ConfigureAwait(false); - return TimeoutExecutionResult.Success(winningResult, stopwatch.Elapsed); + return await ExecuteWithoutTimeoutAsync(taskFactory, cancellationToken, stopwatch) + .ConfigureAwait(false); } // Timeout path: create linked token so task can observe both timeout @@ -142,50 +119,8 @@ public static async Task> ExecuteWithTimeoutAndDetails if (winner == cancelledTcs.Task) { - // Determine if it was external cancellation or timeout - if (cancellationToken.IsCancellationRequested) - { - 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); - - // Task completed during grace period - it did eventually respond - taskRespondedDuringGrace = true; - } - catch (TimeoutException) - { - // Task still didn't complete - definitely not respecting the token - taskRespondedDuringGrace = false; - TaskObservation.ObserveFault(executionTask); - } - catch (OperationCanceledException) - { - // Task threw OperationCanceledException/TaskCanceledException from - // finally observing the cancellation token - consider it responsive - taskRespondedDuringGrace = true; - } - catch (Exception) - { - // Task threw some other exception during grace period - it did respond - // (with an error), so consider it responsive to the cancellation - taskRespondedDuringGrace = true; - } - - 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); + return await CreateTimeoutResultAsync(executionTask, cancellationToken, stopwatch) + .ConfigureAwait(false); } // Task completed before timeout @@ -203,4 +138,80 @@ public static async Task> ExecuteWithTimeoutAndDetails return TimeoutExecutionResult.TimeoutWithTokenRespected(stopwatch.Elapsed); } } + + 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 = cancellationToken.Register( + static state => ((TaskCompletionSource) state!).TrySetCanceled(), + cancellationTaskSource); + + var winner = await Task.WhenAny(task, cancellationTaskSource.Task).ConfigureAwait(false); + if (winner != task) + { + TaskObservation.ObserveFault(task); + } + + var resultValue = await winner.ConfigureAwait(false); + return TimeoutExecutionResult.Success(resultValue, stopwatch.Elapsed); + } + + 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); + } + + private static async Task DidTaskRespondDuringGracePeriodAsync(Task executionTask) + { + try + { + await executionTask.WaitAsync(GracePeriod, CancellationToken.None).ConfigureAwait(false); + return true; + } + catch (TimeoutException) + { + var taskRespondedDuringGrace = executionTask.IsCompleted; + if (!taskRespondedDuringGrace) + { + TaskObservation.ObserveFault(executionTask); + } + + return taskRespondedDuringGrace; + } + catch (OperationCanceledException) + { + return true; + } + catch (Exception) + { + return true; + } + } } 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/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..d26018404fc 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,31 @@ 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(); + } + } } diff --git a/test/ModularPipelines.UnitTests/Execution/RetryTests.cs b/test/ModularPipelines.UnitTests/Execution/RetryTests.cs index 327caffa354..108fcb71b2a 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,67 @@ 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; + + internal int RetryCallbackCount; + + protected override ModuleConfiguration Configure() => ModuleConfiguration.Create() + .WithTimeout(TimeSpan.FromMilliseconds(50)) + .Advanced + .WithRetryPolicy(Policy + .Handle() + .WaitAndRetryAsync( + DefaultRetryCount, + _ => TimeSpan.FromMinutes(1), + (_, _, _, _) => RetryCallbackCount++)) + .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 +366,82 @@ 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(module.RetryCallbackCount).IsEqualTo(1); + 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); + } } } From 5b8a7d66022f3c35de28e9118529ae9aeb88b35a Mon Sep 17 00:00:00 2001 From: Tom Longhurst <30480171+thomhurst@users.noreply.github.com> Date: Mon, 10 Aug 2026 11:04:04 +0100 Subject: [PATCH 02/12] fix: attribute timeout cancellation by token --- src/ModularPipelines/Helpers/TimeoutHelper.cs | 5 ++++- .../Execution/ModuleTimeoutTests.cs | 19 +++++++++++++++++++ 2 files changed, 23 insertions(+), 1 deletion(-) diff --git a/src/ModularPipelines/Helpers/TimeoutHelper.cs b/src/ModularPipelines/Helpers/TimeoutHelper.cs index 565e2e873a4..627944d713c 100644 --- a/src/ModularPipelines/Helpers/TimeoutHelper.cs +++ b/src/ModularPipelines/Helpers/TimeoutHelper.cs @@ -129,7 +129,10 @@ public static async Task> ExecuteWithTimeoutAndDetails var value = await executionTask.ConfigureAwait(false); return TimeoutExecutionResult.Success(value, stopwatch.Elapsed); } - catch (OperationCanceledException) when (timeoutCts.IsCancellationRequested && !cancellationToken.IsCancellationRequested) + catch (OperationCanceledException exception) when ( + exception.CancellationToken == timeoutCts.Token + && timeoutCts.IsCancellationRequested + && !cancellationToken.IsCancellationRequested) { // The task threw OperationCanceledException/TaskCanceledException in response to // our timeout cancellation - this means it DID respect the token. diff --git a/test/ModularPipelines.UnitTests/Execution/ModuleTimeoutTests.cs b/test/ModularPipelines.UnitTests/Execution/ModuleTimeoutTests.cs index d26018404fc..fb61615129a 100644 --- a/test/ModularPipelines.UnitTests/Execution/ModuleTimeoutTests.cs +++ b/test/ModularPipelines.UnitTests/Execution/ModuleTimeoutTests.cs @@ -224,4 +224,23 @@ public async Task Timeout_Fault_During_Grace_Period_Counts_As_Response() await Assert.That(result.WasCancellationTokenRespected).IsTrue(); } } + + [Test] + public async Task Timeout_Does_Not_Claim_Unrelated_Cancellation_When_Deadline_Elapses() + { + using var unrelatedCancellation = new CancellationTokenSource(); + unrelatedCancellation.Cancel(); + + var exception = await Assert.ThrowsAsync(async () => + await TimeoutHelper.ExecuteWithTimeoutAndDetailsAsync( + timeoutToken => + { + timeoutToken.WaitHandle.WaitOne(TimeSpan.FromSeconds(1)); + return Task.FromCanceled(unrelatedCancellation.Token); + }, + TimeSpan.FromMilliseconds(10), + CancellationToken.None)); + + await Assert.That(exception!.CancellationToken).IsEqualTo(unrelatedCancellation.Token); + } } From 347a2ec9fa34f74c7dc6e3ebf16f651ac482e759 Mon Sep 17 00:00:00 2001 From: Tom Longhurst <30480171+thomhurst@users.noreply.github.com> Date: Mon, 10 Aug 2026 12:43:00 +0100 Subject: [PATCH 03/12] fix(timeout): classify tokenless cancellation Attribute tokenless cooperative cancellation to the deadline only when timeout state was already observed as the execution task won. --- src/ModularPipelines/Helpers/TimeoutHelper.cs | 11 ++++++---- .../Execution/ModuleTimeoutTests.cs | 22 +++++++++++++++++++ 2 files changed, 29 insertions(+), 4 deletions(-) diff --git a/src/ModularPipelines/Helpers/TimeoutHelper.cs b/src/ModularPipelines/Helpers/TimeoutHelper.cs index 627944d713c..47c4ae13ac3 100644 --- a/src/ModularPipelines/Helpers/TimeoutHelper.cs +++ b/src/ModularPipelines/Helpers/TimeoutHelper.cs @@ -123,16 +123,19 @@ public static async Task> ExecuteWithTimeoutAndDetails .ConfigureAwait(false); } - // Task completed before timeout + var timeoutElapsedWhenExecutionCompleted = timeoutCts.IsCancellationRequested + && !cancellationToken.IsCancellationRequested; + + // The execution task won the completion race. try { var value = await executionTask.ConfigureAwait(false); return TimeoutExecutionResult.Success(value, stopwatch.Elapsed); } catch (OperationCanceledException exception) when ( - exception.CancellationToken == timeoutCts.Token - && timeoutCts.IsCancellationRequested - && !cancellationToken.IsCancellationRequested) + timeoutElapsedWhenExecutionCompleted + && (exception.CancellationToken == timeoutCts.Token + || !exception.CancellationToken.CanBeCanceled)) { // The task threw OperationCanceledException/TaskCanceledException in response to // our timeout cancellation - this means it DID respect the token. diff --git a/test/ModularPipelines.UnitTests/Execution/ModuleTimeoutTests.cs b/test/ModularPipelines.UnitTests/Execution/ModuleTimeoutTests.cs index fb61615129a..fe3632af5b5 100644 --- a/test/ModularPipelines.UnitTests/Execution/ModuleTimeoutTests.cs +++ b/test/ModularPipelines.UnitTests/Execution/ModuleTimeoutTests.cs @@ -243,4 +243,26 @@ await TimeoutHelper.ExecuteWithTimeoutAndDetailsAsync( await Assert.That(exception!.CancellationToken).IsEqualTo(unrelatedCancellation.Token); } + + [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(); + } + } } From d22f03e20cc3da7c69a8002dc6d3cb59cd815e64 Mon Sep 17 00:00:00 2001 From: Tom Longhurst <30480171+thomhurst@users.noreply.github.com> Date: Mon, 10 Aug 2026 15:32:44 +0100 Subject: [PATCH 04/12] fix(commands): preserve caller cancellation Normalize framework-owned linked command cancellation back to the caller token so module timeout attribution remains exact without claiming unrelated cancellations. --- src/ModularPipelines/Context/Command.cs | 9 ++++ .../Commands/CommandLoggerTests.cs | 50 +++++++++++++++++++ 2 files changed, 59 insertions(+) diff --git a/src/ModularPipelines/Context/Command.cs b/src/ModularPipelines/Context/Command.cs index c251ea54961..111978134d3 100644 --- a/src/ModularPipelines/Context/Command.cs +++ b/src/ModularPipelines/Context/Command.cs @@ -555,6 +555,15 @@ await WaitForForcefulCancellationAsync( if (ShouldPreserveCallerCancellation(e, failure, callerCancellationToken)) { + if (e is OperationCanceledException cancellationException + && cancellationException.CancellationToken != callerCancellationToken) + { + throw new OperationCanceledException( + cancellationException.Message, + cancellationException, + callerCancellationToken); + } + throw; } 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() { From c5766eada1ac56b14d2e2a3e6c419adcd5308b19 Mon Sep 17 00:00:00 2001 From: Tom Longhurst <30480171+thomhurst@users.noreply.github.com> Date: Mon, 10 Aug 2026 16:46:07 +0100 Subject: [PATCH 05/12] fix(timeout): harden retry cancellation --- src/ModularPipelines/Context/Command.cs | 46 +++++++++++++------ .../Engine/ModuleExecutionPipeline.cs | 27 ++++++++--- 2 files changed, 54 insertions(+), 19 deletions(-) diff --git a/src/ModularPipelines/Context/Command.cs b/src/ModularPipelines/Context/Command.cs index 111978134d3..5587558bb05 100644 --- a/src/ModularPipelines/Context/Command.cs +++ b/src/ModularPipelines/Context/Command.cs @@ -553,19 +553,7 @@ await WaitForForcefulCancellationAsync( command.WorkingDirPath)); var failure = loggingFailures.CombineWith(e); - if (ShouldPreserveCallerCancellation(e, failure, callerCancellationToken)) - { - if (e is OperationCanceledException cancellationException - && cancellationException.CancellationToken != callerCancellationToken) - { - throw new OperationCanceledException( - cancellationException.Message, - cancellationException, - callerCancellationToken); - } - - throw; - } + ThrowCallerCancellationIfRequired(e, failure, callerCancellationToken); throw CreateExecutionFailure( e, @@ -636,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 53350dc27e8..542bac971d7 100644 --- a/src/ModularPipelines/Engine/ModuleExecutionPipeline.cs +++ b/src/ModularPipelines/Engine/ModuleExecutionPipeline.cs @@ -475,10 +475,12 @@ private async Task ExecuteWithPolicies( LogTimeoutConfiguration(config, timeout, moduleContext.Logger); var cancellationToken = executionContext.ModuleCancellationTokenSource.Token; - using var retryCancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); // Get resilience shield if applicable var resilienceShield = GetResilienceShield(config, moduleContext); + 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 @@ -498,13 +500,13 @@ Task ExecuteModuleAttempt(CancellationToken ct) => ExecuteModuleAttemptAsync( result = resilienceShield != null ? await resilienceShield.ExecuteAsync( async shieldToken => await ExecuteModuleAttempt(shieldToken).ConfigureAwait(false), - retryCancellationTokenSource.Token).ConfigureAwait(false) + retryCancellationTokenSource!.Token).ConfigureAwait(false) : await ExecuteModuleAttempt(cancellationToken).ConfigureAwait(false); } catch (OperationCanceledException) when (policyExecutionState.AbandonedAttemptTimeout is not null && !cancellationToken.IsCancellationRequested) { - throw policyExecutionState.AbandonedAttemptTimeout; + return ThrowPreservingStack(policyExecutionState.AbandonedAttemptTimeout); } finally { @@ -515,7 +517,7 @@ Task ExecuteModuleAttempt(CancellationToken ct) => ExecuteModuleAttemptAsync( if (policyExecutionState.AbandonedAttemptTimeout is { } abandonedAttemptTimeout) { - throw abandonedAttemptTimeout; + return ThrowPreservingStack(abandonedAttemptTimeout); } return result; @@ -545,10 +547,15 @@ private static async Task ExecuteModuleAttemptAsync( ModuleExecutionContext executionContext, IModuleContext moduleContext, TimeSpan timeout, - CancellationTokenSource retryCancellationTokenSource, + CancellationTokenSource? retryCancellationTokenSource, PolicyExecutionState policyExecutionState, CancellationToken cancellationToken) { + if (policyExecutionState.AbandonedAttemptTimeout is { } abandonedAttemptTimeout) + { + return ThrowPreservingStack(abandonedAttemptTimeout); + } + policyExecutionState.RecordAttempt(); var timeoutResult = await TimeoutHelper.ExecuteWithTimeoutAndDetailsAsync( @@ -573,12 +580,20 @@ private static async Task ExecuteModuleAttemptAsync( policyExecutionState.AbandonedAttemptTimeout = timeoutException; // Let wrapped policies observe the failure, but cancel their retry delay because // re-entering this module while the abandoned attempt is active is unsafe. - retryCancellationTokenSource.Cancel(); + 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) { return config.Timeout ?? _pipelineOptions.Value.DefaultModuleTimeout; From 40e945f6213209ce406dd6830ae874a538b383a0 Mon Sep 17 00:00:00 2001 From: Tom Longhurst <30480171+thomhurst@users.noreply.github.com> Date: Mon, 10 Aug 2026 17:05:10 +0100 Subject: [PATCH 06/12] fix(timeout): accept linked cancellation --- src/ModularPipelines/Helpers/TimeoutHelper.cs | 13 +++---- .../Execution/ModuleTimeoutTests.cs | 34 +++++++++++++++---- 2 files changed, 32 insertions(+), 15 deletions(-) diff --git a/src/ModularPipelines/Helpers/TimeoutHelper.cs b/src/ModularPipelines/Helpers/TimeoutHelper.cs index 47c4ae13ac3..a5ed825c120 100644 --- a/src/ModularPipelines/Helpers/TimeoutHelper.cs +++ b/src/ModularPipelines/Helpers/TimeoutHelper.cs @@ -132,15 +132,12 @@ public static async Task> ExecuteWithTimeoutAndDetails var value = await executionTask.ConfigureAwait(false); return TimeoutExecutionResult.Success(value, stopwatch.Elapsed); } - catch (OperationCanceledException exception) when ( - timeoutElapsedWhenExecutionCompleted - && (exception.CancellationToken == timeoutCts.Token - || !exception.CancellationToken.CanBeCanceled)) + catch (OperationCanceledException) when (timeoutElapsedWhenExecutionCompleted) { - // 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. + // The deadline elapsed before the completed task was observed. Any cancellation + // from that task therefore counts as a response, including cancellation through + // a token linked to the supplied timeout token. This can happen when executionTask + // and cancelledTcs.Task complete at nearly the same time and executionTask wins. return TimeoutExecutionResult.TimeoutWithTokenRespected(stopwatch.Elapsed); } } diff --git a/test/ModularPipelines.UnitTests/Execution/ModuleTimeoutTests.cs b/test/ModularPipelines.UnitTests/Execution/ModuleTimeoutTests.cs index fe3632af5b5..4dd4e5616fd 100644 --- a/test/ModularPipelines.UnitTests/Execution/ModuleTimeoutTests.cs +++ b/test/ModularPipelines.UnitTests/Execution/ModuleTimeoutTests.cs @@ -226,24 +226,44 @@ public async Task Timeout_Fault_During_Grace_Period_Counts_As_Response() } [Test] - public async Task Timeout_Does_Not_Claim_Unrelated_Cancellation_When_Deadline_Elapses() + 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( - timeoutToken => - { - timeoutToken.WaitHandle.WaitOne(TimeSpan.FromSeconds(1)); - return Task.FromCanceled(unrelatedCancellation.Token); - }, - TimeSpan.FromMilliseconds(10), + _ => 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 Timeout_Claims_Tokenless_Cooperative_Cancellation() { From 6aefa2cdffe00a31802e8f3f9ca475ebbd3197cc Mon Sep 17 00:00:00 2001 From: Tom Longhurst <30480171+thomhurst@users.noreply.github.com> Date: Mon, 10 Aug 2026 18:05:38 +0100 Subject: [PATCH 07/12] fix: classify timeout completion race Record deadline state synchronously when execution completes so faults raised during cancellation count as cooperative timeout responses. --- src/ModularPipelines/Helpers/TimeoutHelper.cs | 35 +++++++++++-------- .../Execution/ModuleTimeoutTests.cs | 12 +++++++ 2 files changed, 32 insertions(+), 15 deletions(-) diff --git a/src/ModularPipelines/Helpers/TimeoutHelper.cs b/src/ModularPipelines/Helpers/TimeoutHelper.cs index a5ed825c120..e8afe792c26 100644 --- a/src/ModularPipelines/Helpers/TimeoutHelper.cs +++ b/src/ModularPipelines/Helpers/TimeoutHelper.cs @@ -113,8 +113,20 @@ public static async Task> ExecuteWithTimeoutAndDetails timeoutCts.CancelAfter(timeout.Value); var executionTask = taskFactory(timeoutCts.Token); + var executionTimedOutTask = executionTask.ContinueWith( + static (_, state) => + { + var (timeoutToken, externalToken) = + ((CancellationToken TimeoutToken, CancellationToken ExternalToken)) state!; + return timeoutToken.IsCancellationRequested + && !externalToken.IsCancellationRequested; + }, + (timeoutCts.Token, cancellationToken), + CancellationToken.None, + TaskContinuationOptions.ExecuteSynchronously, + TaskScheduler.Default); - var winner = await Task.WhenAny(executionTask, cancelledTcs.Task) + var winner = await Task.WhenAny(executionTimedOutTask, cancelledTcs.Task) .ConfigureAwait(false); if (winner == cancelledTcs.Task) @@ -123,23 +135,16 @@ public static async Task> ExecuteWithTimeoutAndDetails .ConfigureAwait(false); } - var timeoutElapsedWhenExecutionCompleted = timeoutCts.IsCancellationRequested - && !cancellationToken.IsCancellationRequested; - - // The execution task won the completion race. - try + if (await executionTimedOutTask.ConfigureAwait(false)) { - var value = await executionTask.ConfigureAwait(false); - return TimeoutExecutionResult.Success(value, stopwatch.Elapsed); - } - catch (OperationCanceledException) when (timeoutElapsedWhenExecutionCompleted) - { - // The deadline elapsed before the completed task was observed. Any cancellation - // from that task therefore counts as a response, including cancellation through - // a token linked to the supplied timeout token. This can happen when executionTask - // and cancelledTcs.Task complete at nearly the same time and executionTask wins. + // Any completion after the deadline counts as a response, including a fault + // raised while handling cancellation. The synchronous continuation records the + // ordering at task completion instead of when the winner is later observed. return TimeoutExecutionResult.TimeoutWithTokenRespected(stopwatch.Elapsed); } + + var value = await executionTask.ConfigureAwait(false); + return TimeoutExecutionResult.Success(value, stopwatch.Elapsed); } private static async Task> ExecuteWithoutTimeoutAsync( diff --git a/test/ModularPipelines.UnitTests/Execution/ModuleTimeoutTests.cs b/test/ModularPipelines.UnitTests/Execution/ModuleTimeoutTests.cs index 4dd4e5616fd..90d45a31ee7 100644 --- a/test/ModularPipelines.UnitTests/Execution/ModuleTimeoutTests.cs +++ b/test/ModularPipelines.UnitTests/Execution/ModuleTimeoutTests.cs @@ -264,6 +264,18 @@ public async Task Timeout_Claims_Cancellation_Through_Linked_Attempt_Token() } } + [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 Timeout_Claims_Tokenless_Cooperative_Cancellation() { From 3a9df99b76598768224a58dda3954f1cebe92e1b Mon Sep 17 00:00:00 2001 From: Tom Longhurst <30480171+thomhurst@users.noreply.github.com> Date: Mon, 10 Aug 2026 18:23:37 +0100 Subject: [PATCH 08/12] fix: capture timeout completion order --- src/ModularPipelines/Helpers/TimeoutHelper.cs | 80 +++++++++++-------- .../Execution/ModuleTimeoutTests.cs | 22 +++++ 2 files changed, 67 insertions(+), 35 deletions(-) diff --git a/src/ModularPipelines/Helpers/TimeoutHelper.cs b/src/ModularPipelines/Helpers/TimeoutHelper.cs index e8afe792c26..6bbb1c7df42 100644 --- a/src/ModularPipelines/Helpers/TimeoutHelper.cs +++ b/src/ModularPipelines/Helpers/TimeoutHelper.cs @@ -96,51 +96,48 @@ public static async Task> ExecuteWithTimeoutAndDetails // Timeout path: create linked token so task can observe both timeout // and external cancellation. - using var timeoutCts = CancellationTokenSource - .CreateLinkedTokenSource(cancellationToken); - - // 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( - TaskCreationOptions.RunContinuationsAsynchronously); - using var registration = timeoutCts.Token.Register( - static state => ((TaskCompletionSource) state!) - .TrySetCanceled(), - cancelledTcs); + using var deadlineCts = new CancellationTokenSource(); + using var attemptCts = CancellationTokenSource.CreateLinkedTokenSource( + cancellationToken, + deadlineCts.Token); + + // Register after linking the attempt token. Cancellation callbacks run in + // LIFO order, so these observers record whether execution had already + // completed before cancellation propagates to the task. + var deadlineState = new CancellationSignalState(); + var externalCancellationState = new CancellationSignalState(); + using var deadlineRegistration = deadlineCts.Token.Register( + static state => ((CancellationSignalState) state!).SignalCancellation(), + deadlineState); + using var externalCancellationRegistration = cancellationToken.Register( + static state => ((CancellationSignalState) state!).SignalCancellation(), + externalCancellationState); // Now schedule the timeout - registration is guaranteed to catch it - timeoutCts.CancelAfter(timeout.Value); + deadlineCts.CancelAfter(timeout.Value); - var executionTask = taskFactory(timeoutCts.Token); - var executionTimedOutTask = executionTask.ContinueWith( - static (_, state) => - { - var (timeoutToken, externalToken) = - ((CancellationToken TimeoutToken, CancellationToken ExternalToken)) state!; - return timeoutToken.IsCancellationRequested - && !externalToken.IsCancellationRequested; - }, - (timeoutCts.Token, cancellationToken), - CancellationToken.None, - TaskContinuationOptions.ExecuteSynchronously, - TaskScheduler.Default); + var executionTask = taskFactory(attemptCts.Token); + Volatile.Write(ref deadlineState.ExecutionTask, executionTask); + Volatile.Write(ref externalCancellationState.ExecutionTask, executionTask); - var winner = await Task.WhenAny(executionTimedOutTask, cancelledTcs.Task) + await Task.WhenAny( + executionTask, + deadlineState.Signal.Task, + externalCancellationState.Signal.Task) .ConfigureAwait(false); - if (winner == cancelledTcs.Task) + if (externalCancellationState.Signal.Task.IsCompletedSuccessfully + && !await externalCancellationState.Signal.Task.ConfigureAwait(false)) { - return await CreateTimeoutResultAsync(executionTask, cancellationToken, stopwatch) - .ConfigureAwait(false); + TaskObservation.ObserveFault(executionTask); + throw new OperationCanceledException(cancellationToken); } - if (await executionTimedOutTask.ConfigureAwait(false)) + if (deadlineState.Signal.Task.IsCompletedSuccessfully + && !await deadlineState.Signal.Task.ConfigureAwait(false)) { - // Any completion after the deadline counts as a response, including a fault - // raised while handling cancellation. The synchronous continuation records the - // ordering at task completion instead of when the winner is later observed. - return TimeoutExecutionResult.TimeoutWithTokenRespected(stopwatch.Elapsed); + return await CreateTimeoutResultAsync(executionTask, cancellationToken, stopwatch) + .ConfigureAwait(false); } var value = await executionTask.ConfigureAwait(false); @@ -222,4 +219,17 @@ private static async Task DidTaskRespondDuringGracePeriodAsync(Task execut return true; } } + + private sealed class CancellationSignalState + { + public Task? ExecutionTask; + + public TaskCompletionSource Signal { get; } = new( + TaskCreationOptions.RunContinuationsAsynchronously); + + public void SignalCancellation() + { + Signal.TrySetResult(Volatile.Read(ref ExecutionTask)?.IsCompleted == true); + } + } } diff --git a/test/ModularPipelines.UnitTests/Execution/ModuleTimeoutTests.cs b/test/ModularPipelines.UnitTests/Execution/ModuleTimeoutTests.cs index 90d45a31ee7..b270b26c82a 100644 --- a/test/ModularPipelines.UnitTests/Execution/ModuleTimeoutTests.cs +++ b/test/ModularPipelines.UnitTests/Execution/ModuleTimeoutTests.cs @@ -276,6 +276,28 @@ await TimeoutHelper.ExecuteWithTimeoutAndDetailsAsync( 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 Timeout_Claims_Tokenless_Cooperative_Cancellation() { From 5b37934b04a25943db0cdc7803fc0edc930eadd2 Mon Sep 17 00:00:00 2001 From: Tom Longhurst <30480171+thomhurst@users.noreply.github.com> Date: Mon, 10 Aug 2026 18:57:07 +0100 Subject: [PATCH 09/12] fix: decouple timeout signal ordering --- .../Engine/ModuleExecutionPipeline.cs | 2 +- src/ModularPipelines/Helpers/TimeoutHelper.cs | 20 +++++++++---------- 2 files changed, 10 insertions(+), 12 deletions(-) diff --git a/src/ModularPipelines/Engine/ModuleExecutionPipeline.cs b/src/ModularPipelines/Engine/ModuleExecutionPipeline.cs index 542bac971d7..a1dc94baf11 100644 --- a/src/ModularPipelines/Engine/ModuleExecutionPipeline.cs +++ b/src/ModularPipelines/Engine/ModuleExecutionPipeline.cs @@ -806,7 +806,7 @@ private Status ClassifyException( private bool IsPipelineCancelled(Exception exception) { - return exception is TaskCanceledException or OperationCanceledException or ModuleTimeoutException + return exception is OperationCanceledException or ModuleTimeoutException && _engineCancellationToken.IsCancelled; } diff --git a/src/ModularPipelines/Helpers/TimeoutHelper.cs b/src/ModularPipelines/Helpers/TimeoutHelper.cs index 6bbb1c7df42..3cd4d3c8c61 100644 --- a/src/ModularPipelines/Helpers/TimeoutHelper.cs +++ b/src/ModularPipelines/Helpers/TimeoutHelper.cs @@ -94,18 +94,13 @@ public static async Task> ExecuteWithTimeoutAndDetails .ConfigureAwait(false); } - // Timeout path: create linked token so task can observe both timeout - // and external cancellation. + // 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 = CancellationTokenSource.CreateLinkedTokenSource( - cancellationToken, - deadlineCts.Token); + using var attemptCts = new CancellationTokenSource(); - // Register after linking the attempt token. Cancellation callbacks run in - // LIFO order, so these observers record whether execution had already - // completed before cancellation propagates to the task. - var deadlineState = new CancellationSignalState(); - var externalCancellationState = new CancellationSignalState(); + var deadlineState = new CancellationSignalState(attemptCts); + var externalCancellationState = new CancellationSignalState(attemptCts); using var deadlineRegistration = deadlineCts.Token.Register( static state => ((CancellationSignalState) state!).SignalCancellation(), deadlineState); @@ -220,7 +215,7 @@ private static async Task DidTaskRespondDuringGracePeriodAsync(Task execut } } - private sealed class CancellationSignalState + private sealed class CancellationSignalState(CancellationTokenSource attemptCts) { public Task? ExecutionTask; @@ -229,7 +224,10 @@ private sealed class CancellationSignalState public void SignalCancellation() { + // The attempt token is not linked to either source, so this sample is + // independent of CancellationTokenSource callback ordering. Signal.TrySetResult(Volatile.Read(ref ExecutionTask)?.IsCompleted == true); + attemptCts.Cancel(); } } } From 877c8dccb000bc039a4444cd6ab797dc4ab489ea Mon Sep 17 00:00:00 2001 From: Tom Longhurst <30480171+thomhurst@users.noreply.github.com> Date: Mon, 17 Aug 2026 02:35:01 +0100 Subject: [PATCH 10/12] fix: coordinate timeout task publication --- src/ModularPipelines/Helpers/TimeoutHelper.cs | 44 ++++++++++++++++--- .../Execution/ModuleTimeoutTests.cs | 28 ++++++++++++ 2 files changed, 65 insertions(+), 7 deletions(-) diff --git a/src/ModularPipelines/Helpers/TimeoutHelper.cs b/src/ModularPipelines/Helpers/TimeoutHelper.cs index 3cd4d3c8c61..d03eaa4862c 100644 --- a/src/ModularPipelines/Helpers/TimeoutHelper.cs +++ b/src/ModularPipelines/Helpers/TimeoutHelper.cs @@ -112,8 +112,8 @@ public static async Task> ExecuteWithTimeoutAndDetails deadlineCts.CancelAfter(timeout.Value); var executionTask = taskFactory(attemptCts.Token); - Volatile.Write(ref deadlineState.ExecutionTask, executionTask); - Volatile.Write(ref externalCancellationState.ExecutionTask, executionTask); + deadlineState.PublishExecutionTask(executionTask); + externalCancellationState.PublishExecutionTask(executionTask); await Task.WhenAny( executionTask, @@ -215,18 +215,48 @@ private static async Task DidTaskRespondDuringGracePeriodAsync(Task execut } } - private sealed class CancellationSignalState(CancellationTokenSource attemptCts) + internal sealed class CancellationSignalState(CancellationTokenSource attemptCts) { - public Task? ExecutionTask; + 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) + { + _executionTask = executionTask; + cancellationSignaled = _cancellationSignaled; + } + + if (cancellationSignaled) + { + // 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); + } + } + public void SignalCancellation() { - // The attempt token is not linked to either source, so this sample is - // independent of CancellationTokenSource callback ordering. - Signal.TrySetResult(Volatile.Read(ref ExecutionTask)?.IsCompleted == true); + Task? executionTask; + lock (_lock) + { + _cancellationSignaled = true; + executionTask = _executionTask; + } + + if (executionTask is not null) + { + // Record ordering before propagating cancellation to the attempt. + Signal.TrySetResult(executionTask.IsCompleted); + } + attemptCts.Cancel(); } } diff --git a/test/ModularPipelines.UnitTests/Execution/ModuleTimeoutTests.cs b/test/ModularPipelines.UnitTests/Execution/ModuleTimeoutTests.cs index b270b26c82a..3b211af24b8 100644 --- a/test/ModularPipelines.UnitTests/Execution/ModuleTimeoutTests.cs +++ b/test/ModularPipelines.UnitTests/Execution/ModuleTimeoutTests.cs @@ -298,6 +298,34 @@ public async Task Completion_With_Asynchronous_Continuations_Before_Deadline_Is_ } } + [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() { From a48c1d1a7b21e9d1cd9b237272a8f8a1e116049d Mon Sep 17 00:00:00 2001 From: Tom Longhurst <30480171+thomhurst@users.noreply.github.com> Date: Sun, 23 Aug 2026 19:22:14 +0100 Subject: [PATCH 11/12] fix(timeout): adapt retry guard to Kevlar Kevlar skips retry callbacks when its shield token is canceled. Keep non-cooperative timeout handling safe by canceling the shield and assert public retry behavior instead. Refs #3837 --- .../Engine/ModuleExecutionPipeline.cs | 4 ++-- .../Execution/RetryTests.cs | 11 +---------- 2 files changed, 3 insertions(+), 12 deletions(-) diff --git a/src/ModularPipelines/Engine/ModuleExecutionPipeline.cs b/src/ModularPipelines/Engine/ModuleExecutionPipeline.cs index a1dc94baf11..7fa61e831fa 100644 --- a/src/ModularPipelines/Engine/ModuleExecutionPipeline.cs +++ b/src/ModularPipelines/Engine/ModuleExecutionPipeline.cs @@ -578,8 +578,8 @@ private static async Task ExecuteModuleAttemptAsync( if (!timeoutResult.WasCancellationTokenRespected) { policyExecutionState.AbandonedAttemptTimeout = timeoutException; - // Let wrapped policies observe the failure, but cancel their retry delay because - // re-entering this module while the abandoned attempt is active is unsafe. + // Cancel the shield before its retry delay because re-entering this module while + // the abandoned attempt is active is unsafe. retryCancellationTokenSource?.Cancel(); } diff --git a/test/ModularPipelines.UnitTests/Execution/RetryTests.cs b/test/ModularPipelines.UnitTests/Execution/RetryTests.cs index 108fcb71b2a..7361d9a0eb0 100644 --- a/test/ModularPipelines.UnitTests/Execution/RetryTests.cs +++ b/test/ModularPipelines.UnitTests/Execution/RetryTests.cs @@ -305,17 +305,9 @@ private class NonCancellableModuleWithTimeout : Module internal int ExecutionCount; - internal int RetryCallbackCount; - protected override ModuleConfiguration Configure() => ModuleConfiguration.Create() .WithTimeout(TimeSpan.FromMilliseconds(50)) - .Advanced - .WithRetryPolicy(Policy - .Handle() - .WaitAndRetryAsync( - DefaultRetryCount, - _ => TimeSpan.FromMinutes(1), - (_, _, _, _) => RetryCallbackCount++)) + .WithRetry(DefaultRetryCount, TimeSpan.FromMinutes(1)) .Build(); protected internal override async Task ExecuteAsync(IModuleContext context, CancellationToken cancellationToken) @@ -409,7 +401,6 @@ public async Task When_Timed_Out_Attempt_Remains_Active_Then_Do_Not_Retry() using (Assert.Multiple()) { await Assert.That(module.ExecutionCount).IsEqualTo(ExpectedSingleExecutionCount); - await Assert.That(module.RetryCallbackCount).IsEqualTo(1); await Assert.That(timeoutException).IsNotNull(); await Assert.That(timeoutException!.WasCancellationTokenRespected).IsFalse(); } From 7aa0127a5a17d717b48ffa2814d962d00b9b42b9 Mon Sep 17 00:00:00 2001 From: Tom Longhurst <30480171+thomhurst@users.noreply.github.com> Date: Sun, 23 Aug 2026 23:11:20 +0100 Subject: [PATCH 12/12] test(retry): cover advanced abandoned attempt --- .../Execution/RetryTests.cs | 61 +++++++++++++++++++ 1 file changed, 61 insertions(+) diff --git a/test/ModularPipelines.UnitTests/Execution/RetryTests.cs b/test/ModularPipelines.UnitTests/Execution/RetryTests.cs index 7361d9a0eb0..276c4543da5 100644 --- a/test/ModularPipelines.UnitTests/Execution/RetryTests.cs +++ b/test/ModularPipelines.UnitTests/Execution/RetryTests.cs @@ -319,6 +319,37 @@ protected internal override async Task ExecuteAsync(IModuleContext context 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 = @@ -411,6 +442,36 @@ public async Task When_Timed_Out_Attempt_Remains_Active_Then_Do_Not_Retry() } } + [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() {