Enhances distributed lock handling with resilience (#7161)
* Add retry mechanism for distributed locks with transient error handling and logging - Introduced Polly-based retry pipeline for distributed lock acquisition in `DistributedWorkflowClient` to handle transient errors such as network issues or database connection failures. - Added detailed logging for retry attempts and lock release errors. - Updated project dependencies to include Polly. * Refactor transient exception handling to shared resilience module. Migrated transient exception detection logic from scheduling module to a new shared resilience module. Updated services, jobs, and features to utilize the centralized `ITransientExceptionDetectionService`. This change improves maintainability and promotes reusability across modules. * Add unit tests for transient exception detection and resilience strategy evaluation. - Introduced comprehensive unit tests for `DefaultTransientExceptionDetector`, `ResilienceStrategyCatalog`, `ResilienceStrategyConfigEvaluator`, and `TransientExceptionDetectionService`. - Added helper classes and test data factories to facilitate reusable test patterns for resilience modules. - Updated solution to include `Elsa.Resilience.Core.UnitTests` project. * Add component tests for distributed lock resilience - Introduced new tests to verify retry behavior during transient lock acquisition and release failures. - Added `TestDistributedLockProvider` and related mocks for simulating transient failures. - Updated `WorkflowServer` test services to support the new distributed lock test scenarios. * Refactor distributed lock resilience tests - Consolidated test logic: streamlined test providers, injected services, and reusable test patterns. - Simplified `TestDistributedLockProvider` implementation with enhanced initialization and failure simulation. - Reorganized tests for transient acquisition/release failures to use parameterized `Theory` for improved maintainability. * Refactor transient exception handling: rename interfaces and classes for consistency, update references across codebase, and improve code readability. * Apply suggestion from @Copilot Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Simplify WorkflowServer setup and DistributedLockResilienceTests by replacing IDistributedLockProvider with TestDistributedLockProvider. * Remove unused `using` directives in unit tests to improve code cleanliness. * Remove `TransientExceptionTypes` helper and inline its usage in tests for improved maintainability. * Add descriptive `DisplayName` attributes to unit tests for improved test clarity. * Fix redundant exception checking in TransientExceptionDetector (#7162) * Initial plan * Fix redundant exception checking in TransientExceptionDetector Co-authored-by: sfmskywalker <938393+sfmskywalker@users.noreply.github.com> --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: sfmskywalker <938393+sfmskywalker@users.noreply.github.com> * Extract MaxRetryAttempts constant in DistributedLockResilienceTests (#7164) * Initial plan * Extract MaxRetryAttempts constant to eliminate hardcoded magic number Co-authored-by: sfmskywalker <938393+sfmskywalker@users.noreply.github.com> --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: sfmskywalker <938393+sfmskywalker@users.noreply.github.com> * Make TestDistributedLockProvider thread-safe with Interlocked operations (#7163) * Initial plan * Make TestDistributedLockProvider thread-safe using Interlocked operations Co-authored-by: sfmskywalker <938393+sfmskywalker@users.noreply.github.com> --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: sfmskywalker <938393+sfmskywalker@users.noreply.github.com> * Update src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowClient.cs Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Update test/component/Elsa.Workflows.ComponentTests/Scenarios/DistributedLockResilience/DistributedLockResilienceTests.cs Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Update src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowClient.cs Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Update src/modules/Elsa.Resilience.Core/Services/DefaultTransientExceptionStrategy.cs Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Include `CancellationToken` in distributed lock handling methods for improved cancellation support. * Initial plan (#7168) Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> * Cache detector list in TransientExceptionDetector to avoid repeated allocations (#7167) * Initial plan * Cache detector list in field to avoid repeated allocations Co-authored-by: sfmskywalker <938393+sfmskywalker@users.noreply.github.com> * Use IReadOnlyList instead of List for better intent expression Co-authored-by: sfmskywalker <938393+sfmskywalker@users.noreply.github.com> --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: sfmskywalker <938393+sfmskywalker@users.noreply.github.com> * Fix TestDistributedLockProvider registration to properly decorate IDistributedLockProvider (#7166) * Initial plan * Fix TestDistributedLockProvider registration to use Decorate pattern and fix variable reference bug Co-authored-by: sfmskywalker <938393+sfmskywalker@users.noreply.github.com> * Add runtime check for TestDistributedLockProvider registration Co-authored-by: sfmskywalker <938393+sfmskywalker@users.noreply.github.com> --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: sfmskywalker <938393+sfmskywalker@users.noreply.github.com> * Refactor `DistributedWorkflowClient` to simplify `Lazy<ResiliencePipeline>` initialization. * Add integration tests for DistributedWorkflowClient lock resilience (#7165) * Initial plan * Fix compilation error: use correct parameter name transientExceptionDetector Co-authored-by: sfmskywalker <938393+sfmskywalker@users.noreply.github.com> * Add integration tests for DistributedWorkflowClient lock resilience - Add SimpleWorkflow for testing distributed lock scenarios - Add tests exercising RunInstanceAsync with transient lock failures - Verify retry logic works correctly with actual workflow execution - Test both acquisition and release failure scenarios - Decorate IDistributedLockProvider to use TestDistributedLockProvider Co-authored-by: sfmskywalker <938393+sfmskywalker@users.noreply.github.com> * Address code review feedback - Add explanatory comment for TestDistributedLockProvider cast - Remove unnecessary blank line for consistent formatting Co-authored-by: sfmskywalker <938393+sfmskywalker@users.noreply.github.com> --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: sfmskywalker <938393+sfmskywalker@users.noreply.github.com> Co-authored-by: Sipke Schoorstra <sipkeschoorstra@outlook.com> * Simplify transient exception strategy by refactoring message pattern matching logic. * Enhance distributed lock mock to support per-lock failure configuration and improve resilience tests. * Refactor `TestDistributedLockProvider` to streamline failure handling logic and improve code clarity. * Remove unused methods and redundant test case from `DistributedLockResilienceTests`. * Refactor `DistributedLockResilienceTests` to simplify workflow client creation, consolidate assertion logic, and remove redundant test cases. * Format `ResilienceStrategyCatalogTests` by removing redundant line breaks in test setup. * Refactor `TransientExceptionDetectorTests` to simplify test setup, consolidate test cases, and remove redundant logic. * Handle `InvalidOperationException` in `XunitLogger` to suppress logging errors during inactive tests. * Update workflows to use .NET 10 and adjust resilience tests project configuration. * Refactor distributed runtime feature and integrations to improve resilience handling, configure services fluently, and add cancellation safeguards in background services. * Update `base_version` to `3.7.0` in GitHub workflow configuration. --------- Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> Co-authored-by: Copilot <198982749+Copilot@users.noreply.github.com> Co-authored-by: sfmskywalker <938393+sfmskywalker@users.noreply.github.com>
This commit is contained in:
Sipke Schoorstra
GitHub
parent
411ca0a332
commit
ca268c16ad
4
.github/workflows/packages.yml
vendored
4
.github/workflows/packages.yml
vendored
|
|
@ -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
|
||||
|
|
|
|||
7
Elsa.sln
7
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}
|
||||
|
|
|
|||
|
|
@ -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<CommandContext> 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<ICommandSender>();
|
||||
try
|
||||
{
|
||||
// Create a fresh scope for each command to ensure proper service lifetime
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var commandSender = scope.ServiceProvider.GetRequiredService<ICommandSender>();
|
||||
|
||||
// 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");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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
|
|||
/// <param name="cancellationToken">Cancellation token from the hosted service</param>
|
||||
private async Task ReadOutputAsync(Channel<NotificationContext> 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");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -12,10 +12,18 @@ public class XunitLogger(ITestOutputHelper testOutputHelper, string categoryName
|
|||
|
||||
public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func<TState, Exception?, string> 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
|
||||
|
|
|
|||
|
|
@ -0,0 +1,14 @@
|
|||
namespace Elsa.Resilience;
|
||||
|
||||
/// <summary>
|
||||
/// Service for detecting whether exceptions are transient and may be resolved by retrying.
|
||||
/// </summary>
|
||||
public interface ITransientExceptionDetector
|
||||
{
|
||||
/// <summary>
|
||||
/// Determines whether the specified exception is transient.
|
||||
/// </summary>
|
||||
/// <param name="exception">The exception to check.</param>
|
||||
/// <returns>True if the exception is transient; otherwise, false.</returns>
|
||||
bool IsTransient(Exception exception);
|
||||
}
|
||||
|
|
@ -0,0 +1,14 @@
|
|||
namespace Elsa.Resilience;
|
||||
|
||||
/// <summary>
|
||||
/// Defines a contract for detecting whether an exception is transient and may be resolved by retrying.
|
||||
/// </summary>
|
||||
public interface ITransientExceptionStrategy
|
||||
{
|
||||
/// <summary>
|
||||
/// Determines whether the specified exception is transient.
|
||||
/// </summary>
|
||||
/// <param name="exception">The exception to check.</param>
|
||||
/// <returns>True if the exception is transient; otherwise, false.</returns>
|
||||
bool IsTransient(Exception exception);
|
||||
}
|
||||
|
|
@ -0,0 +1,65 @@
|
|||
namespace Elsa.Resilience;
|
||||
|
||||
/// <summary>
|
||||
/// Default implementation that detects common transient exceptions from the .NET framework and common patterns.
|
||||
/// </summary>
|
||||
public class DefaultTransientExceptionStrategy : ITransientExceptionStrategy
|
||||
{
|
||||
private static readonly HashSet<string> 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<string> 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",
|
||||
};
|
||||
|
||||
/// <inheritdoc />
|
||||
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));
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,42 @@
|
|||
namespace Elsa.Resilience;
|
||||
|
||||
/// <summary>
|
||||
/// Default implementation of <see cref="ITransientExceptionDetector"/> that delegates to registered detectors.
|
||||
/// </summary>
|
||||
public class TransientExceptionDetector(IEnumerable<ITransientExceptionStrategy> detectors) : ITransientExceptionDetector
|
||||
{
|
||||
private readonly IReadOnlyList<ITransientExceptionStrategy> _detectors = detectors.ToList();
|
||||
|
||||
/// <inheritdoc />
|
||||
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;
|
||||
}
|
||||
}
|
||||
|
|
@ -84,5 +84,10 @@ public class ResilienceFeature(IModule module) : FeatureBase(module)
|
|||
.AddScoped(_retryAttemptRecorder)
|
||||
.AddScoped(_retryAttemptReader)
|
||||
.AddHandlersFrom<ResilienceFeature>();
|
||||
|
||||
// Register transient exception detection infrastructure
|
||||
Services
|
||||
.AddSingleton<ITransientExceptionStrategy, DefaultTransientExceptionStrategy>()
|
||||
.AddSingleton<ITransientExceptionDetector, TransientExceptionDetector>();
|
||||
}
|
||||
}
|
||||
|
|
@ -10,11 +10,13 @@
|
|||
<ItemGroup>
|
||||
<PackageReference Include="DistributedLock.FileSystem" />
|
||||
<PackageReference Include="Microsoft.Extensions.DependencyInjection" />
|
||||
<PackageReference Include="Microsoft.Extensions.Hosting.Abstractions"/>
|
||||
<PackageReference Include="Microsoft.Extensions.Hosting.Abstractions"/>
|
||||
<PackageReference Include="Open.Linq.AsyncExtensions" />
|
||||
<PackageReference Include="Polly" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\Elsa.Resilience\Elsa.Resilience.csproj" />
|
||||
<ProjectReference Include="..\Elsa.Workflows.Runtime\Elsa.Workflows.Runtime.csproj" />
|
||||
</ItemGroup>
|
||||
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
/// </summary>
|
||||
[DependsOn(typeof(WorkflowRuntimeFeature))]
|
||||
public class DistributedRuntimeFeature : FeatureBase
|
||||
[DependsOn(typeof(ResilienceFeature))]
|
||||
public class DistributedRuntimeFeature(IModule module) : FeatureBase(module)
|
||||
{
|
||||
/// <inheritdoc />
|
||||
public DistributedRuntimeFeature(IModule module) : base(module)
|
||||
{
|
||||
}
|
||||
|
||||
public override void Configure()
|
||||
{
|
||||
Module.UseWorkflowRuntime(runtime =>
|
||||
|
|
|
|||
|
|
@ -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> distributedLockingOptions,
|
||||
IServiceProvider serviceProvider)
|
||||
IServiceProvider serviceProvider,
|
||||
ILogger<DistributedWorkflowClient> logger)
|
||||
: IWorkflowClient
|
||||
{
|
||||
private readonly LocalWorkflowClient _localWorkflowClient = ActivatorUtilities.CreateInstance<LocalWorkflowClient>(serviceProvider, workflowInstanceId);
|
||||
|
||||
private readonly Lazy<ResiliencePipeline> _retryPipeline = new(() => CreateRetryPipeline(transientExceptionDetector, logger, workflowInstanceId));
|
||||
public string WorkflowInstanceId => workflowInstanceId;
|
||||
|
||||
public async Task<CreateWorkflowInstanceResponse> CreateInstanceAsync(CreateWorkflowInstanceRequest request, CancellationToken cancellationToken = default)
|
||||
|
|
@ -25,7 +30,7 @@ public class DistributedWorkflowClient(
|
|||
|
||||
public async Task<RunWorkflowInstanceResponse> 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<bool> 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<R> WithLockAsync<R>(Func<Task<R>> func)
|
||||
private async Task<TReturn> WithLockAsync<TReturn>(Func<Task<TReturn>> 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<IDistributedSynchronizationHandle?> 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<DistributedWorkflowClient> 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<Exception>(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();
|
||||
}
|
||||
}
|
||||
|
|
@ -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<IDistributedLockProvider, TestDistributedLockProvider>();
|
||||
|
||||
// Also register TestDistributedLockProvider as itself so tests can access it directly for configuration
|
||||
services.AddSingleton(sp =>
|
||||
{
|
||||
var provider = sp.GetRequiredService<IDistributedLockProvider>();
|
||||
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<SignalManager>()
|
||||
.AddScoped<AsyncWorkflowRunner>()
|
||||
|
|
@ -138,6 +152,7 @@ public class WorkflowServer(Infrastructure infrastructure, string url) : WebAppl
|
|||
.AddWorkflowsProvider<TestWorkflowProvider>()
|
||||
.AddNotificationHandlersFrom<WorkflowEventHandlers>()
|
||||
.Decorate<IChangeTokenSignaler, EventPublishingChangeTokenSignaler>()
|
||||
.Decorate<IDistributedLockProvider, TestDistributedLockProvider>()
|
||||
;
|
||||
});
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<IDistributedLockProvider>();
|
||||
private ITransientExceptionDetector TransientExceptionDetector => Scope.ServiceProvider.GetRequiredService<ITransientExceptionDetector>();
|
||||
private ILogger<DistributedLockResilienceTests> Logger => Scope.ServiceProvider.GetRequiredService<ILogger<DistributedLockResilienceTests>>();
|
||||
private DistributedLockingOptions LockOptions => Scope.ServiceProvider.GetRequiredService<IOptions<DistributedLockingOptions>>().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<TimeoutException>(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<TimeoutException>(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<IDistributedSynchronizationHandle?> AcquireLockWithRetryAsync(string lockName) =>
|
||||
await RetryPipeline.ExecuteAsync(async ct =>
|
||||
await MockProvider.AcquireLockAsync(lockName, LockOptions.LockAcquisitionTimeout, ct),
|
||||
CancellationToken.None);
|
||||
|
||||
/// <summary>
|
||||
/// Creates a workflow client with an optional workflow instance already created.
|
||||
/// </summary>
|
||||
private async Task<IWorkflowClient> CreateWorkflowClientAsync(bool createInstance = true)
|
||||
{
|
||||
var workflowRuntime = Scope.ServiceProvider.GetRequiredService<IWorkflowRuntime>();
|
||||
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<Exception>(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();
|
||||
}
|
||||
|
|
@ -0,0 +1,56 @@
|
|||
using Medallion.Threading;
|
||||
|
||||
namespace Elsa.Workflows.ComponentTests.Scenarios.DistributedLockResilience.Mocks;
|
||||
|
||||
/// <summary>
|
||||
/// Test implementation of IDistributedLock that delegates to an inner lock
|
||||
/// but can simulate transient failures.
|
||||
/// </summary>
|
||||
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<IDistributedSynchronizationHandle> 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<IDistributedSynchronizationHandle?> 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);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,84 @@
|
|||
using JetBrains.Annotations;
|
||||
using Medallion.Threading;
|
||||
|
||||
namespace Elsa.Workflows.ComponentTests.Scenarios.DistributedLockResilience.Mocks;
|
||||
|
||||
/// <summary>
|
||||
/// Test implementation of IDistributedLockProvider that allows simulating transient failures.
|
||||
/// </summary>
|
||||
[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);
|
||||
|
||||
/// <summary>
|
||||
/// Configure failures for a specific lock name prefix. Only locks matching this prefix will fail.
|
||||
/// </summary>
|
||||
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);
|
||||
|
||||
/// <summary>
|
||||
/// Atomically decrements the failure counter if it's greater than 0.
|
||||
/// Returns true if a failure was consumed, false otherwise.
|
||||
/// </summary>
|
||||
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;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,31 @@
|
|||
using Medallion.Threading;
|
||||
|
||||
namespace Elsa.Workflows.ComponentTests.Scenarios.DistributedLockResilience.Mocks;
|
||||
|
||||
/// <summary>
|
||||
/// Test implementation of IDistributedSynchronizationHandle that can simulate failures on disposal.
|
||||
/// </summary>
|
||||
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;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,26 @@
|
|||
using Elsa.Extensions;
|
||||
using Elsa.Workflows.Activities;
|
||||
|
||||
namespace Elsa.Workflows.ComponentTests.Scenarios.DistributedLockResilience.Workflows;
|
||||
|
||||
/// <summary>
|
||||
/// A simple workflow for testing distributed lock resilience.
|
||||
/// </summary>
|
||||
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")
|
||||
}
|
||||
};
|
||||
}
|
||||
}
|
||||
|
|
@ -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<DistributedWorkflowRuntime>();
|
||||
},
|
||||
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<DistributedWorkflowRuntime>();
|
||||
workflowRuntime.UseDistributedRuntime();
|
||||
workflowRuntime.ActivityExecutionLogStore = _ => activityExecutionStore;
|
||||
});
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,87 @@
|
|||
using System.Net.Sockets;
|
||||
|
||||
namespace Elsa.Resilience.Core.UnitTests;
|
||||
|
||||
public class DefaultTransientExceptionStrategyTests
|
||||
{
|
||||
private readonly DefaultTransientExceptionStrategy _strategy = new();
|
||||
|
||||
public static TheoryData<Type> TransientExceptionTypes =>
|
||||
[
|
||||
typeof(HttpRequestException),
|
||||
typeof(TimeoutException),
|
||||
typeof(TaskCanceledException),
|
||||
typeof(IOException),
|
||||
typeof(SocketException),
|
||||
typeof(EndOfStreamException)
|
||||
];
|
||||
|
||||
public static TheoryData<string> 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<string> 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));
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,13 @@
|
|||
<Project Sdk="Microsoft.NET.Sdk">
|
||||
|
||||
<PropertyGroup>
|
||||
<Include>[Elsa.Resilience.Core]*</Include>
|
||||
<Threshold>49</Threshold>
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\..\..\src\common\Elsa.Testing.Shared\Elsa.Testing.Shared.csproj" />
|
||||
<ProjectReference Include="..\..\..\src\modules\Elsa.Resilience.Core\Elsa.Resilience.Core.csproj" />
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
||||
|
|
@ -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<CancellationToken>());
|
||||
}
|
||||
|
||||
[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<CancellationToken>());
|
||||
}
|
||||
}
|
||||
|
|
@ -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<IResilienceStrategyCatalog>();
|
||||
private readonly IExpressionEvaluator _expressionEvaluator = Substitute.For<IExpressionEvaluator>();
|
||||
private readonly ResilienceStrategyConfigEvaluator _evaluator;
|
||||
private readonly ExpressionExecutionContext _context = new(Substitute.For<IServiceProvider>(), 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<CancellationToken>());
|
||||
}
|
||||
|
||||
[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<string>(), Arg.Any<CancellationToken>());
|
||||
}
|
||||
|
||||
[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<object>(Arg.Any<Expression>(), Arg.Any<ExpressionExecutionContext>(), Arg.Any<ExpressionEvaluatorOptions>());
|
||||
}
|
||||
|
||||
[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<CancellationToken>());
|
||||
}
|
||||
|
||||
[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<string>(), Arg.Any<CancellationToken>());
|
||||
}
|
||||
|
||||
[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<CancellationToken>()).Returns(strategy);
|
||||
}
|
||||
|
||||
private void SetupExpressionResult(Expression expression, object? result)
|
||||
{
|
||||
_expressionEvaluator.EvaluateAsync<object>(expression, _context, Arg.Any<ExpressionEvaluatorOptions>()).Returns(result);
|
||||
}
|
||||
}
|
||||
|
|
@ -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<IResilienceStrategy>();
|
||||
strategy.Id.Returns(id);
|
||||
strategy.DisplayName.Returns(displayName);
|
||||
return strategy;
|
||||
}
|
||||
|
||||
public static IResilienceStrategySource CreateStrategySource(params IResilienceStrategy[] strategies)
|
||||
{
|
||||
var source = Substitute.For<IResilienceStrategySource>();
|
||||
source.GetStrategiesAsync(Arg.Any<CancellationToken>()).Returns(strategies);
|
||||
return source;
|
||||
}
|
||||
}
|
||||
|
|
@ -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<ITransientExceptionStrategy>();
|
||||
var strategy2 = Substitute.For<ITransientExceptionStrategy>();
|
||||
strategy1.IsTransient(Arg.Any<Exception>()).Returns(false);
|
||||
strategy2.IsTransient(Arg.Any<Exception>()).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<ITransientExceptionStrategy>();
|
||||
strategy.IsTransient(transientException).Returns(true);
|
||||
strategy.IsTransient(Arg.Is<Exception>(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<ITransientExceptionStrategy> configureDetector,
|
||||
bool expectedResult)
|
||||
{
|
||||
var strategy = Substitute.For<ITransientExceptionStrategy>();
|
||||
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<object[]> 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<object[]> AggregateExceptionTestCases
|
||||
{
|
||||
get
|
||||
{
|
||||
// Scenario 1: One transient inner exception
|
||||
var timeoutException = new TimeoutException("timeout");
|
||||
yield return
|
||||
[
|
||||
new AggregateException("aggregate", timeoutException),
|
||||
(Action<ITransientExceptionStrategy>)(detector =>
|
||||
{
|
||||
detector.IsTransient(Arg.Is<TimeoutException>(e => e.Message == "timeout")).Returns(true);
|
||||
detector.IsTransient(Arg.Is<Exception>(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<ITransientExceptionStrategy>)(detector =>
|
||||
{
|
||||
detector.IsTransient(Arg.Any<Exception>()).Returns(false);
|
||||
}),
|
||||
false
|
||||
];
|
||||
}
|
||||
}
|
||||
|
||||
private static ITransientExceptionStrategy CreateStrategy(params (Exception exception, bool isTransient)[] behaviors)
|
||||
{
|
||||
var detector = Substitute.For<ITransientExceptionStrategy>();
|
||||
foreach (var (exception, isTransient) in behaviors)
|
||||
detector.IsTransient(exception).Returns(isTransient);
|
||||
return detector;
|
||||
}
|
||||
|
||||
private static TransientExceptionDetector CreateDetector(params ITransientExceptionStrategy[] detectors) => new(detectors);
|
||||
}
|
||||
Loading…
Reference in a new issue