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>

Author: Copilot

test/component/Elsa.Workflows.ComponentTests/Helpers/Fixtures/WorkflowServer.cs

@@ -152,6 +152,7 @@ protected override void ConfigureWebHost(IWebHostBuilder builder)
                 .AddWorkflowsProvider<TestWorkflowProvider>()
                 .AddNotificationHandlersFrom<WorkflowEventHandlers>()
                 .Decorate<IChangeTokenSignaler, EventPublishingChangeTokenSignaler>()
+                .Decorate<IDistributedLockProvider, TestDistributedLockProvider>()
                 ;
         });
     }

test/component/Elsa.Workflows.ComponentTests/Scenarios/DistributedLockResilience/DistributedLockResilienceTests.cs

@@ -1,8 +1,14 @@
 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.Management;
+using Elsa.Workflows.Models;
+using Elsa.Workflows.Runtime;
+using Elsa.Workflows.Runtime.Messages;
 using Medallion.Threading;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
@@ -15,7 +21,9 @@ public class DistributedLockResilienceTests(App app) : AppComponentTest(app)
 {
     private const int MaxRetryAttempts = 3;
 
-    private TestDistributedLockProvider MockProvider => Scope.ServiceProvider.GetRequiredService<TestDistributedLockProvider>();
+    // 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;
@@ -59,6 +67,107 @@ public async Task AcquireLockWithRetry_TransientFailureOnRelease_ShouldLogButNot
         Assert.Equal(1, MockProvider.ReleaseAttemptCount);
     }
 
+    [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 RunInstanceAsync_TransientLockFailures_RetriesCorrectly(int failureCount, int expectedAttemptCount, bool shouldThrow)
+    {
+        // Arrange
+        var workflowRuntime = Scope.ServiceProvider.GetRequiredService<IWorkflowRuntime>();
+        var workflowClient = await workflowRuntime.CreateClientAsync();
+        
+        var createRequest = new CreateAndRunWorkflowInstanceRequest
+        {
+            WorkflowDefinitionHandle = WorkflowDefinitionHandle.ByDefinitionId(SimpleWorkflow.DefinitionId, VersionOptions.Latest)
+        };
+
+        // Reset and configure failures right before the operation to minimize interference from background workers
+        MockProvider.Reset();
+        MockProvider.FailAcquisitionTimes(failureCount);
+
+        // Act & Assert
+        if (shouldThrow)
+        {
+            // Should throw after exhausting retries
+            await Assert.ThrowsAsync<TimeoutException>(async () => 
+                await workflowClient.CreateAndRunInstanceAsync(createRequest));
+        }
+        else
+        {
+            // Should succeed after retries
+            var response = await workflowClient.CreateAndRunInstanceAsync(createRequest);
+            Assert.NotNull(response);
+            Assert.NotNull(response.WorkflowInstanceId);
+        }
+
+        // Verify retries occurred (allow for some background noise, but verify minimum attempts)
+        Assert.True(MockProvider.AcquisitionAttemptCount >= expectedAttemptCount, 
+            $"Expected at least {expectedAttemptCount} acquisition attempts, but got {MockProvider.AcquisitionAttemptCount}");
+    }
+
+    [Fact]
+    public async Task RunInstanceAsync_TransientLockFailureOnSecondRun_RetriesCorrectly()
+    {
+        // Arrange
+        var workflowRuntime = Scope.ServiceProvider.GetRequiredService<IWorkflowRuntime>();
+        
+        // First run - reset and don't inject failures
+        MockProvider.Reset();
+        var workflowClient = await workflowRuntime.CreateClientAsync();
+        
+        var createRequest = new CreateAndRunWorkflowInstanceRequest
+        {
+            WorkflowDefinitionHandle = WorkflowDefinitionHandle.ByDefinitionId(SimpleWorkflow.DefinitionId, VersionOptions.Latest)
+        };
+        
+        var firstResponse = await workflowClient.CreateAndRunInstanceAsync(createRequest);
+        Assert.NotNull(firstResponse.WorkflowInstanceId);
+        
+        // Verify first run succeeded (at least 1 attempt, possibly more from background workers)
+        Assert.True(MockProvider.AcquisitionAttemptCount >= 1);
+        
+        // Setup second workflow with failure - reset right before to minimize background noise
+        MockProvider.Reset();
+        MockProvider.FailAcquisitionTimes(2); // Fail twice, succeed on third
+        
+        var secondClient = await workflowRuntime.CreateClientAsync();
+        var secondResponse = await secondClient.CreateAndRunInstanceAsync(createRequest);
+        
+        // Act & Assert - Second run should succeed after retries
+        Assert.NotNull(secondResponse);
+        Assert.NotNull(secondResponse.WorkflowInstanceId);
+        Assert.NotEqual(firstResponse.WorkflowInstanceId, secondResponse.WorkflowInstanceId);
+        
+        // Verify retries occurred (at least 3 attempts: 2 failures + 1 success, possibly more from background)
+        Assert.True(MockProvider.AcquisitionAttemptCount >= 3,
+            $"Expected at least 3 acquisition attempts, but got {MockProvider.AcquisitionAttemptCount}");
+    }
+
+    [Fact]
+    public async Task RunInstanceAsync_TransientReleaseFailure_ShouldLogButNotThrow()
+    {
+        // Arrange
+        MockProvider.Reset();
+        MockProvider.FailReleaseOnce();
+        
+        var workflowRuntime = Scope.ServiceProvider.GetRequiredService<IWorkflowRuntime>();
+        var workflowClient = await workflowRuntime.CreateClientAsync();
+        
+        var createRequest = new CreateAndRunWorkflowInstanceRequest
+        {
+            WorkflowDefinitionHandle = WorkflowDefinitionHandle.ByDefinitionId(SimpleWorkflow.DefinitionId, VersionOptions.Latest)
+        };
+
+        // Act - Release failure should be caught and logged, not thrown
+        var response = await workflowClient.CreateAndRunInstanceAsync(createRequest);
+        
+        // Assert
+        Assert.NotNull(response);
+        Assert.NotNull(response.WorkflowInstanceId);
+        Assert.Equal(1, MockProvider.ReleaseAttemptCount);
+    }
+
     private async Task<IDistributedSynchronizationHandle?> AcquireLockWithRetryAsync(string lockName) =>
         await RetryPipeline.ExecuteAsync(async ct =>
             await MockProvider.AcquireLockAsync(lockName, LockOptions.LockAcquisitionTimeout, ct),

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;
+
+/// <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")
+            }
+        };
+    }
+}