diff --git a/.github/workflows/packages.yml b/.github/workflows/packages.yml index d78513c43..7b58b55c5 100644 --- a/.github/workflows/packages.yml +++ b/.github/workflows/packages.yml @@ -17,7 +17,7 @@ on: types: [prereleased, published] env: - base_version: '3.6.0' + base_version: '3.7.0' feedz_feed_source: 'https://f.feedz.io/elsa-workflows/elsa-3/nuget/index.json' nuget_feed_source: 'https://api.nuget.org/v3/index.json' @@ -155,7 +155,7 @@ jobs: - uses: actions/setup-dotnet@v4 with: - dotnet-version: 9.x + dotnet-version: 10.x - name: Compile+Pack run: ./build.sh Compile+Pack --version ${VERSION} --analyseCode true diff --git a/Elsa.sln b/Elsa.sln index 7a77eb07c..c1e044cb0 100644 --- a/Elsa.sln +++ b/Elsa.sln @@ -323,6 +323,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "issue_templates", "issue_te EndProject Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "dsl", "dsl", "{477C2416-312D-46AE-BCD6-8FA1FAB43624}" EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Resilience.Core.UnitTests", "test\unit\Elsa.Resilience.Core.UnitTests\Elsa.Resilience.Core.UnitTests.csproj", "{B8006D70-1630-43DB-A043-FA89FAC70F37}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -579,6 +581,10 @@ Global {2B7FB49D-E4B6-4AD5-981B-3D85B94F6F48}.Debug|Any CPU.Build.0 = Debug|Any CPU {2B7FB49D-E4B6-4AD5-981B-3D85B94F6F48}.Release|Any CPU.ActiveCfg = Release|Any CPU {2B7FB49D-E4B6-4AD5-981B-3D85B94F6F48}.Release|Any CPU.Build.0 = Release|Any CPU + {B8006D70-1630-43DB-A043-FA89FAC70F37}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {B8006D70-1630-43DB-A043-FA89FAC70F37}.Debug|Any CPU.Build.0 = Debug|Any CPU + {B8006D70-1630-43DB-A043-FA89FAC70F37}.Release|Any CPU.ActiveCfg = Release|Any CPU + {B8006D70-1630-43DB-A043-FA89FAC70F37}.Release|Any CPU.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE @@ -680,6 +686,7 @@ Global {2B7FB49D-E4B6-4AD5-981B-3D85B94F6F48} = {B08B4E00-C2AB-48F3-8389-449F42AEF179} {477C2416-312D-46AE-BCD6-8FA1FAB43624} = {5BA4A8FA-F7F4-45B3-AEC8-8886D35AAC79} {874F5A44-DB06-47AB-A18C-2D13942E0147} = {477C2416-312D-46AE-BCD6-8FA1FAB43624} + {B8006D70-1630-43DB-A043-FA89FAC70F37} = {18453B51-25EB-4317-A4B3-B10518252E92} EndGlobalSection GlobalSection(ExtensibilityGlobals) = postSolution SolutionGuid = {D4B5CEAA-7D70-4FCB-A68E-B03FBE5E0E5E} diff --git a/src/common/Elsa.Mediator/HostedServices/BackgroundCommandSenderHostedService.cs b/src/common/Elsa.Mediator/HostedServices/BackgroundCommandSenderHostedService.cs index f0606a1f5..26916c831 100644 --- a/src/common/Elsa.Mediator/HostedServices/BackgroundCommandSenderHostedService.cs +++ b/src/common/Elsa.Mediator/HostedServices/BackgroundCommandSenderHostedService.cs @@ -45,12 +45,19 @@ public class BackgroundCommandSenderHostedService : BackgroundService } // Main dispatcher loop: read from the input channel and distribute to worker channels - await foreach (var commandContext in _commandsChannel.Reader.ReadAllAsync(cancellationToken)) + try { - var output = _outputs[index]; - await output.Writer.WriteAsync(commandContext, cancellationToken); - // Round-robin distribution - move to next worker - index = (index + 1) % _workerCount; + await foreach (var commandContext in _commandsChannel.Reader.ReadAllAsync(cancellationToken)) + { + var output = _outputs[index]; + await output.Writer.WriteAsync(commandContext, cancellationToken); + // Round-robin distribution - move to next worker + index = (index + 1) % _workerCount; + } + } + catch (OperationCanceledException ex) + { + _logger.LogDebug(ex, "An operation was cancelled while processing the queue"); } // If the input channel is completed, complete all worker channels @@ -61,31 +68,38 @@ public class BackgroundCommandSenderHostedService : BackgroundService private async Task ReadOutputAsync(Channel output, CancellationToken cancellationToken) { // Worker task: process commands from the worker's channel - await foreach (var commandContext in output.Reader.ReadAllAsync(cancellationToken)) + try { - try + await foreach (var commandContext in output.Reader.ReadAllAsync(cancellationToken)) { - // Create a fresh scope for each command to ensure proper service lifetime - using var scope = _scopeFactory.CreateScope(); - var commandSender = scope.ServiceProvider.GetRequiredService(); + try + { + // Create a fresh scope for each command to ensure proper service lifetime + using var scope = _scopeFactory.CreateScope(); + var commandSender = scope.ServiceProvider.GetRequiredService(); - // Link the service cancellation token with the command's token to ensure proper cancellation - using var linkedTokenSource = CancellationTokenSource.CreateLinkedTokenSource( - cancellationToken, - commandContext.CancellationToken); + // Link the service cancellation token with the command's token to ensure proper cancellation + using var linkedTokenSource = CancellationTokenSource.CreateLinkedTokenSource( + cancellationToken, + commandContext.CancellationToken); - // Process the command using the command sender service with the linked token - await commandSender.SendAsync( - commandContext.Command, - CommandStrategy.Default, - commandContext.Headers, - linkedTokenSource.Token); - } - catch (Exception e) - { - // Log errors but continue processing other commands - _logger.LogError(e, "An unhandled exception occurred while processing the queue"); + // Process the command using the command sender service with the linked token + await commandSender.SendAsync( + commandContext.Command, + CommandStrategy.Default, + commandContext.Headers, + linkedTokenSource.Token); + } + catch (Exception e) + { + // Log errors but continue processing other commands + _logger.LogError(e, "An unhandled exception occurred while processing the queue"); + } } } + catch (OperationCanceledException ex) + { + _logger.LogDebug(ex, "An operation was cancelled while processing the queue"); + } } } \ No newline at end of file diff --git a/src/common/Elsa.Mediator/HostedServices/BackgroundEventPublisherHostedService.cs b/src/common/Elsa.Mediator/HostedServices/BackgroundEventPublisherHostedService.cs index 00c29bba1..9bb516bdf 100644 --- a/src/common/Elsa.Mediator/HostedServices/BackgroundEventPublisherHostedService.cs +++ b/src/common/Elsa.Mediator/HostedServices/BackgroundEventPublisherHostedService.cs @@ -52,12 +52,19 @@ public class BackgroundEventPublisherHostedService : BackgroundService // Continuously read notifications from the input channel and distribute them to worker channels // using round-robin distribution for load balancing - await foreach (var notification in channelReader.ReadAllAsync(cancellationToken)) + try { - var output = _outputs[index]; - await output.Writer.WriteAsync(notification, cancellationToken); - // Move to the next worker in a circular fashion - index = (index + 1) % _workerCount; + await foreach (var notification in channelReader.ReadAllAsync(cancellationToken)) + { + var output = _outputs[index]; + await output.Writer.WriteAsync(notification, cancellationToken); + // Move to the next worker in a circular fashion + index = (index + 1) % _workerCount; + } + } + catch (OperationCanceledException ex) + { + _logger.LogDebug(ex, "An operation was cancelled while processing the queue"); } // When the input channel is completed, complete all output channels @@ -75,23 +82,30 @@ public class BackgroundEventPublisherHostedService : BackgroundService /// Cancellation token from the hosted service private async Task ReadOutputAsync(Channel output, INotificationSender notificationSender, CancellationToken cancellationToken) { - await foreach (var notificationContext in output.Reader.ReadAllAsync(cancellationToken)) + try { - try + await foreach (var notificationContext in output.Reader.ReadAllAsync(cancellationToken)) { - var notification = notificationContext.Notification; - // Link the cancellation tokens so that cancellation can happen from either source - using var linkedTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, notificationContext.CancellationToken); - await notificationSender.SendAsync(notification, NotificationStrategy.Sequential, linkedTokenSource.Token); - } - catch (OperationCanceledException e) - { - _logger.LogDebug(e, "An operation was cancelled while processing the queue"); - } - catch (Exception e) - { - _logger.LogError(e, "An unhandled exception occurred while processing the queue"); + try + { + var notification = notificationContext.Notification; + // Link the cancellation tokens so that cancellation can happen from either source + using var linkedTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, notificationContext.CancellationToken); + await notificationSender.SendAsync(notification, NotificationStrategy.Sequential, linkedTokenSource.Token); + } + catch (OperationCanceledException e) + { + _logger.LogDebug(e, "An operation was cancelled while processing the queue"); + } + catch (Exception e) + { + _logger.LogError(e, "An unhandled exception occurred while processing the queue"); + } } } + catch (OperationCanceledException ex) + { + _logger.LogDebug(ex, "An operation was cancelled while processing the queue"); + } } } \ No newline at end of file diff --git a/src/common/Elsa.Testing.Shared/XunitLogger.cs b/src/common/Elsa.Testing.Shared/XunitLogger.cs index e02d37fb5..9d0e033f1 100644 --- a/src/common/Elsa.Testing.Shared/XunitLogger.cs +++ b/src/common/Elsa.Testing.Shared/XunitLogger.cs @@ -12,10 +12,18 @@ public class XunitLogger(ITestOutputHelper testOutputHelper, string categoryName public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func formatter) { - testOutputHelper.WriteLine($"{categoryName} [{eventId}] {formatter(state, exception)}"); + try + { + testOutputHelper.WriteLine($"{categoryName} [{eventId}] {formatter(state, exception)}"); - if (exception != null) - testOutputHelper.WriteLine(exception.ToString()); + if (exception != null) + testOutputHelper.WriteLine(exception.ToString()); + } + catch (InvalidOperationException) + { + // Suppress "no currently active test" exceptions that can occur when background tasks + // (like timers) try to log after tests have completed + } } private class NoopDisposable : IDisposable diff --git a/src/modules/Elsa.Resilience.Core/Contracts/ITransientExceptionDetector.cs b/src/modules/Elsa.Resilience.Core/Contracts/ITransientExceptionDetector.cs new file mode 100644 index 000000000..1a0623422 --- /dev/null +++ b/src/modules/Elsa.Resilience.Core/Contracts/ITransientExceptionDetector.cs @@ -0,0 +1,14 @@ +namespace Elsa.Resilience; + +/// +/// Service for detecting whether exceptions are transient and may be resolved by retrying. +/// +public interface ITransientExceptionDetector +{ + /// + /// Determines whether the specified exception is transient. + /// + /// The exception to check. + /// True if the exception is transient; otherwise, false. + bool IsTransient(Exception exception); +} diff --git a/src/modules/Elsa.Resilience.Core/Contracts/ITransientExceptionStrategy.cs b/src/modules/Elsa.Resilience.Core/Contracts/ITransientExceptionStrategy.cs new file mode 100644 index 000000000..0cbfcb039 --- /dev/null +++ b/src/modules/Elsa.Resilience.Core/Contracts/ITransientExceptionStrategy.cs @@ -0,0 +1,14 @@ +namespace Elsa.Resilience; + +/// +/// Defines a contract for detecting whether an exception is transient and may be resolved by retrying. +/// +public interface ITransientExceptionStrategy +{ + /// + /// Determines whether the specified exception is transient. + /// + /// The exception to check. + /// True if the exception is transient; otherwise, false. + bool IsTransient(Exception exception); +} diff --git a/src/modules/Elsa.Resilience.Core/Services/DefaultTransientExceptionStrategy.cs b/src/modules/Elsa.Resilience.Core/Services/DefaultTransientExceptionStrategy.cs new file mode 100644 index 000000000..79f85a487 --- /dev/null +++ b/src/modules/Elsa.Resilience.Core/Services/DefaultTransientExceptionStrategy.cs @@ -0,0 +1,65 @@ +namespace Elsa.Resilience; + +/// +/// Default implementation that detects common transient exceptions from the .NET framework and common patterns. +/// +public class DefaultTransientExceptionStrategy : ITransientExceptionStrategy +{ + private static readonly HashSet TransientExceptionTypeNames = new(StringComparer.OrdinalIgnoreCase) + { + // Common framework exceptions + "HttpRequestException", + "TimeoutException", + "TaskCanceledException", + "IOException", + "SocketException", + "EndOfStreamException", + + // Database-related transient exceptions (by name, not type reference) + "DbException", + "SqlException", + "NpgsqlException", + "MongoConnectionException", + "MongoExecutionTimeoutException", + "MongoNodeIsRecoveringException", + "MongoNotPrimaryException", + "MySqlException", + + // Network-related exceptions + "HttpIOException", + "WebException", + }; + + private static readonly HashSet TransientExceptionMessagePatterns = new(StringComparer.OrdinalIgnoreCase) + { + "timeout", + "timed out", + "connection reset", + "connection refused", + "broken pipe", + "network", + "end of stream", + "attempted to read past the end", + "the connection is closed", + "connection is not open", + "failed to connect", + "no connection could be made", + "an existing connection was forcibly closed", + }; + + /// + public bool IsTransient(Exception exception) + { + // Check if the exception type name matches any known transient exception + var exceptionTypeName = exception.GetType().Name; + if (TransientExceptionTypeNames.Contains(exceptionTypeName)) + return true; + + // Check if the exception message contains any transient patterns + var message = exception.Message; + + return !string.IsNullOrEmpty(message) && + TransientExceptionMessagePatterns + .Any(pattern => message.Contains(pattern, StringComparison.OrdinalIgnoreCase)); + } +} diff --git a/src/modules/Elsa.Resilience.Core/Services/TransientExceptionDetector.cs b/src/modules/Elsa.Resilience.Core/Services/TransientExceptionDetector.cs new file mode 100644 index 000000000..35c57ba1d --- /dev/null +++ b/src/modules/Elsa.Resilience.Core/Services/TransientExceptionDetector.cs @@ -0,0 +1,42 @@ +namespace Elsa.Resilience; + +/// +/// Default implementation of that delegates to registered detectors. +/// +public class TransientExceptionDetector(IEnumerable detectors) : ITransientExceptionDetector +{ + private readonly IReadOnlyList _detectors = detectors.ToList(); + + /// + public bool IsTransient(Exception exception) + { + var detectorsList = _detectors; + + // Handle aggregate exceptions specially to avoid redundant checks + if (exception is AggregateException aggregateException) + { + // Check the aggregate exception itself + if (detectorsList.Any(detector => detector.IsTransient(aggregateException))) + return true; + + // Recursively check each inner exception (this will walk their chains) + if (aggregateException.InnerExceptions.Any(IsTransient)) + return true; + + return false; + } + + // Walk the exception chain for non-aggregate exceptions + var currentException = exception; + while (currentException != null) + { + // Check if any detector identifies this exception as transient + if (detectorsList.Any(detector => detector.IsTransient(currentException))) + return true; + + currentException = currentException.InnerException; + } + + return false; + } +} diff --git a/src/modules/Elsa.Resilience/Features/ResilienceFeature.cs b/src/modules/Elsa.Resilience/Features/ResilienceFeature.cs index fe5f8f64e..cefc31421 100644 --- a/src/modules/Elsa.Resilience/Features/ResilienceFeature.cs +++ b/src/modules/Elsa.Resilience/Features/ResilienceFeature.cs @@ -84,5 +84,10 @@ public class ResilienceFeature(IModule module) : FeatureBase(module) .AddScoped(_retryAttemptRecorder) .AddScoped(_retryAttemptReader) .AddHandlersFrom(); + + // Register transient exception detection infrastructure + Services + .AddSingleton() + .AddSingleton(); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime.Distributed/Elsa.Workflows.Runtime.Distributed.csproj b/src/modules/Elsa.Workflows.Runtime.Distributed/Elsa.Workflows.Runtime.Distributed.csproj index 0439f66b4..02aedc10f 100644 --- a/src/modules/Elsa.Workflows.Runtime.Distributed/Elsa.Workflows.Runtime.Distributed.csproj +++ b/src/modules/Elsa.Workflows.Runtime.Distributed/Elsa.Workflows.Runtime.Distributed.csproj @@ -10,11 +10,13 @@ - + + + diff --git a/src/modules/Elsa.Workflows.Runtime.Distributed/Features/DistributedRuntimeFeature.cs b/src/modules/Elsa.Workflows.Runtime.Distributed/Features/DistributedRuntimeFeature.cs index d47e0d75d..e783dd43e 100644 --- a/src/modules/Elsa.Workflows.Runtime.Distributed/Features/DistributedRuntimeFeature.cs +++ b/src/modules/Elsa.Workflows.Runtime.Distributed/Features/DistributedRuntimeFeature.cs @@ -2,6 +2,7 @@ using Elsa.Extensions; using Elsa.Features.Abstractions; using Elsa.Features.Attributes; using Elsa.Features.Services; +using Elsa.Resilience.Features; using Elsa.Workflows.Runtime.Features; using Microsoft.Extensions.DependencyInjection; @@ -11,13 +12,9 @@ namespace Elsa.Workflows.Runtime.Distributed.Features; /// Installs and configures workflow runtime features. /// [DependsOn(typeof(WorkflowRuntimeFeature))] -public class DistributedRuntimeFeature : FeatureBase +[DependsOn(typeof(ResilienceFeature))] +public class DistributedRuntimeFeature(IModule module) : FeatureBase(module) { - /// - public DistributedRuntimeFeature(IModule module) : base(module) - { - } - public override void Configure() { Module.UseWorkflowRuntime(runtime => diff --git a/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowClient.cs b/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowClient.cs index 80aa11d5a..131c71e41 100644 --- a/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowClient.cs +++ b/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowClient.cs @@ -1,21 +1,26 @@ using Elsa.Common.DistributedHosting; +using Elsa.Resilience; using Elsa.Workflows.Runtime.Messages; using Elsa.Workflows.State; using Medallion.Threading; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; +using Polly; namespace Elsa.Workflows.Runtime.Distributed; public class DistributedWorkflowClient( string workflowInstanceId, IDistributedLockProvider distributedLockProvider, + ITransientExceptionDetector transientExceptionDetector, IOptions distributedLockingOptions, - IServiceProvider serviceProvider) + IServiceProvider serviceProvider, + ILogger logger) : IWorkflowClient { private readonly LocalWorkflowClient _localWorkflowClient = ActivatorUtilities.CreateInstance(serviceProvider, workflowInstanceId); - + private readonly Lazy _retryPipeline = new(() => CreateRetryPipeline(transientExceptionDetector, logger, workflowInstanceId)); public string WorkflowInstanceId => workflowInstanceId; public async Task CreateInstanceAsync(CreateWorkflowInstanceRequest request, CancellationToken cancellationToken = default) @@ -25,7 +30,7 @@ public class DistributedWorkflowClient( public async Task RunInstanceAsync(RunWorkflowInstanceRequest request, CancellationToken cancellationToken = default) { - var result = await WithLockAsync(async () => await _localWorkflowClient.RunInstanceAsync(request, cancellationToken)); + var result = await WithLockAsync(async () => await _localWorkflowClient.RunInstanceAsync(request, cancellationToken), cancellationToken); return result; } @@ -52,7 +57,7 @@ public class DistributedWorkflowClient( TriggerActivityId = request.TriggerActivityId, ActivityHandle = request.ActivityHandle, IncludeWorkflowOutput = request.IncludeWorkflowOutput - }, cancellationToken)); + }, cancellationToken), cancellationToken); } public async Task CancelAsync(CancellationToken cancellationToken = default) @@ -78,15 +83,71 @@ public class DistributedWorkflowClient( public async Task DeleteAsync(CancellationToken cancellationToken = default) { // Use the same distributed lock as for execution to prevent concurrent DB writes - return await WithLockAsync(async () => await _localWorkflowClient.DeleteAsync(cancellationToken)); + return await WithLockAsync(async () => await _localWorkflowClient.DeleteAsync(cancellationToken), cancellationToken); } - private async Task WithLockAsync(Func> func) + private async Task WithLockAsync(Func> func, CancellationToken cancellationToken = default) { var lockKey = $"workflow-instance:{WorkflowInstanceId}"; + var lockHandle = await AcquireLockWithRetryAsync(lockKey, cancellationToken); + + try + { + return await func(); + } + finally + { + await ReleaseLockAsync(lockHandle); + } + } + + private async Task AcquireLockWithRetryAsync(string lockKey, CancellationToken cancellationToken = default) + { var lockTimeout = distributedLockingOptions.Value.LockAcquisitionTimeout; - await using var @lock = await distributedLockProvider.AcquireLockAsync(lockKey, lockTimeout); - var result = await func(); - return result; + + return await _retryPipeline.Value.ExecuteAsync(async ct => + await distributedLockProvider.AcquireLockAsync(lockKey, lockTimeout, ct), + cancellationToken); + } + + private async Task ReleaseLockAsync(IDistributedSynchronizationHandle? lockHandle) + { + if (lockHandle == null) + return; + + try + { + await lockHandle.DisposeAsync(); + } + catch (Exception ex) + { + // Log but don't throw - the work is already done, and the lock + // will be automatically released when the connection dies + logger.LogWarning(ex, "Failed to release distributed lock for workflow instance {WorkflowInstanceId}. The lock will be automatically released by the database.", WorkflowInstanceId); + } + } + + private static ResiliencePipeline CreateRetryPipeline( + ITransientExceptionDetector transientExceptionDetector, + ILogger logger, + string workflowInstanceId) + { + const int maxRetryAttempts = 3; + + return new ResiliencePipelineBuilder() + .AddRetry(new() + { + MaxRetryAttempts = maxRetryAttempts, + Delay = TimeSpan.FromMilliseconds(500), + BackoffType = DelayBackoffType.Exponential, + UseJitter = true, + ShouldHandle = new PredicateBuilder().Handle(transientExceptionDetector.IsTransient), + OnRetry = args => + { + logger.LogWarning(args.Outcome.Exception, "Transient error acquiring lock for workflow instance {WorkflowInstanceId}. Attempt {AttemptNumber} of {MaxAttempts}.", workflowInstanceId, args.AttemptNumber + 1, maxRetryAttempts); + return ValueTask.CompletedTask; + } + }) + .Build(); } } \ No newline at end of file diff --git a/test/component/Elsa.Workflows.ComponentTests/Helpers/Fixtures/WorkflowServer.cs b/test/component/Elsa.Workflows.ComponentTests/Helpers/Fixtures/WorkflowServer.cs index 1a10ecba1..7446dbe2e 100644 --- a/test/component/Elsa.Workflows.ComponentTests/Helpers/Fixtures/WorkflowServer.cs +++ b/test/component/Elsa.Workflows.ComponentTests/Helpers/Fixtures/WorkflowServer.cs @@ -12,11 +12,13 @@ using Elsa.Testing.Shared.Handlers; using Elsa.Testing.Shared.Services; using Elsa.Workflows.ComponentTests.Decorators; using Elsa.Workflows.ComponentTests.Materializers; +using Elsa.Workflows.ComponentTests.Scenarios.DistributedLockResilience.Mocks; using Elsa.Workflows.ComponentTests.WorkflowProviders; using Elsa.Workflows.Management; using Elsa.Workflows.Runtime.Distributed.Extensions; using FluentStorage; using JetBrains.Annotations; +using Medallion.Threading; using Microsoft.AspNetCore.Hosting; using Microsoft.AspNetCore.Mvc.Testing; using Microsoft.AspNetCore.TestHost; @@ -127,6 +129,18 @@ public class WorkflowServer(Infrastructure infrastructure, string url) : WebAppl builder.ConfigureTestServices(services => { + // Decorate IDistributedLockProvider with TestDistributedLockProvider so tests use it + services.Decorate(); + + // Also register TestDistributedLockProvider as itself so tests can access it directly for configuration + services.AddSingleton(sp => + { + var provider = sp.GetRequiredService(); + if (provider is not TestDistributedLockProvider testProvider) + throw new InvalidOperationException($"Expected IDistributedLockProvider to be decorated with TestDistributedLockProvider, but got {provider.GetType().Name}"); + return testProvider; + }); + services .AddSingleton() .AddScoped() @@ -138,6 +152,7 @@ public class WorkflowServer(Infrastructure infrastructure, string url) : WebAppl .AddWorkflowsProvider() .AddNotificationHandlersFrom() .Decorate() + .Decorate() ; }); } diff --git a/test/component/Elsa.Workflows.ComponentTests/Scenarios/DistributedLockResilience/DistributedLockResilienceTests.cs b/test/component/Elsa.Workflows.ComponentTests/Scenarios/DistributedLockResilience/DistributedLockResilienceTests.cs new file mode 100644 index 000000000..efa063902 --- /dev/null +++ b/test/component/Elsa.Workflows.ComponentTests/Scenarios/DistributedLockResilience/DistributedLockResilienceTests.cs @@ -0,0 +1,165 @@ +using Elsa.Common.DistributedHosting; +using Elsa.Common.Models; +using Elsa.Resilience; +using Elsa.Workflows.ComponentTests.Abstractions; +using Elsa.Workflows.ComponentTests.Fixtures; +using Elsa.Workflows.ComponentTests.Scenarios.DistributedLockResilience.Mocks; +using Elsa.Workflows.ComponentTests.Scenarios.DistributedLockResilience.Workflows; +using Elsa.Workflows.Models; +using Elsa.Workflows.Runtime; +using Elsa.Workflows.Runtime.Messages; +using Medallion.Threading; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using Polly; + +namespace Elsa.Workflows.ComponentTests.Scenarios.DistributedLockResilience; + +public class DistributedLockResilienceTests(App app) : AppComponentTest(app) +{ + private const int MaxRetryAttempts = 3; + + // The IDistributedLockProvider is decorated with TestDistributedLockProvider in WorkflowServer.ConfigureTestServices + // This cast is safe because the decorator pattern ensures TestDistributedLockProvider wraps the actual provider + private TestDistributedLockProvider MockProvider => (TestDistributedLockProvider)Scope.ServiceProvider.GetRequiredService(); + private ITransientExceptionDetector TransientExceptionDetector => Scope.ServiceProvider.GetRequiredService(); + private ILogger Logger => Scope.ServiceProvider.GetRequiredService>(); + private DistributedLockingOptions LockOptions => Scope.ServiceProvider.GetRequiredService>().Value; + private ResiliencePipeline RetryPipeline => CreateRetryPipeline(TransientExceptionDetector, Logger); + + [Theory] + [InlineData(1, 2, false)] // Single failure, succeeds on retry + [InlineData(2, 3, false)] // Two failures, succeeds on third attempt + [InlineData(4, 4, true)] // Four failures, exhausts retries (MaxRetryAttempts = 3) + public async Task AcquireLockWithRetry_AcquisitionFailures_BehavesAsExpected(int failureCount, int expectedAttemptCount, bool shouldThrow) + { + // Arrange + MockProvider.Reset(); + MockProvider.FailAcquisitionTimes(failureCount); + + // Act & Assert + if (shouldThrow) + { + await Assert.ThrowsAsync(async () => await AcquireLockWithRetryAsync($"test-lock-{failureCount}")); + } + else + { + await using var handle = await AcquireLockWithRetryAsync($"test-lock-{failureCount}"); + Assert.NotNull(handle); + } + + Assert.Equal(expectedAttemptCount, MockProvider.AcquisitionAttemptCount); + } + + [Theory] + [InlineData(1, false)] // Single failure, succeeds on retry (2 attempts) + [InlineData(2, false)] // Two failures, succeeds on third attempt (3 attempts) + [InlineData(4, true)] // Four failures, exhausts retries (MaxRetryAttempts = 3, so 4 attempts total) + public async Task RunInstanceAsync_TransientLockFailures_RetriesCorrectly(int failureCount, bool shouldThrow) + { + // Arrange + var workflowClient = await CreateWorkflowClientAsync(); + var workflowInstanceId = workflowClient.WorkflowInstanceId; + + // Reset and configure failures for this specific workflow instance's lock + MockProvider.Reset(); + MockProvider.FailAcquisitionTimesForLock($"workflow-instance:{workflowInstanceId}", failureCount); + var attemptCountBefore = MockProvider.AcquisitionAttemptCount; + + // Now run the instance with the configured lock failures + var runRequest = new RunWorkflowInstanceRequest(); + + // Act & Assert + if (shouldThrow) + { + await Assert.ThrowsAsync(async () => + await workflowClient.RunInstanceAsync(runRequest)); + } + else + { + var response = await workflowClient.RunInstanceAsync(runRequest); + Assert.NotNull(response); + } + + // Verify retries occurred - check the delta from before the operation to account for background noise + var expectedAttempts = failureCount + 1; // failures + 1 success (or final failure for shouldThrow case) + AssertMinimumAttempts(MockProvider.AcquisitionAttemptCount - attemptCountBefore, expectedAttempts, "acquisition"); + } + + [Fact] + public async Task RunInstanceAsync_TransientReleaseFailure_ShouldLogButNotThrow() + { + // Arrange + MockProvider.Reset(); + var workflowClient = await CreateWorkflowClientAsync(createInstance: false); + + // Configure failure after client creation to minimize background interference + MockProvider.FailReleaseOnce(); + var releaseCountBefore = MockProvider.ReleaseAttemptCount; + + // Act - Release failure should be caught and logged, not thrown + var response = await workflowClient.CreateAndRunInstanceAsync(CreateAndRunRequest()); + + // Assert + Assert.NotNull(response); + Assert.NotNull(response.WorkflowInstanceId); + AssertMinimumAttempts(MockProvider.ReleaseAttemptCount - releaseCountBefore, 1, "release"); + } + + private async Task AcquireLockWithRetryAsync(string lockName) => + await RetryPipeline.ExecuteAsync(async ct => + await MockProvider.AcquireLockAsync(lockName, LockOptions.LockAcquisitionTimeout, ct), + CancellationToken.None); + + /// + /// Creates a workflow client with an optional workflow instance already created. + /// + private async Task CreateWorkflowClientAsync(bool createInstance = true) + { + var workflowRuntime = Scope.ServiceProvider.GetRequiredService(); + var workflowClient = await workflowRuntime.CreateClientAsync(); + + if (createInstance) + { + var createRequest = new CreateWorkflowInstanceRequest + { + WorkflowDefinitionHandle = WorkflowDefinitionHandle.ByDefinitionId(SimpleWorkflow.DefinitionId, VersionOptions.Latest) + }; + await workflowClient.CreateInstanceAsync(createRequest); + } + + return workflowClient; + } + + private static CreateAndRunWorkflowInstanceRequest CreateAndRunRequest() => + new() + { + WorkflowDefinitionHandle = WorkflowDefinitionHandle.ByDefinitionId(SimpleWorkflow.DefinitionId, VersionOptions.Latest) + }; + + private static void AssertMinimumAttempts(int actualAttempts, int expectedAttempts, string attemptType) => + Assert.True(actualAttempts >= expectedAttempts, + $"Expected at least {expectedAttempts} {attemptType} attempts, but got {actualAttempts}"); + + private static ResiliencePipeline CreateRetryPipeline(ITransientExceptionDetector transientExceptionDetector, ILogger logger) => + new ResiliencePipelineBuilder() + .AddRetry(new() + { + MaxRetryAttempts = MaxRetryAttempts, + // NOTE: The test retry policy intentionally differs from the production configuration. + // - We use a short, constant delay (10ms) to keep tests fast. + // - We disable jitter and exponential backoff to make timing deterministic and assertions stable. + // The production pipeline uses a larger delay with exponential backoff and jitter for robustness. + Delay = TimeSpan.FromMilliseconds(10), + BackoffType = DelayBackoffType.Constant, + UseJitter = false, + ShouldHandle = new PredicateBuilder().Handle(transientExceptionDetector.IsTransient), + OnRetry = args => + { + logger.LogWarning(args.Outcome.Exception, "Transient error acquiring lock. Attempt {AttemptNumber} of {MaxAttempts}.", args.AttemptNumber + 1, MaxRetryAttempts); + return ValueTask.CompletedTask; + } + }) + .Build(); +} diff --git a/test/component/Elsa.Workflows.ComponentTests/Scenarios/DistributedLockResilience/Mocks/TestDistributedLock.cs b/test/component/Elsa.Workflows.ComponentTests/Scenarios/DistributedLockResilience/Mocks/TestDistributedLock.cs new file mode 100644 index 000000000..1418243bc --- /dev/null +++ b/test/component/Elsa.Workflows.ComponentTests/Scenarios/DistributedLockResilience/Mocks/TestDistributedLock.cs @@ -0,0 +1,56 @@ +using Medallion.Threading; + +namespace Elsa.Workflows.ComponentTests.Scenarios.DistributedLockResilience.Mocks; + +/// +/// Test implementation of IDistributedLock that delegates to an inner lock +/// but can simulate transient failures. +/// +public class TestDistributedLock(IDistributedLock innerLock, TestDistributedLockProvider provider, string lockName) : IDistributedLock +{ + public string Name => innerLock.Name; + + public IDistributedSynchronizationHandle Acquire(TimeSpan? timeout = null, CancellationToken cancellationToken = default) + { + if (provider.ShouldFailAcquisition(lockName)) + { + throw new TimeoutException("Simulated transient timeout during lock acquisition"); + } + + var handle = innerLock.Acquire(timeout, cancellationToken); + return new TestDistributedSynchronizationHandle(handle, provider); + } + + public async ValueTask AcquireAsync(TimeSpan? timeout = null, CancellationToken cancellationToken = default) + { + if (provider.ShouldFailAcquisition(lockName)) + { + throw new TimeoutException("Simulated transient timeout during lock acquisition"); + } + + var handle = await innerLock.AcquireAsync(timeout, cancellationToken); + return new TestDistributedSynchronizationHandle(handle, provider); + } + + public async ValueTask TryAcquireAsync(TimeSpan timeout = default, CancellationToken cancellationToken = default) + { + if (provider.ShouldFailAcquisition(lockName)) + { + throw new TimeoutException("Simulated transient timeout during lock acquisition"); + } + + var handle = await innerLock.TryAcquireAsync(timeout, cancellationToken); + return handle == null ? null : new TestDistributedSynchronizationHandle(handle, provider); + } + + public IDistributedSynchronizationHandle? TryAcquire(TimeSpan timeout = default, CancellationToken cancellationToken = default) + { + if (provider.ShouldFailAcquisition(lockName)) + { + throw new TimeoutException("Simulated transient timeout during lock acquisition"); + } + + var handle = innerLock.TryAcquire(timeout, cancellationToken); + return handle == null ? null : new TestDistributedSynchronizationHandle(handle, provider); + } +} diff --git a/test/component/Elsa.Workflows.ComponentTests/Scenarios/DistributedLockResilience/Mocks/TestDistributedLockProvider.cs b/test/component/Elsa.Workflows.ComponentTests/Scenarios/DistributedLockResilience/Mocks/TestDistributedLockProvider.cs new file mode 100644 index 000000000..fa50c825a --- /dev/null +++ b/test/component/Elsa.Workflows.ComponentTests/Scenarios/DistributedLockResilience/Mocks/TestDistributedLockProvider.cs @@ -0,0 +1,84 @@ +using JetBrains.Annotations; +using Medallion.Threading; + +namespace Elsa.Workflows.ComponentTests.Scenarios.DistributedLockResilience.Mocks; + +/// +/// Test implementation of IDistributedLockProvider that allows simulating transient failures. +/// +[UsedImplicitly] +public class TestDistributedLockProvider(IDistributedLockProvider innerProvider) : IDistributedLockProvider +{ + private int _acquisitionFailuresRemaining; + private int _releaseFailuresRemaining; + private int _acquisitionAttemptCount; + private int _releaseAttemptCount; + private string? _targetLockPrefix; + + public int AcquisitionAttemptCount => _acquisitionAttemptCount; + public int ReleaseAttemptCount => _releaseAttemptCount; + + public void FailAcquisitionOnce() => ConfigureAcquisitionFailures(1); + public void FailAcquisitionTimes(int count) => ConfigureAcquisitionFailures(count); + public void FailReleaseOnce() => Interlocked.Exchange(ref _releaseFailuresRemaining, 1); + + /// + /// Configure failures for a specific lock name prefix. Only locks matching this prefix will fail. + /// + public void FailAcquisitionTimesForLock(string lockNamePrefix, int count) + { + _targetLockPrefix = lockNamePrefix; + ConfigureAcquisitionFailures(count); + } + + public void Reset() + { + ConfigureAcquisitionFailures(0); + Interlocked.Exchange(ref _releaseFailuresRemaining, 0); + Interlocked.Exchange(ref _acquisitionAttemptCount, 0); + Interlocked.Exchange(ref _releaseAttemptCount, 0); + _targetLockPrefix = null; + } + + public IDistributedLock CreateLock(string name) => + new TestDistributedLock(innerProvider.CreateLock(name), this, name); + + internal bool ShouldFailAcquisition(string lockName) + { + Interlocked.Increment(ref _acquisitionAttemptCount); + + // If a target lock prefix is configured, only fail locks matching that prefix + if (_targetLockPrefix != null && !lockName.StartsWith(_targetLockPrefix)) + return false; + + return TryConsumeFailure(ref _acquisitionFailuresRemaining); + } + + internal bool ShouldFailRelease() + { + Interlocked.Increment(ref _releaseAttemptCount); + return TryConsumeFailure(ref _releaseFailuresRemaining); + } + + private void ConfigureAcquisitionFailures(int count) => + Interlocked.Exchange(ref _acquisitionFailuresRemaining, count); + + /// + /// Atomically decrements the failure counter if it's greater than 0. + /// Returns true if a failure was consumed, false otherwise. + /// + private static bool TryConsumeFailure(ref int failureCounter) + { + int currentValue, newValue; + do + { + currentValue = Volatile.Read(ref failureCounter); + if (currentValue <= 0) + return false; + + newValue = currentValue - 1; + } while (Interlocked.CompareExchange(ref failureCounter, newValue, currentValue) != currentValue); + + return true; + } +} diff --git a/test/component/Elsa.Workflows.ComponentTests/Scenarios/DistributedLockResilience/Mocks/TestDistributedSynchronizationHandle.cs b/test/component/Elsa.Workflows.ComponentTests/Scenarios/DistributedLockResilience/Mocks/TestDistributedSynchronizationHandle.cs new file mode 100644 index 000000000..92879c70f --- /dev/null +++ b/test/component/Elsa.Workflows.ComponentTests/Scenarios/DistributedLockResilience/Mocks/TestDistributedSynchronizationHandle.cs @@ -0,0 +1,31 @@ +using Medallion.Threading; + +namespace Elsa.Workflows.ComponentTests.Scenarios.DistributedLockResilience.Mocks; + +/// +/// Test implementation of IDistributedSynchronizationHandle that can simulate failures on disposal. +/// +public class TestDistributedSynchronizationHandle( + IDistributedSynchronizationHandle? innerHandle, + TestDistributedLockProvider provider) : IDistributedSynchronizationHandle +{ + public CancellationToken HandleLostToken => innerHandle?.HandleLostToken ?? CancellationToken.None; + + public void Dispose() + { + if (provider.ShouldFailRelease()) + { + throw new TimeoutException("Simulated transient timeout during lock release"); + } + innerHandle?.Dispose(); + } + + public ValueTask DisposeAsync() + { + if (provider.ShouldFailRelease()) + { + throw new TimeoutException("Simulated transient timeout during lock release"); + } + return innerHandle?.DisposeAsync() ?? ValueTask.CompletedTask; + } +} diff --git a/test/component/Elsa.Workflows.ComponentTests/Scenarios/DistributedLockResilience/Workflows/SimpleWorkflow.cs b/test/component/Elsa.Workflows.ComponentTests/Scenarios/DistributedLockResilience/Workflows/SimpleWorkflow.cs new file mode 100644 index 000000000..bc66778c6 --- /dev/null +++ b/test/component/Elsa.Workflows.ComponentTests/Scenarios/DistributedLockResilience/Workflows/SimpleWorkflow.cs @@ -0,0 +1,26 @@ +using Elsa.Extensions; +using Elsa.Workflows.Activities; + +namespace Elsa.Workflows.ComponentTests.Scenarios.DistributedLockResilience.Workflows; + +/// +/// A simple workflow for testing distributed lock resilience. +/// +public class SimpleWorkflow : WorkflowBase +{ + public static readonly string DefinitionId = Guid.NewGuid().ToString(); + + protected override void Build(IWorkflowBuilder builder) + { + builder.WithDefinitionId(DefinitionId); + + builder.Root = new Sequence + { + Activities = + { + new WriteLine("Workflow execution started"), + new WriteLine("Workflow execution completed") + } + }; + } +} diff --git a/test/integration/Elsa.Workflows.IntegrationTests/Scenarios/RunAsynchronousActivityOutput/Tests.cs b/test/integration/Elsa.Workflows.IntegrationTests/Scenarios/RunAsynchronousActivityOutput/Tests.cs index a5b37596c..10c0c2889 100644 --- a/test/integration/Elsa.Workflows.IntegrationTests/Scenarios/RunAsynchronousActivityOutput/Tests.cs +++ b/test/integration/Elsa.Workflows.IntegrationTests/Scenarios/RunAsynchronousActivityOutput/Tests.cs @@ -3,9 +3,8 @@ using Elsa.Testing.Shared; using Elsa.Workflows.Activities; using Elsa.Workflows.IntegrationTests.Scenarios.RunAsynchronousActivityOutput.Activities; using Elsa.Workflows.Memory; -using Elsa.Workflows.Runtime.Distributed; +using Elsa.Workflows.Runtime.Distributed.Extensions; using Elsa.Workflows.Runtime.Stores; -using Microsoft.Extensions.DependencyInjection; namespace Elsa.Workflows.IntegrationTests.Scenarios.RunAsynchronousActivityOutput; @@ -116,16 +115,13 @@ public class Tests // Act var workflowFinishedRecord = await workflow.DispatchWorkflowAndRunToCompletion( - configureServices: services => - { - services.AddScoped(); - }, configureElsa: elsa => { + // Use the distributed runtime feature so the correct bookmark queue worker and its dependencies are registered. elsa.UseWorkflowRuntime(workflowRuntime => { - workflowRuntime.ActivityExecutionLogStore = sp => activityExecutionStore; - workflowRuntime.WorkflowRuntime = sp => sp.GetRequiredService(); + workflowRuntime.UseDistributedRuntime(); + workflowRuntime.ActivityExecutionLogStore = _ => activityExecutionStore; }); }); diff --git a/test/unit/Elsa.Resilience.Core.UnitTests/DefaultTransientExceptionStrategyTests.cs b/test/unit/Elsa.Resilience.Core.UnitTests/DefaultTransientExceptionStrategyTests.cs new file mode 100644 index 000000000..2dccf5bb9 --- /dev/null +++ b/test/unit/Elsa.Resilience.Core.UnitTests/DefaultTransientExceptionStrategyTests.cs @@ -0,0 +1,87 @@ +using System.Net.Sockets; + +namespace Elsa.Resilience.Core.UnitTests; + +public class DefaultTransientExceptionStrategyTests +{ + private readonly DefaultTransientExceptionStrategy _strategy = new(); + + public static TheoryData TransientExceptionTypes => + [ + typeof(HttpRequestException), + typeof(TimeoutException), + typeof(TaskCanceledException), + typeof(IOException), + typeof(SocketException), + typeof(EndOfStreamException) + ]; + + public static TheoryData TransientMessagePatterns => + [ + "timeout", + "timed out", + "connection reset", + "connection refused", + "broken pipe", + "network", + "end of stream", + "attempted to read past the end", + "the connection is closed", + "connection is not open", + "failed to connect", + "no connection could be made", + "an existing connection was forcibly closed", + "TIMEOUT", + "Connection Reset" + ]; + + public static TheoryData NonTransientMessagePatterns => + [ + "Some random error", + "Invalid operation", + "Null reference" + ]; + + [Theory(DisplayName = "Known transient exception types should be detected as transient")] + [MemberData(nameof(TransientExceptionTypes))] + public void IsTransient_KnownTransientExceptionType_ReturnsTrue(Type exceptionType) + { + var exception = (Exception)Activator.CreateInstance(exceptionType)!; + Assert.True(_strategy.IsTransient(exception)); + } + + [Theory(DisplayName = "Exceptions with transient message patterns should be detected as transient")] + [MemberData(nameof(TransientMessagePatterns))] + public void IsTransient_ExceptionWithTransientMessagePattern_ReturnsTrue(string message) + { + var exception = new Exception(message); + Assert.True(_strategy.IsTransient(exception)); + } + + [Theory(DisplayName = "Exceptions with non-transient messages should not be detected as transient")] + [MemberData(nameof(NonTransientMessagePatterns))] + public void IsTransient_ExceptionWithNonTransientMessage_ReturnsFalse(string message) + { + var exception = new Exception(message); + Assert.False(_strategy.IsTransient(exception)); + } + + [Theory(DisplayName = "Non-transient exception types should not be detected as transient regardless of message")] + [InlineData(typeof(InvalidOperationException), "Some error")] + [InlineData(typeof(ArgumentException), "Invalid argument")] + [InlineData(typeof(NullReferenceException), "Object reference not set")] + public void IsTransient_NonTransientExceptionType_ReturnsFalse(Type exceptionType, string message) + { + var exception = (Exception)Activator.CreateInstance(exceptionType, message)!; + Assert.False(_strategy.IsTransient(exception)); + } + + [Theory(DisplayName = "Exceptions with empty messages should be detected based on type only")] + [InlineData(typeof(TimeoutException), true)] + [InlineData(typeof(InvalidOperationException), false)] + public void IsTransient_ExceptionWithEmptyMessage_ChecksTypeOnly(Type exceptionType, bool expectedResult) + { + var exception = (Exception)Activator.CreateInstance(exceptionType, "")!; + Assert.Equal(expectedResult, _strategy.IsTransient(exception)); + } +} diff --git a/test/unit/Elsa.Resilience.Core.UnitTests/Elsa.Resilience.Core.UnitTests.csproj b/test/unit/Elsa.Resilience.Core.UnitTests/Elsa.Resilience.Core.UnitTests.csproj new file mode 100644 index 000000000..f86df2cad --- /dev/null +++ b/test/unit/Elsa.Resilience.Core.UnitTests/Elsa.Resilience.Core.UnitTests.csproj @@ -0,0 +1,13 @@ + + + + [Elsa.Resilience.Core]* + 49 + + + + + + + + diff --git a/test/unit/Elsa.Resilience.Core.UnitTests/ResilienceStrategyCatalogTests.cs b/test/unit/Elsa.Resilience.Core.UnitTests/ResilienceStrategyCatalogTests.cs new file mode 100644 index 000000000..3e8ae3462 --- /dev/null +++ b/test/unit/Elsa.Resilience.Core.UnitTests/ResilienceStrategyCatalogTests.cs @@ -0,0 +1,118 @@ +using Elsa.Resilience.Core.UnitTests.TestHelpers; +using NSubstitute; +using Open.Linq.AsyncExtensions; + +namespace Elsa.Resilience.Core.UnitTests; + +public class ResilienceStrategyCatalogTests +{ + private static ResilienceStrategyCatalog CreateCatalog(params IResilienceStrategySource[] sources) => new(sources); + + [Fact(DisplayName = "Catalog with no sources should return empty list")] + public async Task ListAsync_NoProviders_ReturnsEmptyList() + { + var catalog = CreateCatalog(); + var result = await catalog.ListAsync(); + + Assert.Empty(result); + } + + [Fact(DisplayName = "Catalog should return all strategies from a single source")] + public async Task ListAsync_SingleProviderWithStrategies_ReturnsStrategies() + { + var strategies = new[] + { + TestDataFactory.CreateStrategy("strategy1", "Strategy 1"), + TestDataFactory.CreateStrategy("strategy2", "Strategy 2") + }; + var provider = TestDataFactory.CreateStrategySource(strategies); + var catalog = CreateCatalog(provider); + + var result = await catalog.ListAsync(); + + Assert.Collection(result, + s => Assert.Equal("strategy1", s.Id), + s => Assert.Equal("strategy2", s.Id)); + } + + [Fact(DisplayName = "Catalog should combine strategies from multiple sources")] + public async Task ListAsync_MultipleProviders_CombinesAllStrategies() + { + var provider1 = TestDataFactory.CreateStrategySource( + TestDataFactory.CreateStrategy("strategy1", "Strategy 1"), + TestDataFactory.CreateStrategy("strategy2", "Strategy 2")); + var provider2 = TestDataFactory.CreateStrategySource( + TestDataFactory.CreateStrategy("strategy3", "Strategy 3")); + var catalog = CreateCatalog(provider1, provider2); + + var result = await catalog.ListAsync().ToList(); + + Assert.Equal(3, result.Count); + Assert.Contains(result, s => s.Id == "strategy1"); + Assert.Contains(result, s => s.Id == "strategy2"); + Assert.Contains(result, s => s.Id == "strategy3"); + } + + [Theory(DisplayName = "Catalog should retrieve strategy by ID or return null if not found")] + [InlineData("test-id", "Test Strategy", true)] + [InlineData("non-existent", null, false)] + public async Task GetAsync_WithStrategyId_ReturnsExpectedResult(string searchId, string? expectedDisplayName, bool shouldExist) + { + var strategy = TestDataFactory.CreateStrategy("test-id", "Test Strategy"); + var provider = TestDataFactory.CreateStrategySource(strategy); + var catalog = CreateCatalog(provider); + + var result = await catalog.GetAsync(searchId); + + if (shouldExist) + { + Assert.NotNull(result); + Assert.Equal(searchId, result.Id); + Assert.Equal(expectedDisplayName, result.DisplayName); + } + else + { + Assert.Null(result); + } + } + + [Fact(DisplayName = "Catalog should search all sources to find a strategy")] + public async Task GetAsync_MultipleProvidersStrategyInSecond_ReturnsStrategy() + { + var provider1 = TestDataFactory.CreateStrategySource(TestDataFactory.CreateStrategy("strategy1", "Strategy 1")); + var provider2 = TestDataFactory.CreateStrategySource(TestDataFactory.CreateStrategy("strategy2", "Strategy 2")); + var catalog = CreateCatalog(provider1, provider2); + + var result = await catalog.GetAsync("strategy2"); + + Assert.NotNull(result); + Assert.Equal("strategy2", result.Id); + } + + [Fact(DisplayName = "Catalog should cache strategy list after first retrieval")] + public async Task ListAsync_CalledMultipleTimes_CachesResult() + { + var strategy = TestDataFactory.CreateStrategy("test", "Test"); + var provider = TestDataFactory.CreateStrategySource(strategy); + var catalog = CreateCatalog(provider); + + await catalog.ListAsync(); + await catalog.ListAsync(); + await catalog.ListAsync(); + + await provider.Received(1).GetStrategiesAsync(Arg.Any()); + } + + [Fact(DisplayName = "GetAsync should use cached list when available")] + public async Task GetAsync_CalledAfterList_UsesCachedResult() + { + var strategy = TestDataFactory.CreateStrategy("test", "Test"); + var provider = TestDataFactory.CreateStrategySource(strategy); + var catalog = CreateCatalog(provider); + + await catalog.ListAsync(); + await catalog.GetAsync("test"); + + await provider.Received(1).GetStrategiesAsync(Arg.Any()); + } +} diff --git a/test/unit/Elsa.Resilience.Core.UnitTests/ResilienceStrategyConfigEvaluatorTests.cs b/test/unit/Elsa.Resilience.Core.UnitTests/ResilienceStrategyConfigEvaluatorTests.cs new file mode 100644 index 000000000..ea061c445 --- /dev/null +++ b/test/unit/Elsa.Resilience.Core.UnitTests/ResilienceStrategyConfigEvaluatorTests.cs @@ -0,0 +1,135 @@ +using Elsa.Expressions.Contracts; +using Elsa.Expressions.Models; +using Elsa.Resilience.Core.UnitTests.TestHelpers; +using Elsa.Resilience.Models; +using Elsa.Resilience.Options; +using Elsa.Resilience.Serialization; +using NSubstitute; + +namespace Elsa.Resilience.Core.UnitTests; + +public class ResilienceStrategyConfigEvaluatorTests +{ + private readonly IResilienceStrategyCatalog _catalog = Substitute.For(); + private readonly IExpressionEvaluator _expressionEvaluator = Substitute.For(); + private readonly ResilienceStrategyConfigEvaluator _evaluator; + private readonly ExpressionExecutionContext _context = new(Substitute.For(), null!); + + public ResilienceStrategyConfigEvaluatorTests() + { + var options = Microsoft.Extensions.Options.Options.Create(new ResilienceOptions()); + var serializer = new ResilienceStrategySerializer(options); + _evaluator = new(_catalog, _expressionEvaluator, serializer); + } + + [Fact(DisplayName = "Evaluator should return null when config is null")] + public async Task EvaluateAsync_NullConfig_ReturnsNull() + { + var result = await _evaluator.EvaluateAsync(null, _context); + Assert.Null(result); + } + + [Fact(DisplayName = "Evaluator in identifier mode should resolve strategy from catalog")] + public async Task EvaluateAsync_IdentifierMode_WithValidId_ReturnsStrategyFromCatalog() + { + var strategy = TestDataFactory.CreateStrategy("test-strategy", "Test Strategy"); + SetupCatalogStrategy("test-strategy", strategy); + var config = CreateConfig(ResilienceStrategyConfigMode.Identifier, "test-strategy"); + + var result = await _evaluator.EvaluateAsync(config, _context); + + Assert.NotNull(result); + Assert.Equal("test-strategy", result.Id); + await _catalog.Received(1).GetAsync("test-strategy", Arg.Any()); + } + + [Theory(DisplayName = "Evaluator in identifier mode should return null for invalid strategy IDs")] + [InlineData("")] + [InlineData(" ")] + [InlineData(null)] + public async Task EvaluateAsync_IdentifierMode_WithInvalidId_ReturnsNull(string? strategyId) + { + var config = CreateConfig(ResilienceStrategyConfigMode.Identifier, strategyId); + + var result = await _evaluator.EvaluateAsync(config, _context); + + Assert.Null(result); + await _catalog.DidNotReceive().GetAsync(Arg.Any(), Arg.Any()); + } + + [Fact(DisplayName = "Evaluator in expression mode should return null when expression is null")] + public async Task EvaluateAsync_ExpressionMode_WithNullExpression_ReturnsNull() + { + var config = CreateConfig(ResilienceStrategyConfigMode.Expression); + + var result = await _evaluator.EvaluateAsync(config, _context); + + Assert.Null(result); + await _expressionEvaluator.DidNotReceive().EvaluateAsync(Arg.Any(), Arg.Any(), Arg.Any()); + } + + [Fact(DisplayName = "Evaluator in expression mode should resolve string IDs from catalog")] + public async Task EvaluateAsync_ExpressionMode_ReturnsStringId_ResolvesFromCatalog() + { + var expression = new Expression("C#", "\"test-strategy\""); + var strategy = TestDataFactory.CreateStrategy("test-strategy", "Test Strategy"); + SetupExpressionResult(expression, "test-strategy"); + SetupCatalogStrategy("test-strategy", strategy); + var config = CreateConfig(ResilienceStrategyConfigMode.Expression, expression: expression); + + var result = await _evaluator.EvaluateAsync(config, _context); + + Assert.NotNull(result); + Assert.Equal("test-strategy", result.Id); + await _catalog.Received(1).GetAsync("test-strategy", Arg.Any()); + } + + [Fact(DisplayName = "Evaluator in expression mode should return strategy objects directly")] + public async Task EvaluateAsync_ExpressionMode_ReturnsStrategyObject_ReturnsStrategyDirectly() + { + var expression = new Expression("C#", "strategy"); + var strategy = TestDataFactory.CreateStrategy("direct-strategy", "Direct Strategy"); + SetupExpressionResult(expression, strategy); + var config = CreateConfig(ResilienceStrategyConfigMode.Expression, expression: expression); + + var result = await _evaluator.EvaluateAsync(config, _context); + + Assert.NotNull(result); + Assert.Same(strategy, result); + await _catalog.DidNotReceive().GetAsync(Arg.Any(), Arg.Any()); + } + + [Theory(DisplayName = "Evaluator in expression mode should return null for unexpected result types")] + [InlineData("null", null)] + [InlineData("42", 42)] + public async Task EvaluateAsync_ExpressionMode_ReturnsUnexpectedType_ReturnsNull(string expressionCode, object? expressionResult) + { + var expression = new Expression("C#", expressionCode); + SetupExpressionResult(expression, expressionResult); + var config = CreateConfig(ResilienceStrategyConfigMode.Expression, expression: expression); + + var result = await _evaluator.EvaluateAsync(config, _context); + + Assert.Null(result); + } + + private ResilienceStrategyConfig CreateConfig(ResilienceStrategyConfigMode mode, string? strategyId = null, Expression? expression = null) + { + return new() + { + Mode = mode, + StrategyId = strategyId, + Expression = expression + }; + } + + private void SetupCatalogStrategy(string id, IResilienceStrategy strategy) + { + _catalog.GetAsync(id, Arg.Any()).Returns(strategy); + } + + private void SetupExpressionResult(Expression expression, object? result) + { + _expressionEvaluator.EvaluateAsync(expression, _context, Arg.Any()).Returns(result); + } +} diff --git a/test/unit/Elsa.Resilience.Core.UnitTests/TestHelpers/TestDataFactory.cs b/test/unit/Elsa.Resilience.Core.UnitTests/TestHelpers/TestDataFactory.cs new file mode 100644 index 000000000..e5c937eb9 --- /dev/null +++ b/test/unit/Elsa.Resilience.Core.UnitTests/TestHelpers/TestDataFactory.cs @@ -0,0 +1,21 @@ +using NSubstitute; + +namespace Elsa.Resilience.Core.UnitTests.TestHelpers; + +internal static class TestDataFactory +{ + public static IResilienceStrategy CreateStrategy(string id, string displayName) + { + var strategy = Substitute.For(); + strategy.Id.Returns(id); + strategy.DisplayName.Returns(displayName); + return strategy; + } + + public static IResilienceStrategySource CreateStrategySource(params IResilienceStrategy[] strategies) + { + var source = Substitute.For(); + source.GetStrategiesAsync(Arg.Any()).Returns(strategies); + return source; + } +} diff --git a/test/unit/Elsa.Resilience.Core.UnitTests/TransientExceptionDetectorTests.cs b/test/unit/Elsa.Resilience.Core.UnitTests/TransientExceptionDetectorTests.cs new file mode 100644 index 000000000..2862280b1 --- /dev/null +++ b/test/unit/Elsa.Resilience.Core.UnitTests/TransientExceptionDetectorTests.cs @@ -0,0 +1,151 @@ +using NSubstitute; + +namespace Elsa.Resilience.Core.UnitTests; + +public class TransientExceptionDetectorTests +{ + [Fact(DisplayName = "Service with no registered strategies should return false for any exception")] + public void IsTransient_NoDetectors_ReturnsFalse() + { + var detector = CreateDetector(); + var exception = new Exception("test"); + + Assert.False(detector.IsTransient(exception)); + } + + [Fact(DisplayName = "Service should return true when any strategy detects the exception as transient")] + public void IsTransient_DetectorReturnsTrue_ReturnsTrue() + { + var exception = new Exception("test"); + var strategy = CreateStrategy((exception, true)); + var detector = CreateDetector(strategy); + + Assert.True(detector.IsTransient(exception)); + } + + [Fact(DisplayName = "Service with multiple strategies should return true if any one detects as transient")] + public void IsTransient_MultipleDetectorsOneReturnsTrue_ReturnsTrue() + { + var exception = new Exception("test"); + var strategy1 = CreateStrategy((exception, false)); + var strategy2 = CreateStrategy((exception, true)); + var detector = CreateDetector(strategy1, strategy2); + + Assert.True(detector.IsTransient(exception)); + } + + [Fact(DisplayName = "Service should return false when all strategies detect the exception as non-transient")] + public void IsTransient_AllDetectorsReturnFalse_ReturnsFalse() + { + var exception = new Exception("test"); + var strategy1 = Substitute.For(); + var strategy2 = Substitute.For(); + strategy1.IsTransient(Arg.Any()).Returns(false); + strategy2.IsTransient(Arg.Any()).Returns(false); + var detector = CreateDetector(strategy1, strategy2); + + Assert.False(detector.IsTransient(exception)); + } + + [Theory(DisplayName = "Service should walk the inner exception chain to find transient exceptions")] + [MemberData(nameof(InnerExceptionChainTestCases))] + public void IsTransient_InnerExceptionChainHasTransient_ReturnsTrue(Exception exception, Exception transientException) + { + var strategy = Substitute.For(); + strategy.IsTransient(transientException).Returns(true); + strategy.IsTransient(Arg.Is(e => e != transientException)).Returns(false); + var detector = CreateDetector(strategy); + + Assert.True(detector.IsTransient(exception)); + } + + [Theory(DisplayName = "Service should inspect AggregateException inner exceptions")] + [MemberData(nameof(AggregateExceptionTestCases))] + public void IsTransient_AggregateException_ChecksInnerExceptions( + AggregateException aggregateException, + Action configureDetector, + bool expectedResult) + { + var strategy = Substitute.For(); + configureDetector(strategy); + var detector = CreateDetector(strategy); + + Assert.Equal(expectedResult, detector.IsTransient(aggregateException)); + } + + [Fact(DisplayName = "AggregateException with mixed inner exceptions should be transient if any inner is transient")] + public void IsTransient_AggregateExceptionWithMultipleInnerOneTransient_ReturnsTrue() + { + var transientException = new TimeoutException("timeout"); + var nonTransientException = new InvalidOperationException("invalid"); + var aggregateException = new AggregateException("aggregate", nonTransientException, transientException); + + var strategy = CreateStrategy( + (aggregateException, false), + (nonTransientException, false), + (transientException, true)); + var detector = CreateDetector(strategy); + + Assert.True(detector.IsTransient(aggregateException)); + } + + public static IEnumerable InnerExceptionChainTestCases + { + get + { + var transientException = new TimeoutException("timeout"); + + yield return + [ + new Exception("outer", transientException), + transientException + ]; + + yield return + [ + new Exception("outer", new("middle", transientException)), + transientException + ]; + } + } + + public static IEnumerable AggregateExceptionTestCases + { + get + { + // Scenario 1: One transient inner exception + var timeoutException = new TimeoutException("timeout"); + yield return + [ + new AggregateException("aggregate", timeoutException), + (Action)(detector => + { + detector.IsTransient(Arg.Is(e => e.Message == "timeout")).Returns(true); + detector.IsTransient(Arg.Is(e => e.GetType() != typeof(TimeoutException))).Returns(false); + }), + true + ]; + + // Scenario 2: No transient inner exceptions + yield return + [ + new AggregateException("aggregate", new InvalidOperationException("error1"), new ArgumentException("error2")), + (Action)(detector => + { + detector.IsTransient(Arg.Any()).Returns(false); + }), + false + ]; + } + } + + private static ITransientExceptionStrategy CreateStrategy(params (Exception exception, bool isTransient)[] behaviors) + { + var detector = Substitute.For(); + foreach (var (exception, isTransient) in behaviors) + detector.IsTransient(exception).Returns(isTransient); + return detector; + } + + private static TransientExceptionDetector CreateDetector(params ITransientExceptionStrategy[] detectors) => new(detectors); +}