Improves workflow deletion and task scheduling (#7088)
* Refactor workflow instance deletion to use `IWorkflowRuntime` for enhanced coordination and separation of concerns. * Remove `EnumerableTypeConverter` and update related usages for serialization. - Deleted the `EnumerableTypeConverter` class and its JSON serialization logic. - Removed associated type descriptor attribute in `DefaultFormattersFeature`. - Updated `ObjectFormatter` to handle collection serialization directly with JSON. * Remove `EnumerableTypeConverter` tests and consolidate serialization logic into `ObjectFormatter`. - Deleted `EnumerableTypeConverterTests` as the related functionality was removed. - Added comprehensive tests in `ObjectFormatterTests` to handle serialization of collections and arrays with JSON. * Add integration tests for `TriggerIndexer` to handle workflows with failing materialization - Introduced comprehensive test scenarios verifying `DeleteTriggersAsync` behavior when workflows fail to load or partially succeed. - Enhanced error handling in `TriggerIndexer` to skip failed workflows while ensuring remaining workflows are processed. * Add exception handling in `TriggerIndexer.DeleteTriggersAsync` and integration tests - Enhanced `DeleteTriggersAsync` with exception handling to skip failed workflows while processing others. - Logged warnings for failed workflows without halting execution. - Added comprehensive integration tests to verify behavior across success, failure, and mixed scenarios. - Refactored tests for improved clarity, maintainability, and consistency. * Add exception handling for `ResumeWorkflowTask` to skip deleted workflow instances - Enhanced `ResumeWorkflowTask.ExecuteAsync` to handle `WorkflowInstanceNotFoundException` gracefully when a scheduled workflow instance is missing. - Logged warnings for skipped executions to improve observability. - Ensured remaining workflows and scheduled tasks are processed seamlessly without disruption. * Add thread safety to `LocalScheduler` to prevent race conditions during concurrent scheduling - Introduced a `lock` object to synchronize access to internal dictionaries. - Resolved `IndexOutOfRangeException` caused by concurrent modifications during startup. - Ensured thread-safe operations in `ScheduleAsync`, `ClearScheduleAsync`, and related methods. - Improved reliability and stability of scheduling under concurrent workloads. * Improve exception handling, thread safety, and workflow instance deletion - Added exception handling in `TriggerIndexer.DeleteTriggersAsync` to skip failed workflows while continuing processing. - Enhanced `ResumeWorkflowTask` to handle missing workflow instances gracefully and log warnings. - Introduced thread synchronization in `LocalScheduler` with `lock` to prevent concurrent access issues. - Implemented and refactored tests to ensure behavior consistency and improve maintainability. - Added component tests for workflow deletion scenarios, covering running, completed, and non-existent workflows. * Add component tests for workflow instance deletion and refactor bulk delete logic - Added comprehensive component tests for workflow instance deletion scenarios (running, completed, bulk, and non-existent instances). - Refactored `BulkDelete` API to use `IWorkflowInstanceManager` for proper cleanup of related records (execution logs, activity executions, bookmarks). * Add integration tests and fakes for `TriggerIndexer` to verify behavior with failing and successful workflows - Introduced `FailingMaterializer` and `WorkingMaterializer` for simulating failing and successful workflow materializations. - Added `TriggerDeletionTestScenario`, `TriggerTestDataBuilder`, and related test data classes to define comprehensive test cases. - Updated `DeleteTriggersAsync` tests with scenarios for materialization failures and mixed success. - Improved test coverage and maintainability with reusable test data builders and utilities.
This commit is contained in:
Sipke Schoorstra
GitHub
parent
2f9eac110e
commit
aee09bb06f
|
|
@ -0,0 +1,153 @@
|
|||
# Configuration Binding Issue: SigningKey Not Bound from appsettings.json
|
||||
|
||||
**Date**: 2025-11-20
|
||||
**Issue**: IdentityTokenOptions.SigningKey property remaining null despite being present in appsettings.json
|
||||
**Related GitHub Issue**: #7019 (Array serialization)
|
||||
|
||||
---
|
||||
|
||||
## Symptoms
|
||||
|
||||
- `IdentityTokenOptions.SigningKey` property was not being bound from configuration
|
||||
- Other properties in the same section (`AccessTokenLifetime`, `RefreshTokenLifetime`) were binding correctly
|
||||
- Manual debugging showed:
|
||||
- ✅ `identityTokenSection["SigningKey"]` returned the correct value
|
||||
- ✅ `configuration.GetSection("Identity:Tokens:SigningKey").Value` returned the correct value
|
||||
- ❌ `identityTokenSection.GetValue<string>("SigningKey")` returned `null`
|
||||
- ❌ `identityTokenSection.Bind(options)` did not populate `SigningKey`
|
||||
- Creating a fresh instance worked: `new IdentityTokenOptions()` could be bound successfully
|
||||
|
||||
## Initial False Leads
|
||||
|
||||
During troubleshooting, several potential causes were investigated but ruled out:
|
||||
|
||||
1. **Property initializers** - Initially suspected that default values like `= "http://elsa.api"` prevented binding, but this was disproven when fresh instances with the same defaults worked
|
||||
2. **Configuration section retrieval** - The section was correctly retrieved and contained the value
|
||||
3. **Nullable vs non-nullable** - Making `SigningKey` nullable didn't resolve the issue
|
||||
4. **JSON encoding issues** - No BOM or hidden characters found in appsettings.json
|
||||
5. **Configuration overrides** - No environment variables or other sources overriding the value
|
||||
6. **Lambda capture issues** - The configuration section was correctly captured and accessible
|
||||
|
||||
## Root Cause
|
||||
|
||||
The actual root cause was **global TypeConverter registration interfering with the configuration binder**.
|
||||
|
||||
In `DefaultFormattersFeature.cs`, the following registration was made:
|
||||
|
||||
```csharp
|
||||
TypeDescriptor.AddAttributes(typeof(IEnumerable),
|
||||
new TypeConverterAttribute(typeof(EnumerableTypeConverter)));
|
||||
```
|
||||
|
||||
This caused a cascade of issues:
|
||||
|
||||
1. **String binding failure**: Since `string` implements `IEnumerable<char>`, the configuration binder attempted to use `EnumerableTypeConverter` for string properties
|
||||
2. **Silent failure**: Even though `EnumerableTypeConverter.CanConvertTo()` correctly returned `false` for strings, the global registration interfered with the fallback mechanism
|
||||
3. **Partial success**: TimeSpan properties worked because they have built-in conversion support that bypassed the string converter issue
|
||||
|
||||
## Solution
|
||||
|
||||
### Phase 1: Remove Global TypeConverter Registration
|
||||
|
||||
**File**: `src/modules/Elsa.Common/Features/DefaultFormattersFeature.cs`
|
||||
|
||||
**Change**: Removed the global `TypeDescriptor.AddAttributes` call for `IEnumerable`
|
||||
|
||||
**Result**:
|
||||
- ✅ Configuration binding now works for all string properties
|
||||
- ❌ Arrays started serializing as `"Int32[] Array"` instead of JSON (exposed GitHub issue #7019)
|
||||
|
||||
### Phase 2: Fix Array Serialization
|
||||
|
||||
**File**: `src/modules/Elsa.Expressions/Helpers/ObjectFormatter.cs`
|
||||
|
||||
**Change**: Added explicit JSON serialization for arrays and collections:
|
||||
|
||||
```csharp
|
||||
// Byte arrays are base64-encoded
|
||||
if (value is byte[] byteArray)
|
||||
return Convert.ToBase64String(byteArray);
|
||||
|
||||
// Memory-like types preserved via TypeDescriptor
|
||||
if (IsMemoryLike(sourceType))
|
||||
{
|
||||
var sourceTypeConverter = TypeDescriptor.GetConverter(underlyingSourceType);
|
||||
if (sourceTypeConverter.CanConvertTo(typeof(string)))
|
||||
return (string?)sourceTypeConverter.ConvertTo(value, typeof(string));
|
||||
return value.ToString();
|
||||
}
|
||||
|
||||
// Serialize arrays and collections to JSON
|
||||
if (value is IEnumerable)
|
||||
{
|
||||
return JsonSerializer.Serialize(value);
|
||||
}
|
||||
```
|
||||
|
||||
**Result**:
|
||||
- ✅ Arrays serialize to JSON: `[1,2,3]` instead of `"Int32[] Array"`
|
||||
- ✅ Configuration binding still works
|
||||
- ✅ No TypeConverter interference with .NET type system
|
||||
|
||||
### Phase 3: Consolidate Logic (Cleanup)
|
||||
|
||||
**File Deleted**: `src/modules/Elsa.Common/Serialization/EnumerableTypeConverter.cs`
|
||||
|
||||
**Rationale**: The `EnumerableTypeConverter` was only used by `ObjectFormatter.Format()`, so consolidating the logic directly into `ObjectFormatter` simplified the codebase and eliminated the need for TypeConverter registration entirely.
|
||||
|
||||
## Key Learnings
|
||||
|
||||
1. **Global TypeConverter registration is dangerous**: It can interfere with .NET's type system, configuration binding, and other framework features
|
||||
2. **String implements IEnumerable**: This often-forgotten fact can cause unexpected behavior with type converters
|
||||
3. **Configuration binder behavior is opaque**: When `.GetValue<T>()` fails but indexer `["key"]` works, it indicates type conversion issues
|
||||
4. **Explicit is better than implicit**: Direct serialization logic in `ObjectFormatter` is clearer and safer than relying on TypeConverter infrastructure
|
||||
5. **Test fresh instances vs captured instances**: Configuration binding through the Options pattern behaves differently than direct `.Bind()` calls
|
||||
|
||||
## Verification
|
||||
|
||||
### Tests Added/Consolidated
|
||||
**File**: `test/integration/Elsa.Common.IntegrationTests/Serialization/ObjectFormatterTests.cs`
|
||||
|
||||
All tests consolidated into a single test class with 9 tests:
|
||||
1. String is preserved as-is
|
||||
2. Byte array is serialized as base64 string
|
||||
3. Integer array is serialized as JSON array
|
||||
4. String array is serialized as JSON array
|
||||
5. String array with multiple elements is serialized as JSON array
|
||||
6. Custom class array is serialized as JSON array
|
||||
7. List of integers is serialized as JSON array
|
||||
8. List with different values is serialized as JSON array
|
||||
9. Null returns null
|
||||
|
||||
### Runtime Verification
|
||||
- ✅ Server starts successfully
|
||||
- ✅ `IdentityTokenOptions.SigningKey` binds correctly from configuration
|
||||
- ✅ No `MissingConfigurationException` thrown
|
||||
- ✅ JWT authentication works
|
||||
- ✅ Workflow variable arrays serialize to JSON
|
||||
- ✅ All 9 integration tests pass
|
||||
|
||||
## Files Modified
|
||||
|
||||
1. `src/modules/Elsa.Common/Features/DefaultFormattersFeature.cs` - Removed global TypeConverter registration
|
||||
2. `src/modules/Elsa.Expressions/Helpers/ObjectFormatter.cs` - Added explicit JSON serialization for collections
|
||||
3. `src/modules/Elsa.Identity/Options/IdentityTokenOptions.cs` - Cleaned up property initializers
|
||||
4. `test/integration/Elsa.Common.IntegrationTests/Serialization/ObjectFormatterTests.cs` - Consolidated all tests (9 total)
|
||||
|
||||
## Files Deleted
|
||||
|
||||
1. `src/modules/Elsa.Common/Serialization/EnumerableTypeConverter.cs` - Consolidated into ObjectFormatter
|
||||
2. `test/integration/Elsa.Common.IntegrationTests/Serialization/EnumerableTypeConverterTests.cs` - Consolidated into ObjectFormatterTests
|
||||
|
||||
---
|
||||
|
||||
## Future Reference
|
||||
|
||||
**Filename Pattern**: `YYYY-MM-DD_issue-description_root-cause.md`
|
||||
|
||||
When encountering similar configuration binding issues:
|
||||
1. Check if the value exists using the indexer: `section["key"]`
|
||||
2. Compare with `GetValue<T>("key")` - if they differ, suspect type conversion issues
|
||||
3. Look for global TypeConverter registrations that might interfere
|
||||
4. Test with a fresh instance to isolate Options pattern behavior
|
||||
5. Check if the property type implements unexpected interfaces (like `string` implementing `IEnumerable<char>`)
|
||||
|
|
@ -0,0 +1,294 @@
|
|||
# Exception Handling Improvements for Trigger Deletion and Scheduled Workflows
|
||||
|
||||
**Date**: November 20, 2025
|
||||
**Issue**: #7077
|
||||
**Branch**: bug/7077
|
||||
**Type**: Bug Fix + Integration Tests
|
||||
|
||||
## Problem
|
||||
|
||||
When deleting triggers, the `TriggerIndexer.DeleteTriggersAsync` method would fail completely if any workflow failed to load due to a missing materializer or other exceptions. This resulted in the error:
|
||||
|
||||
```
|
||||
System.Exception: Provider not found
|
||||
at Elsa.Workflows.Management.Services.WorkflowDefinitionService.MaterializeWorkflowAsync
|
||||
```
|
||||
|
||||
The method would stop processing remaining workflows, leaving triggers in an inconsistent state.
|
||||
|
||||
## Solution
|
||||
|
||||
### 1. Exception Handling (`src/modules/Elsa.Workflows.Runtime/Services/TriggerIndexer.cs`)
|
||||
|
||||
Added a try-catch block in the `DeleteTriggersAsync` method (lines 67-79):
|
||||
|
||||
```csharp
|
||||
foreach (var workflowDefinitionVersionId in workflowDefinitionVersionIds)
|
||||
{
|
||||
try
|
||||
{
|
||||
var workflowGraph = await _workflowDefinitionService.FindWorkflowGraphAsync(
|
||||
workflowDefinitionVersionId, cancellationToken);
|
||||
|
||||
if (workflowGraph == null)
|
||||
continue;
|
||||
|
||||
await DeleteTriggersAsync(workflowGraph.Workflow, cancellationToken);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex,
|
||||
"Failed to load workflow graph for workflow definition version {WorkflowDefinitionVersionId}. " +
|
||||
"Skipping trigger deletion for this workflow.",
|
||||
workflowDefinitionVersionId);
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**Benefits**:
|
||||
- Logs warning with full exception details for debugging
|
||||
- Continues processing remaining workflows instead of failing completely
|
||||
- Maintains system stability when individual workflows have issues
|
||||
|
||||
### 2. Integration Tests (`test/integration/Elsa.Workflows.IntegrationTests/Scenarios/TriggerIndexing/`)
|
||||
|
||||
Created comprehensive integration tests in a new `TriggerIndexing` scenario folder with `Tests.cs`:
|
||||
|
||||
#### Test 1: All Workflows Fail to Load
|
||||
```csharp
|
||||
[Fact(DisplayName = "DeleteTriggersAsync should continue deleting triggers even when workflow fails to load")]
|
||||
public async Task DeleteTriggersAsync_WorkflowFailsToLoad_ContinuesWithOtherTriggers()
|
||||
```
|
||||
|
||||
- Verifies the method completes without throwing when all workflows fail
|
||||
- Tests multiple failure types (JSON serialization errors, provider not found)
|
||||
- Confirms all triggers remain when workflows can't be loaded
|
||||
|
||||
#### Test 2: Mixed Success and Failure
|
||||
```csharp
|
||||
[Fact(DisplayName = "DeleteTriggersAsync processes all workflows even when one fails")]
|
||||
public async Task DeleteTriggersAsync_OneWorkflowFails_ProcessesAllWorkflows()
|
||||
```
|
||||
|
||||
- Tests scenario where one workflow fails (workflow2 with "Provider not found")
|
||||
- Two other workflows load successfully (workflow1 and workflow3)
|
||||
- Verifies triggers for successful workflows are deleted
|
||||
- Confirms failed workflow's trigger remains
|
||||
- **This is the key test** - proves exception handling allows processing to continue
|
||||
|
||||
#### Test Infrastructure
|
||||
|
||||
**Custom Materializers**:
|
||||
- `FailingMaterializer`: Throws "Provider not found" exception to simulate the reported issue
|
||||
- `WorkingMaterializer`: Successfully materializes workflows from a registry for valid test cases
|
||||
|
||||
**Helper Methods** (DRY principle):
|
||||
- `CreateWorkflowDefinition()`: Factory method for workflow definitions
|
||||
- `CreateTrigger()`: Factory method for triggers
|
||||
- `SaveWorkflowsAndTriggersAsync()`: Common save and verification logic
|
||||
|
||||
**Modern C# Features**:
|
||||
- `file` scoped classes for helper materializers (limits visibility)
|
||||
- Primary constructors for `WorkingMaterializer`
|
||||
- Expression-bodied members for simple methods
|
||||
|
||||
## Code Quality Improvements
|
||||
|
||||
### Test Refactoring
|
||||
- Renamed from `TriggerIndexerTests.cs` to `Tests.cs` (project convention)
|
||||
- Reduced test methods from ~90 lines to ~30 lines each
|
||||
- Eliminated duplicate workflow/trigger creation code
|
||||
- Improved readability with array-based test data
|
||||
- Clear Arrange-Act-Assert structure
|
||||
|
||||
### Benefits
|
||||
- **More maintainable**: Changes only need to be made in one place
|
||||
- **More readable**: Test intent is clearer with less boilerplate
|
||||
- **More consistent**: Follows project naming conventions
|
||||
- **Equally functional**: All tests pass with same behavior
|
||||
|
||||
## Test Results
|
||||
|
||||
Both tests pass successfully:
|
||||
|
||||
```
|
||||
Passed! - Failed: 0, Passed: 2, Skipped: 0, Total: 2, Duration: 304 ms
|
||||
|
||||
✓ DeleteTriggersAsync should continue deleting triggers even when workflow fails to load [153 ms]
|
||||
✓ DeleteTriggersAsync processes all workflows even when one fails [49 ms]
|
||||
```
|
||||
|
||||
## 3. Resume Workflow Task Exception Handling (`src/modules/Elsa.Scheduling/Tasks/ResumeWorkflowTask.cs`)
|
||||
|
||||
### Problem
|
||||
|
||||
When a scheduled workflow instance is deleted before its scheduled resume time, the `ResumeWorkflowTask` would throw an unhandled exception:
|
||||
|
||||
```
|
||||
Elsa.Workflows.Runtime.Exceptions.WorkflowInstanceNotFoundException: Workflow instance not found.
|
||||
at Elsa.Workflows.Runtime.LocalWorkflowClient.GetWorkflowInstanceAsync
|
||||
```
|
||||
|
||||
This would cause the scheduled task to fail, even though the workflow instance no longer exists and should simply be skipped.
|
||||
|
||||
### Solution
|
||||
|
||||
Added exception handling to gracefully handle missing workflow instances:
|
||||
|
||||
```csharp
|
||||
public async ValueTask ExecuteAsync(TaskExecutionContext context)
|
||||
{
|
||||
var cancellationToken = context.CancellationToken;
|
||||
var workflowRuntime = context.ServiceProvider.GetRequiredService<IWorkflowRuntime>();
|
||||
var logger = context.ServiceProvider.GetRequiredService<ILogger<ResumeWorkflowTask>>();
|
||||
|
||||
try
|
||||
{
|
||||
var workflowClient = await workflowRuntime.CreateClientAsync(_request.WorkflowInstanceId, cancellationToken);
|
||||
var request = new RunWorkflowInstanceRequest
|
||||
{
|
||||
Input = _request.Input,
|
||||
Properties = _request.Properties,
|
||||
ActivityHandle = _request.ActivityHandle,
|
||||
BookmarkId = _request.BookmarkId
|
||||
};
|
||||
await workflowClient.RunInstanceAsync(request, cancellationToken);
|
||||
}
|
||||
catch (WorkflowInstanceNotFoundException ex)
|
||||
{
|
||||
logger.LogWarning(
|
||||
"Scheduled workflow instance {WorkflowInstanceId} no longer exists and was likely deleted. Skipping execution.",
|
||||
ex.InstanceId);
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**Benefits**:
|
||||
- Scheduled tasks no longer fail when workflow instances are deleted
|
||||
- Clear logging explains why the task was skipped
|
||||
- System continues operating normally
|
||||
- No unnecessary error noise in logs
|
||||
|
||||
## 4. LocalScheduler Thread Safety (`src/modules/Elsa.Scheduling/Services/LocalScheduler.cs`)
|
||||
|
||||
### Problem
|
||||
|
||||
The `LocalScheduler` was experiencing race conditions during concurrent access, particularly during application startup when multiple workflows are being scheduled simultaneously:
|
||||
|
||||
```
|
||||
System.IndexOutOfRangeException: Index was outside the bounds of the array.
|
||||
at System.Collections.Generic.Dictionary`2.TryInsert(TKey key, TValue value, InsertionBehavior behavior)
|
||||
at System.Collections.Generic.Dictionary`2.set_Item(TKey key, TValue value)
|
||||
at Elsa.Scheduling.Services.LocalScheduler.RegisterScheduledTask
|
||||
```
|
||||
|
||||
This exception occurs when multiple threads modify the internal `Dictionary` instances concurrently without synchronization.
|
||||
|
||||
### Root Cause
|
||||
|
||||
Two dictionaries were being accessed concurrently without any thread safety:
|
||||
- `_scheduledTasks`: Maps task names to scheduled tasks
|
||||
- `_scheduledTaskKeys`: Maps scheduled tasks to their keys
|
||||
|
||||
Methods like `ScheduleAsync`, `ClearScheduleAsync`, and internal helper methods all modified these dictionaries without locks.
|
||||
|
||||
### Solution
|
||||
|
||||
Added thread synchronization using a lock object:
|
||||
|
||||
```csharp
|
||||
public class LocalScheduler : IScheduler
|
||||
{
|
||||
private readonly IServiceProvider _serviceProvider;
|
||||
private readonly IDictionary<string, IScheduledTask> _scheduledTasks = new Dictionary<string, IScheduledTask>();
|
||||
private readonly IDictionary<IScheduledTask, ICollection<string>> _scheduledTaskKeys = new Dictionary<IScheduledTask, ICollection<string>>();
|
||||
private readonly object _lock = new(); // Added lock object
|
||||
|
||||
public ValueTask ScheduleAsync(string name, ITask task, ISchedule schedule, IEnumerable<string>? keys = null, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var scheduleContext = new ScheduleContext(_serviceProvider, task);
|
||||
var scheduledTask = schedule.Schedule(scheduleContext);
|
||||
|
||||
lock (_lock) // Protected dictionary access
|
||||
{
|
||||
RegisterScheduledTask(name, scheduledTask, keys);
|
||||
}
|
||||
|
||||
return ValueTask.CompletedTask;
|
||||
}
|
||||
|
||||
public ValueTask ClearScheduleAsync(string name, CancellationToken cancellationToken = default)
|
||||
{
|
||||
lock (_lock) // Protected dictionary access
|
||||
{
|
||||
RemoveScheduledTask(name);
|
||||
}
|
||||
|
||||
return ValueTask.CompletedTask;
|
||||
}
|
||||
|
||||
public ValueTask ClearScheduleAsync(IEnumerable<string> keys, CancellationToken cancellationToken = default)
|
||||
{
|
||||
lock (_lock) // Protected dictionary access
|
||||
{
|
||||
RemoveScheduledTasks(keys);
|
||||
}
|
||||
|
||||
return ValueTask.CompletedTask;
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**Benefits**:
|
||||
- Eliminates race conditions during concurrent scheduling
|
||||
- Prevents `IndexOutOfRangeException` during dictionary modifications
|
||||
- Ensures thread-safe access to internal collections
|
||||
- Simple and efficient locking strategy
|
||||
|
||||
**Implementation Notes**:
|
||||
- Lock is only held during dictionary operations (minimal lock duration)
|
||||
- Schedule creation happens outside the lock to minimize contention
|
||||
- Private methods (`RegisterScheduledTask`, `RemoveScheduledTask`, `RemoveScheduledTasks`) are called within locks, so they don't need additional synchronization
|
||||
|
||||
**Why `lock` instead of `SemaphoreSlim`?**
|
||||
|
||||
The implementation uses a simple `lock` rather than `SemaphoreSlim` for optimal performance:
|
||||
|
||||
| Aspect | lock | SemaphoreSlim |
|
||||
|--------|------|---------------|
|
||||
| **Overhead** | Zero allocation | Heap allocations |
|
||||
| **Complexity** | Automatic release | Manual try/finally |
|
||||
| **Use Case** | Perfect for synchronous ops | Better for async ops |
|
||||
| **Performance** | Compiler optimized | Additional overhead |
|
||||
| **Async Support** | Cannot cross await | Can use WaitAsync |
|
||||
|
||||
**Decision Rationale**:
|
||||
1. ✅ **All critical sections are synchronous** - Only dictionary operations (add/remove/lookup)
|
||||
2. ✅ **Methods return `ValueTask.CompletedTask`** - Not truly async, just implementing interface signature
|
||||
3. ✅ **No await inside critical sections** - Schedule creation happens outside the lock
|
||||
4. ✅ **Zero allocation overhead** - `lock` is ideal for fast synchronous operations
|
||||
|
||||
`SemaphoreSlim` would be appropriate if:
|
||||
- Critical sections contained `await` statements
|
||||
- Methods were truly asynchronous with I/O operations
|
||||
- Cancellation support was needed during lock acquisition
|
||||
|
||||
For this use case, `lock` is the optimal choice for simplicity, performance, and correctness.
|
||||
|
||||
## Files Changed
|
||||
|
||||
- `src/modules/Elsa.Workflows.Runtime/Services/TriggerIndexer.cs` - Added exception handling for workflow loading failures
|
||||
- `test/integration/Elsa.Workflows.IntegrationTests/Scenarios/TriggerIndexing/Tests.cs` - New integration tests
|
||||
- `src/modules/Elsa.Scheduling/Tasks/ResumeWorkflowTask.cs` - Added exception handling for missing workflow instances
|
||||
- `src/modules/Elsa.Scheduling/Services/LocalScheduler.cs` - Added thread synchronization for concurrent access
|
||||
|
||||
## Impact
|
||||
|
||||
- **Stability**: System no longer fails when workflows or instances have issues
|
||||
- **Thread Safety**: LocalScheduler now handles concurrent scheduling operations safely
|
||||
- **Observability**: Warnings are logged with clear explanations for workflow and instance issues
|
||||
- **Reliability**: Other workflows and scheduled tasks continue to be processed correctly
|
||||
- **Coverage**: Comprehensive tests ensure trigger deletion behavior is maintained
|
||||
- **Resilience**:
|
||||
- Scheduled tasks handle deleted workflow instances gracefully
|
||||
- Scheduler handles concurrent access during startup and runtime
|
||||
|
|
@ -0,0 +1,393 @@
|
|||
# Workflow Instance Deletion: Runtime Coordination and Separation of Concerns
|
||||
|
||||
**Date**: 2025-11-20
|
||||
**Branch**: bug/7077
|
||||
**Commit**: 1f326295a38eee236b84d60e69afdf9e3a1eda1a
|
||||
|
||||
---
|
||||
|
||||
## Problem Statement
|
||||
|
||||
Workflow instances were not being properly deleted when using the Delete and BulkDelete API endpoints. The deletion operations were failing silently or inconsistently, particularly for running workflow instances.
|
||||
|
||||
## Root Cause
|
||||
|
||||
The deletion endpoints were directly using `IWorkflowInstanceManager` to delete workflow instances from the database without proper coordination with the workflow runtime. This caused several issues:
|
||||
|
||||
1. **No coordination for running workflows**: Running workflows need to be properly stopped/cancelled before deletion
|
||||
2. **Race conditions**: In distributed scenarios, concurrent operations could cause database inconsistencies
|
||||
3. **Missing cleanup**: Runtime state, bookmarks, and other related data weren't being cleaned up properly
|
||||
4. **Violation of separation of concerns**: API layer was directly manipulating persistence layer without going through the runtime
|
||||
|
||||
## Solution Architecture
|
||||
|
||||
The solution refactors the deletion logic to properly use the workflow runtime infrastructure, ensuring proper coordination and cleanup.
|
||||
|
||||
### Key Design Principles
|
||||
|
||||
1. **Runtime coordination**: All workflow instance operations should go through `IWorkflowRuntime` and `IWorkflowClient`
|
||||
2. **Proper cancellation**: Running workflows must be cancelled before deletion
|
||||
3. **Distributed locking**: Use the same distributed locks as execution to prevent concurrent modifications
|
||||
4. **Separation by status**: Handle running and finished instances differently
|
||||
|
||||
### Implementation Details
|
||||
|
||||
#### 1. Added Delete Method to IWorkflowClient
|
||||
|
||||
**File**: `src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowClient.cs`
|
||||
|
||||
```csharp
|
||||
/// <summary>
|
||||
/// Deletes the workflow instance. This method ensures that the instance is properly stopped before deletion.
|
||||
/// </summary>
|
||||
Task<bool> DeleteAsync(CancellationToken cancellationToken = default);
|
||||
```
|
||||
|
||||
**Rationale**: The `IWorkflowClient` is the proper abstraction for workflow instance operations. Adding deletion here ensures:
|
||||
- Consistent API surface
|
||||
- Proper coordination with other operations
|
||||
- Support for both local and distributed scenarios
|
||||
|
||||
#### 2. Implemented LocalWorkflowClient.DeleteAsync
|
||||
|
||||
**File**: `src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowClient.cs`
|
||||
|
||||
**Key logic**:
|
||||
```csharp
|
||||
public async Task<bool> DeleteAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
// Load the workflow instance (single DB call)
|
||||
var workflowInstance = await TryGetWorkflowInstanceAsync(cancellationToken);
|
||||
if (workflowInstance == null)
|
||||
return false;
|
||||
|
||||
// Cancel if running (ensures proper cleanup)
|
||||
await CancelAsync(workflowInstance, cancellationToken);
|
||||
|
||||
// Delete the workflow instance
|
||||
var filter = new WorkflowInstanceFilter { Id = workflowInstanceId };
|
||||
await workflowInstanceManager.DeleteAsync(filter, cancellationToken);
|
||||
return true;
|
||||
}
|
||||
```
|
||||
|
||||
**Key features**:
|
||||
- **Single DB read**: Loads the instance once and reuses it
|
||||
- **Proper cancellation**: Ensures running workflows are stopped first
|
||||
- **Graceful handling**: Returns `false` if instance doesn't exist
|
||||
- **Clean deletion**: Uses the manager's delete method for proper cleanup
|
||||
|
||||
**Supporting refactoring**:
|
||||
- Extracted `CancelAsync(WorkflowInstance)` private method to avoid loading the instance twice
|
||||
- Added `TryGetWorkflowInstanceAsync()` helper for non-throwing retrieval
|
||||
|
||||
#### 3. Implemented DistributedWorkflowClient.DeleteAsync
|
||||
|
||||
**File**: `src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowClient.cs`
|
||||
|
||||
```csharp
|
||||
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));
|
||||
}
|
||||
```
|
||||
|
||||
**Key feature**: Uses the same distributed lock mechanism (`workflow-instance:{id}`) as execution operations, preventing:
|
||||
- Concurrent modifications during deletion
|
||||
- Race conditions between delete and execute operations
|
||||
- Inconsistent state in distributed scenarios
|
||||
|
||||
#### 4. Refactored Single Instance Delete Endpoint
|
||||
|
||||
**File**: `src/modules/Elsa.Workflows.Api/Endpoints/WorkflowInstances/Delete/Endpoint.cs`
|
||||
|
||||
**Before**:
|
||||
```csharp
|
||||
internal class Delete(IWorkflowInstanceManager store) : ElsaEndpoint<Request>
|
||||
{
|
||||
public override async Task HandleAsync(Request request, CancellationToken cancellationToken)
|
||||
{
|
||||
var filter = new WorkflowInstanceFilter { Id = request.Id };
|
||||
var deleted = await store.DeleteAsync(filter, cancellationToken);
|
||||
// ...
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**After**:
|
||||
```csharp
|
||||
internal class Delete(IWorkflowRuntime workflowRuntime) : ElsaEndpoint<Request>
|
||||
{
|
||||
public override async Task HandleAsync(Request request, CancellationToken cancellationToken)
|
||||
{
|
||||
var client = await workflowRuntime.CreateClientAsync(request.Id, cancellationToken);
|
||||
var deleted = await client.DeleteAsync(cancellationToken);
|
||||
// ...
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**Changes**:
|
||||
- ✅ Uses `IWorkflowRuntime` instead of `IWorkflowInstanceManager`
|
||||
- ✅ Creates a workflow client for proper coordination
|
||||
- ✅ Relies on runtime infrastructure for proper deletion
|
||||
|
||||
#### 5. Refactored Bulk Delete Endpoint
|
||||
|
||||
**File**: `src/modules/Elsa.Workflows.Api/Endpoints/WorkflowInstances/BulkDelete/Endpoint.cs`
|
||||
|
||||
**Strategy**: Split deletion into two phases based on workflow status:
|
||||
|
||||
**Phase 1: Delete Running Instances Individually**
|
||||
```csharp
|
||||
// Step 1: Delete running instances individually (requires coordination by the workflow runtime).
|
||||
var runningFilter = new WorkflowInstanceFilter
|
||||
{
|
||||
Ids = baseFilter.Ids,
|
||||
DefinitionId = baseFilter.DefinitionId,
|
||||
DefinitionIds = baseFilter.DefinitionIds,
|
||||
WorkflowStatus = WorkflowStatus.Running
|
||||
};
|
||||
|
||||
var runningInstanceIds = await workflowInstanceStore.FindManyIdsAsync(runningFilter, cancellationToken);
|
||||
var count = 0L;
|
||||
|
||||
foreach (var instanceId in runningInstanceIds)
|
||||
{
|
||||
var client = await workflowRuntime.CreateClientAsync(instanceId, cancellationToken);
|
||||
var deleted = await client.DeleteAsync(cancellationToken);
|
||||
if (deleted)
|
||||
count++;
|
||||
}
|
||||
```
|
||||
|
||||
**Phase 2: Bulk Delete Finished Instances**
|
||||
```csharp
|
||||
// Step 2: Bulk delete finished instances (no coordination needed).
|
||||
// Use IWorkflowInstanceManager to ensure related records are also deleted.
|
||||
var finishedFilter = new WorkflowInstanceFilter
|
||||
{
|
||||
Ids = baseFilter.Ids,
|
||||
DefinitionId = baseFilter.DefinitionId,
|
||||
DefinitionIds = baseFilter.DefinitionIds,
|
||||
WorkflowStatus = WorkflowStatus.Finished
|
||||
};
|
||||
|
||||
var finishedDeletedCount = await workflowInstanceManager.BulkDeleteAsync(finishedFilter, cancellationToken);
|
||||
count += finishedDeletedCount;
|
||||
```
|
||||
|
||||
**Rationale**:
|
||||
- **Running instances**: Need individual coordination through the runtime (cancellation, locks, cleanup)
|
||||
- **Finished instances**: Can be safely bulk-deleted as they have no active runtime state
|
||||
- **IWorkflowInstanceManager vs IWorkflowInstanceStore**: Using `IWorkflowInstanceManager.BulkDeleteAsync()` ensures that related records (execution logs, bookmarks, activity execution records, etc.) are also deleted, not just the workflow instance record
|
||||
- **Performance**: Avoids unnecessary overhead for finished workflows while ensuring safety for running ones
|
||||
- **Correctness**: Each running workflow gets proper cancellation and distributed locking
|
||||
|
||||
**Trade-offs**:
|
||||
- ✅ **Correctness**: Ensures proper cleanup and coordination
|
||||
- ✅ **Safety**: Prevents race conditions and inconsistencies
|
||||
- ⚠️ **Performance**: Running instances are deleted one-by-one instead of bulk
|
||||
- This is acceptable because:
|
||||
- Running instances are typically a small percentage of total instances
|
||||
- Proper coordination is more important than bulk performance
|
||||
- Finished instances still use efficient bulk deletion
|
||||
|
||||
## Benefits
|
||||
|
||||
### 1. Proper Coordination
|
||||
- Running workflows are properly cancelled before deletion
|
||||
- Runtime state is cleaned up correctly
|
||||
- Bookmarks and related data are removed
|
||||
|
||||
### 2. Distributed Safety
|
||||
- Uses the same distributed locks as execution
|
||||
- Prevents race conditions in multi-node deployments
|
||||
- Ensures consistency across the cluster
|
||||
|
||||
### 3. Separation of Concerns
|
||||
- API layer uses runtime abstractions
|
||||
- No direct persistence manipulation
|
||||
- Easier to maintain and extend
|
||||
|
||||
### 4. Consistent Behavior
|
||||
- Same deletion logic for single and bulk operations
|
||||
- Predictable behavior across all scenarios
|
||||
- Proper error handling and return values
|
||||
|
||||
## Testing Verification
|
||||
|
||||
### Automated Component Tests
|
||||
|
||||
Created comprehensive component tests in `test/component/Elsa.Workflows.ComponentTests/Scenarios/WorkflowInstanceDeletion/`:
|
||||
|
||||
**Test Infrastructure:**
|
||||
1. **Custom Activity**: `LongRunningActivity`
|
||||
- Simulates a long-running workflow operation (3 seconds)
|
||||
- Uses `SignalManager` to notify tests when execution starts
|
||||
- Keeps workflow in memory without hitting a bookmark (crucial for testing the race condition)
|
||||
|
||||
2. **Test Workflows**:
|
||||
- `SuspendingWorkflow`: Uses `LongRunningActivity` to create running workflow instances
|
||||
- `CompletingWorkflow`: Completes immediately for testing finished workflow deletion
|
||||
|
||||
3. **Helper Methods** (following DRY principle):
|
||||
- `StartRunningWorkflowAsync()`: Starts a workflow and waits for it to begin executing
|
||||
- `AssertWorkflowInstanceDeletedAsync()`: Verifies a workflow instance is deleted
|
||||
- Property accessors for common services (`WorkflowRuntime`, `SignalManager`, `WorkflowInstanceStore`)
|
||||
|
||||
**Test Cases:**
|
||||
1. ✅ `DeleteRunningWorkflowInstance_ShouldCancelAndRemove`
|
||||
- Tests deletion of a workflow actively executing in memory
|
||||
- Verifies the instance is deleted and not found afterward
|
||||
|
||||
2. ✅ `DeleteFinishedWorkflowInstance_ShouldRemove`
|
||||
- Tests deletion of a completed workflow instance
|
||||
- Verifies proper cleanup of finished workflows
|
||||
|
||||
3. ✅ `DeleteNonExistentWorkflowInstance_ShouldThrowException`
|
||||
- Tests error handling for non-existent instances
|
||||
- Expects `Refit.ApiException` (404 Not Found)
|
||||
|
||||
4. ✅ `BulkDeleteMixedInstances_ShouldDeleteAll`
|
||||
- Tests bulk deletion of 2 running + 2 finished instances
|
||||
- Verifies all instances are properly deleted
|
||||
|
||||
5. ✅ `DeleteWorkflowInstance_ShouldNotBeResurrectedByPersistence` **(Core #7077 test)**
|
||||
- **Critical test** that validates the distributed locking fix
|
||||
- Starts a workflow with `LongRunningActivity` (stays in memory)
|
||||
- Attempts deletion while workflow is still executing
|
||||
- **Without the fix**: Workflow would be deleted but then resurrected when it completes
|
||||
- **With the fix**: Deletion blocks until workflow completes (distributed lock), preventing resurrection
|
||||
- Verifies the instance remains deleted after workflow completion
|
||||
|
||||
**Key Testing Patterns:**
|
||||
- **SignalManager pattern**: Elegant coordination between workflow execution and test assertions
|
||||
- Handles race conditions gracefully (signal can fire before or after wait)
|
||||
- Uses `ConcurrentDictionary` with `TaskCompletionSource` for thread-safe signaling
|
||||
- No static state to clean up between tests
|
||||
- **WorkflowEvents pattern**: Used to wait for workflow state commits when needed
|
||||
- **DRY principle**: Extracted common patterns into reusable helper methods
|
||||
- **Readable tests**: Focus on "what" not "how" - test intent is clear and concise
|
||||
|
||||
All tests pass successfully, validating the fix works correctly.
|
||||
|
||||
### Manual Testing
|
||||
- ✅ Delete single running workflow instance
|
||||
- ✅ Delete single finished workflow instance
|
||||
- ✅ Delete non-existent workflow instance (returns false)
|
||||
- ✅ Bulk delete mixed running and finished instances
|
||||
- ✅ Bulk delete by definition ID
|
||||
- ✅ Verify proper cancellation before deletion
|
||||
- ✅ Verify distributed locking in multi-node scenario
|
||||
|
||||
### Edge Cases Handled
|
||||
- **Non-existent instance**: Returns `false` gracefully
|
||||
- **Already cancelled/finished**: No-op on cancellation, proceeds to delete
|
||||
- **Concurrent operations**: Distributed lock prevents conflicts
|
||||
- **Partial bulk delete**: Counts successful deletions accurately
|
||||
|
||||
## Performance Considerations
|
||||
|
||||
### Single Delete
|
||||
- **Before**: 1 query (direct delete)
|
||||
- **After**: 2 queries (load + cancel if needed + delete)
|
||||
- **Impact**: Minimal - proper coordination is worth the extra query
|
||||
|
||||
### Bulk Delete
|
||||
- **Running instances**: N queries (where N = number of running instances)
|
||||
- Acceptable because running instances are typically few
|
||||
- Proper coordination requires individual handling
|
||||
- **Finished instances**: 1 query (bulk delete)
|
||||
- Maintains performance for the common case
|
||||
- Most instances are finished, not running
|
||||
|
||||
### Optimization Opportunities
|
||||
- Could add a `BulkCancelAsync` method for running instances
|
||||
- Could batch running instance deletions with parallel processing
|
||||
- Current approach prioritizes correctness over optimization
|
||||
|
||||
## Migration Notes
|
||||
|
||||
### Breaking Changes
|
||||
None - this is a bug fix that doesn't change the API surface.
|
||||
|
||||
### Behavioral Changes
|
||||
1. **Running workflows are now cancelled before deletion**
|
||||
- Previous behavior: Direct delete (potentially inconsistent state)
|
||||
- New behavior: Cancel then delete (proper cleanup)
|
||||
|
||||
2. **Bulk delete now handles running vs finished instances differently**
|
||||
- Previous behavior: Attempted bulk delete of all instances
|
||||
- New behavior: Individual delete for running, bulk for finished
|
||||
|
||||
### Upgrade Path
|
||||
No action required - changes are fully backward compatible.
|
||||
|
||||
## Files Modified
|
||||
|
||||
### Core Implementation
|
||||
|
||||
1. `src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowClient.cs`
|
||||
- Added `DeleteAsync()` method
|
||||
|
||||
2. `src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowClient.cs`
|
||||
- Implemented `DeleteAsync()` with proper cancellation
|
||||
- Refactored `CancelAsync()` to avoid duplicate loading
|
||||
- Added `TryGetWorkflowInstanceAsync()` helper
|
||||
|
||||
3. `src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowClient.cs`
|
||||
- Implemented `DeleteAsync()` with distributed locking
|
||||
|
||||
4. `src/modules/Elsa.Workflows.Api/Endpoints/WorkflowInstances/Delete/Endpoint.cs`
|
||||
- Changed to use `IWorkflowRuntime` instead of `IWorkflowInstanceManager`
|
||||
- Uses workflow client for deletion
|
||||
|
||||
5. `src/modules/Elsa.Workflows.Api/Endpoints/WorkflowInstances/BulkDelete/Endpoint.cs`
|
||||
- Refactored to handle running and finished instances separately
|
||||
- Changed to use `IWorkflowRuntime` for running instances
|
||||
- Changed to use `IWorkflowInstanceManager.BulkDeleteAsync()` for finished instances (ensures related records are deleted)
|
||||
|
||||
### Test Files
|
||||
|
||||
6. `test/component/Elsa.Workflows.ComponentTests/Scenarios/WorkflowInstanceDeletion/Activities/LongRunningActivity.cs` **(NEW)**
|
||||
- Custom activity that simulates long-running operation with 3-second delay
|
||||
- Uses `SignalManager` to notify tests when execution starts
|
||||
- Keeps workflow in memory for testing race conditions
|
||||
|
||||
7. `test/component/Elsa.Workflows.ComponentTests/Scenarios/WorkflowInstanceDeletion/Workflows/SuspendingWorkflow.cs` **(NEW)**
|
||||
- Test workflow using `LongRunningActivity`
|
||||
- Used to create running workflow instances for testing
|
||||
|
||||
8. `test/component/Elsa.Workflows.ComponentTests/Scenarios/WorkflowInstanceDeletion/Workflows/CompletingWorkflow.cs` **(NEW)**
|
||||
- Test workflow that completes immediately
|
||||
- Used for testing finished workflow deletion
|
||||
|
||||
9. `test/component/Elsa.Workflows.ComponentTests/Scenarios/WorkflowInstanceDeletion/WorkflowInstanceDeletionTests.cs` **(NEW)**
|
||||
- 5 comprehensive component tests covering all deletion scenarios
|
||||
- Tests the core #7077 fix (workflow resurrection prevention)
|
||||
- Uses DRY principles with helper methods for maintainability
|
||||
|
||||
## Related Issues
|
||||
|
||||
- **Bug #7077**: Workflow instance not being deleted
|
||||
- Related to proper runtime coordination and distributed locking
|
||||
|
||||
## Future Enhancements
|
||||
|
||||
1. **Batch cancellation**: Add `BulkCancelAsync` for running instances
|
||||
2. **Parallel deletion**: Process running instances in parallel with configurable concurrency
|
||||
3. **Soft delete**: Add option for soft delete with retention period
|
||||
4. **Delete callbacks**: Add hooks for cleanup of related resources
|
||||
5. **Metrics**: Track deletion performance and success rates
|
||||
|
||||
---
|
||||
|
||||
## Key Takeaways
|
||||
|
||||
1. **Always use runtime abstractions**: Don't bypass `IWorkflowClient` to directly manipulate persistence
|
||||
2. **Coordinate running workflows**: Active workflows need proper cancellation before deletion
|
||||
3. **Use distributed locks**: Prevent race conditions in multi-node deployments
|
||||
4. **Separate concerns by status**: Running and finished workflows have different coordination needs
|
||||
5. **Performance vs correctness**: Prioritize correctness, optimize performance within those constraints
|
||||
|
|
@ -4,14 +4,10 @@ using Elsa.Common.RecurringTasks;
|
|||
using Elsa.Expressions.Helpers;
|
||||
using Elsa.Extensions;
|
||||
using Elsa.Features.Services;
|
||||
using Elsa.Identity.Multitenancy;
|
||||
using Elsa.Persistence.EFCore.Extensions;
|
||||
using Elsa.Persistence.EFCore.Modules.Management;
|
||||
using Elsa.Persistence.EFCore.Modules.Runtime;
|
||||
using Elsa.Server.Web.Filters;
|
||||
using Elsa.Tenants.AspNetCore;
|
||||
using Elsa.Tenants.Extensions;
|
||||
using Elsa.Workflows;
|
||||
using Elsa.Workflows.Api;
|
||||
using Elsa.Workflows.CommitStates.Strategies;
|
||||
using Elsa.Workflows.IncidentStrategies;
|
||||
|
|
@ -46,7 +42,7 @@ services
|
|||
.AddWorkflowsFrom<Program>()
|
||||
.UseIdentity(identity =>
|
||||
{
|
||||
identity.TokenOptions = options => identityTokenSection.Bind(options);
|
||||
identity.TokenOptions += options => identityTokenSection.Bind(options);
|
||||
identity.UseConfigurationBasedUserProvider(options => identitySection.Bind(options));
|
||||
identity.UseConfigurationBasedApplicationProvider(options => identitySection.Bind(options));
|
||||
identity.UseConfigurationBasedRoleProvider(options => identitySection.Bind(options));
|
||||
|
|
@ -136,6 +132,7 @@ services.AddCors(cors => cors.AddDefaultPolicy(policy => policy.AllowAnyHeader()
|
|||
// Build the web application.
|
||||
var app = builder.Build();
|
||||
|
||||
|
||||
// Configure the pipeline.
|
||||
if (app.Environment.IsDevelopment())
|
||||
app.UseDeveloperExceptionPage();
|
||||
|
|
|
|||
|
|
@ -1,4 +1,3 @@
|
|||
using System.Collections;
|
||||
using System.ComponentModel;
|
||||
using Elsa.Common.Serialization;
|
||||
using Elsa.Common.Services;
|
||||
|
|
@ -12,7 +11,6 @@ public class DefaultFormattersFeature(IModule module) : FeatureBase(module)
|
|||
{
|
||||
public override void Configure()
|
||||
{
|
||||
TypeDescriptor.AddAttributes(typeof(IEnumerable), new TypeConverterAttribute(typeof(EnumerableTypeConverter)));
|
||||
TypeDescriptor.AddAttributes(typeof(Type), new TypeConverterAttribute(typeof(TypeTypeConverter)));
|
||||
Module.Services.AddSingleton<IFormatter, JsonFormatter>();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,37 +0,0 @@
|
|||
using System.Collections;
|
||||
using System.ComponentModel;
|
||||
using System.Globalization;
|
||||
using System.Text.Json;
|
||||
|
||||
namespace Elsa.Common.Serialization;
|
||||
|
||||
/// <summary>
|
||||
/// A type converter that converts <see cref="IEnumerable"/> types to and from strings using JSON serialization.
|
||||
/// </summary>
|
||||
public class EnumerableTypeConverter : TypeConverter
|
||||
{
|
||||
public override bool CanConvertTo(ITypeDescriptorContext? context, Type? destinationType)
|
||||
{
|
||||
return destinationType == typeof(string) || base.CanConvertTo(context, destinationType);
|
||||
}
|
||||
|
||||
public override object? ConvertTo(ITypeDescriptorContext? context, CultureInfo? culture, object? value, Type destinationType)
|
||||
{
|
||||
if (destinationType == typeof(string) && value is IEnumerable enumerable)
|
||||
{
|
||||
// string, byte[], Memory<T>, ReadOnlyMemory<T>, and Span<T> types implement IEnumerable but should not be serialized as JSON arrays
|
||||
if (value is string or byte[]) return value;
|
||||
|
||||
var valueType = value.GetType();
|
||||
if (valueType.IsGenericType)
|
||||
{
|
||||
var genericTypeDef = valueType.GetGenericTypeDefinition();
|
||||
if (genericTypeDef == typeof(Memory<>) || genericTypeDef == typeof(ReadOnlyMemory<>))
|
||||
return value;
|
||||
}
|
||||
|
||||
return JsonSerializer.Serialize(enumerable);
|
||||
}
|
||||
return base.ConvertTo(context, culture, value, destinationType);
|
||||
}
|
||||
}
|
||||
|
|
@ -1,4 +1,6 @@
|
|||
using System.Collections;
|
||||
using System.ComponentModel;
|
||||
using System.Text.Json;
|
||||
|
||||
namespace Elsa.Expressions.Helpers;
|
||||
|
||||
|
|
@ -26,11 +28,41 @@ public static class ObjectFormatter
|
|||
if (underlyingSourceType == typeof(string))
|
||||
return value as string;
|
||||
|
||||
var sourceTypeConverter = TypeDescriptor.GetConverter(underlyingSourceType);
|
||||
// Byte arrays are base64-encoded for serialization
|
||||
if (value is byte[] byteArray)
|
||||
return Convert.ToBase64String(byteArray);
|
||||
|
||||
if (sourceTypeConverter.CanConvertTo(typeof(string)))
|
||||
return (string?)sourceTypeConverter.ConvertTo(value, typeof(string));
|
||||
// Memory-like types are preserved via TypeDescriptor
|
||||
if (IsMemoryLike(sourceType))
|
||||
{
|
||||
var sourceTypeConverter = TypeDescriptor.GetConverter(underlyingSourceType);
|
||||
if (sourceTypeConverter.CanConvertTo(typeof(string)))
|
||||
return (string?)sourceTypeConverter.ConvertTo(value, typeof(string));
|
||||
|
||||
return value.ToString();
|
||||
}
|
||||
|
||||
// Serialize arrays and collections to JSON instead of "T[] Array" or "(Collection)".
|
||||
// Fixes GitHub issue #7019.
|
||||
if (value is IEnumerable)
|
||||
{
|
||||
return JsonSerializer.Serialize(value);
|
||||
}
|
||||
|
||||
var converter = TypeDescriptor.GetConverter(underlyingSourceType);
|
||||
|
||||
if (converter.CanConvertTo(typeof(string)))
|
||||
return (string?)converter.ConvertTo(value, typeof(string));
|
||||
|
||||
return value.ToString();
|
||||
}
|
||||
|
||||
private static bool IsMemoryLike(Type type)
|
||||
{
|
||||
if (!type.IsGenericType)
|
||||
return false;
|
||||
|
||||
var genericTypeDef = type.GetGenericTypeDefinition();
|
||||
return genericTypeDef == typeof(Memory<>) || genericTypeDef == typeof(ReadOnlyMemory<>);
|
||||
}
|
||||
}
|
||||
|
|
@ -11,6 +11,13 @@ public class LocalScheduler : IScheduler
|
|||
private readonly IDictionary<string, IScheduledTask> _scheduledTasks = new Dictionary<string, IScheduledTask>();
|
||||
private readonly IDictionary<IScheduledTask, ICollection<string>> _scheduledTaskKeys = new Dictionary<IScheduledTask, ICollection<string>>();
|
||||
|
||||
// Note: Using lock instead of SemaphoreSlim because:
|
||||
// 1. All critical sections are synchronous (dictionary operations only)
|
||||
// 2. Methods return ValueTask.CompletedTask (not truly async)
|
||||
// 3. No await inside critical sections
|
||||
// 4. lock has zero allocation overhead, perfect for fast synchronous operations
|
||||
private readonly object _lock = new();
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="LocalScheduler"/> class.
|
||||
/// </summary>
|
||||
|
|
@ -31,21 +38,32 @@ public class LocalScheduler : IScheduler
|
|||
var scheduleContext = new ScheduleContext(_serviceProvider, task);
|
||||
var scheduledTask = schedule.Schedule(scheduleContext);
|
||||
|
||||
RegisterScheduledTask(name, scheduledTask, keys);
|
||||
lock (_lock)
|
||||
{
|
||||
RegisterScheduledTask(name, scheduledTask, keys);
|
||||
}
|
||||
|
||||
return ValueTask.CompletedTask;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public ValueTask ClearScheduleAsync(string name, CancellationToken cancellationToken = default)
|
||||
{
|
||||
RemoveScheduledTask(name);
|
||||
lock (_lock)
|
||||
{
|
||||
RemoveScheduledTask(name);
|
||||
}
|
||||
|
||||
return ValueTask.CompletedTask;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public ValueTask ClearScheduleAsync(IEnumerable<string> keys, CancellationToken cancellationToken = default)
|
||||
{
|
||||
RemoveScheduledTasks(keys);
|
||||
lock (_lock)
|
||||
{
|
||||
RemoveScheduledTasks(keys);
|
||||
}
|
||||
|
||||
return ValueTask.CompletedTask;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,38 +1,40 @@
|
|||
using Elsa.Workflows.Runtime;
|
||||
using Elsa.Workflows.Runtime.Exceptions;
|
||||
using Elsa.Workflows.Runtime.Messages;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace Elsa.Scheduling.Tasks;
|
||||
|
||||
/// <summary>
|
||||
/// A task that resumes a workflow.
|
||||
/// </summary>
|
||||
public class ResumeWorkflowTask : ITask
|
||||
public class ResumeWorkflowTask(ScheduleExistingWorkflowInstanceRequest request) : ITask
|
||||
{
|
||||
private readonly ScheduleExistingWorkflowInstanceRequest _request;
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="ResumeWorkflowTask"/> class.
|
||||
/// </summary>
|
||||
/// <param name="request">The dispatch workflow instance request.</param>
|
||||
public ResumeWorkflowTask(ScheduleExistingWorkflowInstanceRequest request)
|
||||
{
|
||||
_request = request;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async ValueTask ExecuteAsync(TaskExecutionContext context)
|
||||
{
|
||||
var cancellationToken = context.CancellationToken;
|
||||
var workflowRuntime = context.ServiceProvider.GetRequiredService<IWorkflowRuntime>();
|
||||
var workflowClient = await workflowRuntime.CreateClientAsync(_request.WorkflowInstanceId, cancellationToken);
|
||||
var request = new RunWorkflowInstanceRequest
|
||||
var logger = context.ServiceProvider.GetRequiredService<ILogger<ResumeWorkflowTask>>();
|
||||
|
||||
try
|
||||
{
|
||||
Input = _request.Input,
|
||||
Properties = _request.Properties,
|
||||
ActivityHandle = _request.ActivityHandle,
|
||||
BookmarkId = _request.BookmarkId
|
||||
};
|
||||
await workflowClient.RunInstanceAsync(request, cancellationToken);
|
||||
var workflowClient = await workflowRuntime.CreateClientAsync(request.WorkflowInstanceId, cancellationToken);
|
||||
var runWorkflowInstanceRequest = new RunWorkflowInstanceRequest
|
||||
{
|
||||
Input = request.Input,
|
||||
Properties = request.Properties,
|
||||
ActivityHandle = request.ActivityHandle,
|
||||
BookmarkId = request.BookmarkId
|
||||
};
|
||||
await workflowClient.RunInstanceAsync(runWorkflowInstanceRequest, cancellationToken);
|
||||
}
|
||||
catch (WorkflowInstanceNotFoundException ex)
|
||||
{
|
||||
logger.LogWarning(
|
||||
"Scheduled workflow instance {WorkflowInstanceId} no longer exists and was likely deleted. Skipping execution.",
|
||||
ex.InstanceId);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,12 +1,13 @@
|
|||
using Elsa.Abstractions;
|
||||
using Elsa.Workflows.Management;
|
||||
using Elsa.Workflows.Management.Filters;
|
||||
using Elsa.Workflows.Runtime;
|
||||
using JetBrains.Annotations;
|
||||
|
||||
namespace Elsa.Workflows.Api.Endpoints.WorkflowInstances.BulkDelete;
|
||||
|
||||
[PublicAPI]
|
||||
internal class BulkDelete(IWorkflowInstanceManager store) : ElsaEndpoint<Request, Response>
|
||||
internal class BulkDelete(IWorkflowInstanceStore workflowInstanceStore, IWorkflowInstanceManager workflowInstanceManager, IWorkflowRuntime workflowRuntime) : ElsaEndpoint<Request, Response>
|
||||
{
|
||||
public override void Configure()
|
||||
{
|
||||
|
|
@ -19,14 +20,46 @@ internal class BulkDelete(IWorkflowInstanceManager store) : ElsaEndpoint<Request
|
|||
|
||||
public override async Task<Response> ExecuteAsync(Request request, CancellationToken cancellationToken)
|
||||
{
|
||||
var filter = new WorkflowInstanceFilter
|
||||
var baseFilter = new WorkflowInstanceFilter
|
||||
{
|
||||
Ids = request.Ids,
|
||||
DefinitionId = request.WorkflowDefinitionId,
|
||||
DefinitionIds = request.WorkflowDefinitionIds,
|
||||
};
|
||||
var count = await store.BulkDeleteAsync(filter, cancellationToken);
|
||||
|
||||
return new(count);
|
||||
// Step 1: Delete running instances individually (requires coordination by the workflow runtime).
|
||||
var runningFilter = new WorkflowInstanceFilter
|
||||
{
|
||||
Ids = baseFilter.Ids,
|
||||
DefinitionId = baseFilter.DefinitionId,
|
||||
DefinitionIds = baseFilter.DefinitionIds,
|
||||
WorkflowStatus = WorkflowStatus.Running
|
||||
};
|
||||
|
||||
var runningInstanceIds = await workflowInstanceStore.FindManyIdsAsync(runningFilter, cancellationToken);
|
||||
var count = 0L;
|
||||
|
||||
foreach (var instanceId in runningInstanceIds)
|
||||
{
|
||||
var client = await workflowRuntime.CreateClientAsync(instanceId, cancellationToken);
|
||||
var deleted = await client.DeleteAsync(cancellationToken);
|
||||
if (deleted)
|
||||
count++;
|
||||
}
|
||||
|
||||
// Step 2: Bulk delete finished instances (no coordination needed).
|
||||
// Use IWorkflowInstanceManager to ensure related records are also deleted.
|
||||
var finishedFilter = new WorkflowInstanceFilter
|
||||
{
|
||||
Ids = baseFilter.Ids,
|
||||
DefinitionId = baseFilter.DefinitionId,
|
||||
DefinitionIds = baseFilter.DefinitionIds,
|
||||
WorkflowStatus = WorkflowStatus.Finished
|
||||
};
|
||||
|
||||
var finishedDeletedCount = await workflowInstanceManager.BulkDeleteAsync(finishedFilter, cancellationToken);
|
||||
count += finishedDeletedCount;
|
||||
|
||||
return new(count);
|
||||
}
|
||||
}
|
||||
|
|
@ -1,12 +1,11 @@
|
|||
using Elsa.Abstractions;
|
||||
using Elsa.Workflows.Management;
|
||||
using Elsa.Workflows.Management.Filters;
|
||||
using Elsa.Workflows.Runtime;
|
||||
using JetBrains.Annotations;
|
||||
|
||||
namespace Elsa.Workflows.Api.Endpoints.WorkflowInstances.Delete;
|
||||
|
||||
[PublicAPI]
|
||||
internal class Delete(IWorkflowInstanceManager store) : ElsaEndpoint<Request>
|
||||
internal class Delete(IWorkflowRuntime workflowRuntime) : ElsaEndpoint<Request>
|
||||
{
|
||||
public override void Configure()
|
||||
{
|
||||
|
|
@ -16,8 +15,8 @@ internal class Delete(IWorkflowInstanceManager store) : ElsaEndpoint<Request>
|
|||
|
||||
public override async Task HandleAsync(Request request, CancellationToken cancellationToken)
|
||||
{
|
||||
var filter = new WorkflowInstanceFilter { Id = request.Id };
|
||||
var deleted = await store.DeleteAsync(filter, cancellationToken);
|
||||
var client = await workflowRuntime.CreateClientAsync(request.Id, cancellationToken);
|
||||
var deleted = await client.DeleteAsync(cancellationToken);
|
||||
|
||||
if (deleted)
|
||||
await Send.NoContentAsync(cancellationToken);
|
||||
|
|
|
|||
|
|
@ -75,6 +75,12 @@ public class DistributedWorkflowClient(
|
|||
return await _localWorkflowClient.InstanceExistsAsync(cancellationToken);
|
||||
}
|
||||
|
||||
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));
|
||||
}
|
||||
|
||||
private async Task<R> WithLockAsync<R>(Func<Task<R>> func)
|
||||
{
|
||||
var lockKey = $"workflow-instance:{WorkflowInstanceId}";
|
||||
|
|
|
|||
|
|
@ -42,6 +42,11 @@ public interface IWorkflowClient
|
|||
/// Imports the specified <see cref="WorkflowState"/>.
|
||||
/// </summary>
|
||||
Task ImportStateAsync(WorkflowState workflowState, CancellationToken cancellationToken = default);
|
||||
|
||||
|
||||
Task<bool> InstanceExistsAsync(CancellationToken cancellationToken = default);
|
||||
|
||||
/// <summary>
|
||||
/// Deletes the workflow instance. This method ensures that the instance is properly stopped before deletion.
|
||||
/// </summary>
|
||||
Task<bool> DeleteAsync(CancellationToken cancellationToken = default);
|
||||
}
|
||||
|
|
@ -1,5 +1,6 @@
|
|||
using Elsa.Workflows.Management;
|
||||
using Elsa.Workflows.Management.Entities;
|
||||
using Elsa.Workflows.Management.Filters;
|
||||
using Elsa.Workflows.Management.Mappers;
|
||||
using Elsa.Workflows.Management.Options;
|
||||
using Elsa.Workflows.Models;
|
||||
|
|
@ -81,6 +82,11 @@ public class LocalWorkflowClient(
|
|||
public async Task CancelAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
var workflowInstance = await GetWorkflowInstanceAsync(cancellationToken);
|
||||
await CancelAsync(workflowInstance, cancellationToken);
|
||||
}
|
||||
|
||||
private async Task CancelAsync(WorkflowInstance workflowInstance, CancellationToken cancellationToken)
|
||||
{
|
||||
if (workflowInstance.Status != WorkflowStatus.Running) return;
|
||||
var workflowGraph = await GetWorkflowGraphAsync(workflowInstance, cancellationToken);
|
||||
var workflowState = await workflowCanceler.CancelWorkflowAsync(workflowGraph, workflowInstance.WorkflowState, cancellationToken);
|
||||
|
|
@ -106,6 +112,22 @@ public class LocalWorkflowClient(
|
|||
return workflowInstanceManager.ExistsAsync(workflowInstanceId, cancellationToken);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<bool> DeleteAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
// Load the workflow instance (single DB call)
|
||||
var workflowInstance = await TryGetWorkflowInstanceAsync(cancellationToken);
|
||||
if (workflowInstance == null)
|
||||
return false;
|
||||
|
||||
await CancelAsync(workflowInstance, cancellationToken);
|
||||
|
||||
// Delete the workflow instance
|
||||
var filter = new WorkflowInstanceFilter { Id = workflowInstanceId };
|
||||
await workflowInstanceManager.DeleteAsync(filter, cancellationToken);
|
||||
return true;
|
||||
}
|
||||
|
||||
public async Task<RunWorkflowInstanceResponse> RunInstanceAsync(WorkflowInstance workflowInstance, RunWorkflowInstanceRequest request, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var workflowState = workflowInstance.WorkflowState;
|
||||
|
|
@ -167,11 +189,16 @@ public class LocalWorkflowClient(
|
|||
|
||||
private async Task<WorkflowInstance> GetWorkflowInstanceAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
var workflowInstance = await workflowInstanceManager.FindByIdAsync(WorkflowInstanceId, cancellationToken);
|
||||
if (workflowInstance == null) throw new WorkflowInstanceNotFoundException($"Workflow instance not found.", WorkflowInstanceId);
|
||||
var workflowInstance = await TryGetWorkflowInstanceAsync(cancellationToken);
|
||||
if (workflowInstance == null) throw new WorkflowInstanceNotFoundException("Workflow instance not found.", WorkflowInstanceId);
|
||||
return workflowInstance;
|
||||
}
|
||||
|
||||
private Task<WorkflowInstance?> TryGetWorkflowInstanceAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
return workflowInstanceManager.FindByIdAsync(WorkflowInstanceId, cancellationToken);
|
||||
}
|
||||
|
||||
private async Task<WorkflowGraph> GetWorkflowGraphAsync(WorkflowInstance workflowInstance, CancellationToken cancellationToken)
|
||||
{
|
||||
var handle = WorkflowDefinitionHandle.ByDefinitionVersionId(workflowInstance.DefinitionVersionId);
|
||||
|
|
@ -181,7 +208,7 @@ public class LocalWorkflowClient(
|
|||
private async Task<WorkflowGraph> GetWorkflowGraphAsync(WorkflowDefinitionHandle definitionHandle, CancellationToken cancellationToken)
|
||||
{
|
||||
var workflowGraph = await workflowDefinitionService.FindWorkflowGraphAsync(definitionHandle, cancellationToken);
|
||||
if (workflowGraph == null) throw new WorkflowGraphNotFoundException($"Workflow graph not found.", definitionHandle);
|
||||
if (workflowGraph == null) throw new WorkflowGraphNotFoundException("Workflow graph not found.", definitionHandle);
|
||||
return workflowGraph;
|
||||
}
|
||||
}
|
||||
|
|
@ -62,14 +62,21 @@ public class TriggerIndexer : ITriggerIndexer
|
|||
var triggers = (await _triggerStore.FindManyAsync(filter, cancellationToken)).ToList();
|
||||
var workflowDefinitionVersionIds = triggers.Select(x => x.WorkflowDefinitionVersionId).Distinct().ToList();
|
||||
|
||||
foreach (string workflowDefinitionVersionId in workflowDefinitionVersionIds)
|
||||
foreach (var workflowDefinitionVersionId in workflowDefinitionVersionIds)
|
||||
{
|
||||
var workflowGraph = await _workflowDefinitionService.FindWorkflowGraphAsync(workflowDefinitionVersionId, cancellationToken);
|
||||
try
|
||||
{
|
||||
var workflowGraph = await _workflowDefinitionService.FindWorkflowGraphAsync(workflowDefinitionVersionId, cancellationToken);
|
||||
|
||||
if (workflowGraph == null)
|
||||
continue;
|
||||
|
||||
await DeleteTriggersAsync(workflowGraph.Workflow, cancellationToken);
|
||||
if (workflowGraph == null)
|
||||
continue;
|
||||
|
||||
await DeleteTriggersAsync(workflowGraph.Workflow, cancellationToken);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "Failed to load workflow graph for workflow definition version {WorkflowDefinitionVersionId}. Skipping trigger deletion for this workflow.", workflowDefinitionVersionId);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,30 @@
|
|||
using Elsa.Testing.Shared.Services;
|
||||
using Elsa.Workflows.Attributes;
|
||||
|
||||
namespace Elsa.Workflows.ComponentTests.Scenarios.WorkflowInstanceDeletion.Activities;
|
||||
|
||||
/// <summary>
|
||||
/// A custom activity that simulates a long-running operation to test workflow deletion during execution.
|
||||
/// This activity sends a signal when it starts, then delays for a specified duration.
|
||||
/// </summary>
|
||||
[Activity("Elsa.Workflows.ComponentTests", "LongRunningActivity", "A long-running activity for testing")]
|
||||
public class LongRunningActivity : CodeActivity
|
||||
{
|
||||
/// <summary>
|
||||
/// The duration to delay for (in milliseconds)
|
||||
/// </summary>
|
||||
public int DelayMilliseconds { get; set; } = 3000;
|
||||
|
||||
protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
|
||||
{
|
||||
var signalManager = context.GetRequiredService<SignalManager>();
|
||||
var workflowInstanceId = context.WorkflowExecutionContext.Id;
|
||||
|
||||
// Send a signal to notify the test that the activity has started
|
||||
signalManager.Trigger("workflow-started", workflowInstanceId);
|
||||
|
||||
// Simulate a long-running operation by using Task.Delay
|
||||
// This keeps the workflow in memory and executing
|
||||
await Task.Delay(DelayMilliseconds, context.CancellationToken);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,144 @@
|
|||
using Elsa.Api.Client.Resources.WorkflowInstances.Contracts;
|
||||
using Elsa.Common.Models;
|
||||
using Elsa.Workflows.ComponentTests.Abstractions;
|
||||
using Elsa.Workflows.ComponentTests.Fixtures;
|
||||
using Elsa.Workflows.ComponentTests.Scenarios.WorkflowInstanceDeletion.Workflows;
|
||||
using Elsa.Workflows.Management;
|
||||
using Elsa.Workflows.Models;
|
||||
using Elsa.Workflows.Runtime;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
|
||||
namespace Elsa.Workflows.ComponentTests.Scenarios.WorkflowInstanceDeletion;
|
||||
|
||||
/// <summary>
|
||||
/// Tests that verify workflow instance deletion also deletes all related records
|
||||
/// (execution logs, activity execution records, bookmarks, etc.)
|
||||
/// </summary>
|
||||
public class RelatedRecordsDeletionTests(App app) : AppComponentTest(app)
|
||||
{
|
||||
private IWorkflowRuntime WorkflowRuntime => Scope.ServiceProvider.GetRequiredService<IWorkflowRuntime>();
|
||||
private IWorkflowInstanceStore WorkflowInstanceStore => Scope.ServiceProvider.GetRequiredService<IWorkflowInstanceStore>();
|
||||
private IBookmarkStore BookmarkStore => Scope.ServiceProvider.GetRequiredService<IBookmarkStore>();
|
||||
private IActivityExecutionStore ActivityExecutionStore => Scope.ServiceProvider.GetRequiredService<IActivityExecutionStore>();
|
||||
private IWorkflowExecutionLogStore WorkflowExecutionLogStore => Scope.ServiceProvider.GetRequiredService<IWorkflowExecutionLogStore>();
|
||||
|
||||
[Fact(DisplayName = "Delete via API endpoint should remove all related records")]
|
||||
public async Task DeleteViaApiEndpoint_ShouldRemoveAllRelatedRecords()
|
||||
{
|
||||
// Arrange
|
||||
var workflowInstanceId = await CreateWorkflowWithRelatedRecordsAsync();
|
||||
await AssertRelatedRecordsExistAsync(workflowInstanceId);
|
||||
|
||||
// Act
|
||||
var deleteClient = WorkflowServer.CreateApiClient<IWorkflowInstancesApi>();
|
||||
await deleteClient.DeleteAsync(workflowInstanceId);
|
||||
|
||||
// Assert
|
||||
await AssertAllRecordsDeletedAsync(workflowInstanceId);
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "Bulk delete via API endpoint should remove all related records")]
|
||||
public async Task BulkDeleteViaApiEndpoint_ShouldRemoveAllRelatedRecords()
|
||||
{
|
||||
// Arrange - Create 2 workflow instances
|
||||
var workflowInstanceId1 = await CreateWorkflowWithRelatedRecordsAsync();
|
||||
var workflowInstanceId2 = await CreateWorkflowWithRelatedRecordsAsync();
|
||||
|
||||
await AssertRelatedRecordsExistAsync(workflowInstanceId1);
|
||||
await AssertRelatedRecordsExistAsync(workflowInstanceId2);
|
||||
|
||||
var allInstanceIds = new[]
|
||||
{
|
||||
workflowInstanceId1, workflowInstanceId2
|
||||
};
|
||||
|
||||
// Act
|
||||
var bulkDeleteClient = WorkflowServer.CreateApiClient<IWorkflowInstancesApi>();
|
||||
await bulkDeleteClient.BulkDeleteAsync(new()
|
||||
{
|
||||
Ids = allInstanceIds
|
||||
}, CancellationToken.None);
|
||||
|
||||
// Assert
|
||||
foreach (var instanceId in allInstanceIds)
|
||||
{
|
||||
await AssertAllRecordsDeletedAsync(instanceId);
|
||||
}
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "Delete via runtime client should remove all related records")]
|
||||
public async Task DeleteViaRuntimeClient_ShouldRemoveAllRelatedRecords()
|
||||
{
|
||||
// Arrange
|
||||
var workflowInstanceId = await CreateWorkflowWithRelatedRecordsAsync();
|
||||
await AssertRelatedRecordsExistAsync(workflowInstanceId);
|
||||
|
||||
// Act
|
||||
var instanceClient = await WorkflowRuntime.CreateClientAsync(workflowInstanceId);
|
||||
var deleted = await instanceClient.DeleteAsync();
|
||||
|
||||
Assert.True(deleted);
|
||||
|
||||
// Assert
|
||||
await AssertAllRecordsDeletedAsync(workflowInstanceId);
|
||||
}
|
||||
|
||||
private async Task<string> CreateWorkflowWithRelatedRecordsAsync()
|
||||
{
|
||||
var workflowClient = await WorkflowRuntime.CreateClientAsync();
|
||||
var runResponse = await workflowClient.CreateAndRunInstanceAsync(new()
|
||||
{
|
||||
WorkflowDefinitionHandle = WorkflowDefinitionHandle.ByDefinitionId(
|
||||
WorkflowWithRelatedRecords.DefinitionId,
|
||||
VersionOptions.Published)
|
||||
});
|
||||
return runResponse.WorkflowInstanceId;
|
||||
}
|
||||
|
||||
private async Task AssertRelatedRecordsExistAsync(string workflowInstanceId)
|
||||
{
|
||||
var bookmarks = await BookmarkStore.FindManyAsync(new()
|
||||
{
|
||||
WorkflowInstanceId = workflowInstanceId
|
||||
});
|
||||
var activityExecutions = await ActivityExecutionStore.FindManyAsync(new()
|
||||
{
|
||||
WorkflowInstanceId = workflowInstanceId
|
||||
});
|
||||
var executionLogs = await WorkflowExecutionLogStore.FindManyAsync(new()
|
||||
{
|
||||
WorkflowInstanceId = workflowInstanceId
|
||||
}, PageArgs.All);
|
||||
|
||||
Assert.NotEmpty(bookmarks);
|
||||
Assert.NotEmpty(activityExecutions);
|
||||
Assert.NotEmpty(executionLogs.Items);
|
||||
}
|
||||
|
||||
private async Task AssertAllRecordsDeletedAsync(string workflowInstanceId)
|
||||
{
|
||||
var workflowInstance = await WorkflowInstanceStore.FindAsync(new()
|
||||
{
|
||||
Id = workflowInstanceId
|
||||
});
|
||||
Assert.Null(workflowInstance);
|
||||
|
||||
var remainingBookmarks = await BookmarkStore.FindManyAsync(new()
|
||||
{
|
||||
WorkflowInstanceId = workflowInstanceId
|
||||
});
|
||||
Assert.Empty(remainingBookmarks);
|
||||
|
||||
var remainingActivityExecutions = await ActivityExecutionStore.FindManyAsync(new()
|
||||
{
|
||||
WorkflowInstanceId = workflowInstanceId
|
||||
});
|
||||
Assert.Empty(remainingActivityExecutions);
|
||||
|
||||
var remainingExecutionLogs = await WorkflowExecutionLogStore.FindManyAsync(new()
|
||||
{
|
||||
WorkflowInstanceId = workflowInstanceId
|
||||
}, PageArgs.All);
|
||||
Assert.Empty(remainingExecutionLogs.Items);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,151 @@
|
|||
using Elsa.Api.Client.Resources.WorkflowInstances.Contracts;
|
||||
using Elsa.Common.Models;
|
||||
using Elsa.Testing.Shared.Services;
|
||||
using Elsa.Workflows.ComponentTests.Abstractions;
|
||||
using Elsa.Workflows.ComponentTests.Fixtures;
|
||||
using Elsa.Workflows.ComponentTests.Scenarios.WorkflowInstanceDeletion.Workflows;
|
||||
using Elsa.Workflows.Management;
|
||||
using Elsa.Workflows.Models;
|
||||
using Elsa.Workflows.Runtime;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
|
||||
namespace Elsa.Workflows.ComponentTests.Scenarios.WorkflowInstanceDeletion;
|
||||
|
||||
/// <summary>
|
||||
/// Tests that workflow instances can be deleted reliably, even when they are running or being persisted.
|
||||
/// This addresses issue #7077 where in-memory instances could be written back to the DB after deletion.
|
||||
/// </summary>
|
||||
public class WorkflowInstanceDeletionTests(App app) : AppComponentTest(app)
|
||||
{
|
||||
private const string WorkflowStartedSignal = "workflow-started";
|
||||
|
||||
private IWorkflowRuntime WorkflowRuntime => Scope.ServiceProvider.GetRequiredService<IWorkflowRuntime>();
|
||||
private SignalManager SignalManager => Scope.ServiceProvider.GetRequiredService<SignalManager>();
|
||||
private IWorkflowInstanceStore WorkflowInstanceStore => Scope.ServiceProvider.GetRequiredService<IWorkflowInstanceStore>();
|
||||
|
||||
private async Task<string> StartRunningWorkflowAsync()
|
||||
{
|
||||
var workflowClient = await WorkflowRuntime.CreateClientAsync();
|
||||
await workflowClient.CreateAndRunInstanceAsync(new()
|
||||
{
|
||||
WorkflowDefinitionHandle = WorkflowDefinitionHandle.ByDefinitionId(SuspendingWorkflow.DefinitionId, VersionOptions.Published)
|
||||
});
|
||||
|
||||
// Wait for the workflow to signal it has started
|
||||
return await SignalManager.WaitAsync<string>(WorkflowStartedSignal);
|
||||
}
|
||||
|
||||
private async Task AssertWorkflowInstanceDeletedAsync(string workflowInstanceId)
|
||||
{
|
||||
var instance = await WorkflowInstanceStore.FindAsync(new() { Id = workflowInstanceId });
|
||||
Assert.Null(instance);
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "Delete running workflow instance should cancel and remove it")]
|
||||
public async Task DeleteRunningWorkflowInstance_ShouldCancelAndRemove()
|
||||
{
|
||||
// Arrange - Start a workflow that will be running in memory
|
||||
var workflowInstanceId = await StartRunningWorkflowAsync();
|
||||
|
||||
// Act - Delete the running workflow instance using the API
|
||||
var deleteClient = WorkflowServer.CreateApiClient<IWorkflowInstancesApi>();
|
||||
await deleteClient.DeleteAsync(workflowInstanceId);
|
||||
|
||||
// Assert - Instance should be deleted
|
||||
await AssertWorkflowInstanceDeletedAsync(workflowInstanceId);
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "Delete finished workflow instance should remove it")]
|
||||
public async Task DeleteFinishedWorkflowInstance_ShouldRemove()
|
||||
{
|
||||
// Arrange - Create and run a workflow that completes immediately
|
||||
var workflowClient = await WorkflowRuntime.CreateClientAsync();
|
||||
var runResponse = await workflowClient.CreateAndRunInstanceAsync(new()
|
||||
{
|
||||
WorkflowDefinitionHandle = WorkflowDefinitionHandle.ByDefinitionId(CompletingWorkflow.DefinitionId, VersionOptions.Published)
|
||||
});
|
||||
var workflowInstanceId = runResponse.WorkflowInstanceId;
|
||||
|
||||
// Verify the workflow finished
|
||||
var finishedInstance = await WorkflowInstanceStore.FindAsync(new() { Id = workflowInstanceId });
|
||||
Assert.NotNull(finishedInstance);
|
||||
Assert.Equal(WorkflowStatus.Finished, finishedInstance.WorkflowState.Status);
|
||||
|
||||
// Act - Delete the finished workflow instance
|
||||
var deleteClient = WorkflowServer.CreateApiClient<IWorkflowInstancesApi>();
|
||||
await deleteClient.DeleteAsync(workflowInstanceId);
|
||||
|
||||
// Assert - Instance should be deleted
|
||||
await AssertWorkflowInstanceDeletedAsync(workflowInstanceId);
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "Delete non-existent workflow instance should throw exception")]
|
||||
public async Task DeleteNonExistentWorkflowInstance_ShouldThrowException()
|
||||
{
|
||||
// Arrange - Use a non-existent workflow instance ID
|
||||
var nonExistentId = "non-existent-workflow-instance-id";
|
||||
|
||||
// Act & Assert - Should throw an ApiException for non-existent instance (404 Not Found)
|
||||
var deleteClient = WorkflowServer.CreateApiClient<IWorkflowInstancesApi>();
|
||||
await Assert.ThrowsAsync<Refit.ApiException>(async () => await deleteClient.DeleteAsync(nonExistentId));
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "Bulk delete mixed running and finished instances should delete all")]
|
||||
public async Task BulkDeleteMixedInstances_ShouldDeleteAll()
|
||||
{
|
||||
// Arrange - Create 2 running and 2 finished workflow instances
|
||||
var workflowClient = await WorkflowRuntime.CreateClientAsync();
|
||||
|
||||
// Start 2 running instances
|
||||
var runningId1 = await StartRunningWorkflowAsync();
|
||||
var runningId2 = await StartRunningWorkflowAsync();
|
||||
|
||||
// Start 2 finished instances
|
||||
var finishedResult1 = await workflowClient.CreateAndRunInstanceAsync(new()
|
||||
{
|
||||
WorkflowDefinitionHandle = WorkflowDefinitionHandle.ByDefinitionId(CompletingWorkflow.DefinitionId, VersionOptions.Published)
|
||||
});
|
||||
var finishedResult2 = await workflowClient.CreateAndRunInstanceAsync(new()
|
||||
{
|
||||
WorkflowDefinitionHandle = WorkflowDefinitionHandle.ByDefinitionId(CompletingWorkflow.DefinitionId, VersionOptions.Published)
|
||||
});
|
||||
|
||||
var allInstanceIds = new[] { runningId1, runningId2, finishedResult1.WorkflowInstanceId, finishedResult2.WorkflowInstanceId };
|
||||
|
||||
// Act - Bulk delete all instances
|
||||
var bulkDeleteClient = WorkflowServer.CreateApiClient<IWorkflowInstancesApi>();
|
||||
await bulkDeleteClient.BulkDeleteAsync(new() { Ids = allInstanceIds }, CancellationToken.None);
|
||||
|
||||
// Assert - All instances should be deleted
|
||||
foreach (var instanceId in allInstanceIds)
|
||||
{
|
||||
await AssertWorkflowInstanceDeletedAsync(instanceId);
|
||||
}
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "Delete workflow instance should not be resurrected by persistence")]
|
||||
public async Task DeleteWorkflowInstance_ShouldNotBeResurrectedByPersistence()
|
||||
{
|
||||
// This test addresses the core issue in #7077: ensuring that a deleted workflow instance
|
||||
// is not written back to the database by the persistence mechanism while it's still executing.
|
||||
//
|
||||
// The race condition occurs when:
|
||||
// 1. Workflow starts executing (stays in memory)
|
||||
// 2. Delete request comes in
|
||||
// 3. Without the fix: deletion happens, but workflow continues and persists itself again
|
||||
// 4. With the fix: deletion blocks until workflow completes, using distributed locking
|
||||
|
||||
// Arrange - Start a workflow that will be running in memory
|
||||
var workflowInstanceId = await StartRunningWorkflowAsync();
|
||||
|
||||
// Act - Attempt to delete using the runtime client while workflow is executing
|
||||
// With the fix, this will block until the workflow completes (about 3 seconds)
|
||||
var instanceClient = await WorkflowRuntime.CreateClientAsync(workflowInstanceId);
|
||||
var deleted = await instanceClient.DeleteAsync();
|
||||
|
||||
Assert.True(deleted);
|
||||
|
||||
// Assert - Instance should be deleted and NOT resurrected
|
||||
await AssertWorkflowInstanceDeletedAsync(workflowInstanceId);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,22 @@
|
|||
using Elsa.Workflows.Activities;
|
||||
|
||||
namespace Elsa.Workflows.ComponentTests.Scenarios.WorkflowInstanceDeletion.Workflows;
|
||||
|
||||
public class CompletingWorkflow : WorkflowBase
|
||||
{
|
||||
public static readonly string DefinitionId = "completing-workflow-test";
|
||||
|
||||
protected override void Build(IWorkflowBuilder builder)
|
||||
{
|
||||
builder.WithDefinitionId(DefinitionId);
|
||||
|
||||
builder.Root = new Sequence
|
||||
{
|
||||
Activities =
|
||||
{
|
||||
new WriteLine("Starting workflow"),
|
||||
new WriteLine("Completing immediately")
|
||||
}
|
||||
};
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,28 @@
|
|||
using Elsa.Workflows.Activities;
|
||||
using Elsa.Workflows.ComponentTests.Scenarios.WorkflowInstanceDeletion.Activities;
|
||||
|
||||
namespace Elsa.Workflows.ComponentTests.Scenarios.WorkflowInstanceDeletion.Workflows;
|
||||
|
||||
/// <summary>
|
||||
/// A workflow that uses a long-running activity to test deletion during active execution.
|
||||
/// This simulates the race condition where a workflow instance is deleted while still executing in memory.
|
||||
/// </summary>
|
||||
public class SuspendingWorkflow : WorkflowBase
|
||||
{
|
||||
public static readonly string DefinitionId = "suspending-workflow-test";
|
||||
|
||||
protected override void Build(IWorkflowBuilder builder)
|
||||
{
|
||||
builder.WithDefinitionId(DefinitionId);
|
||||
|
||||
builder.Root = new Sequence
|
||||
{
|
||||
Activities =
|
||||
{
|
||||
new WriteLine("Starting workflow"),
|
||||
new LongRunningActivity { DelayMilliseconds = 3000 }, // Runs for 3 seconds in memory
|
||||
new WriteLine("Workflow completed")
|
||||
}
|
||||
};
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,33 @@
|
|||
using Elsa.Scheduling.Activities;
|
||||
using Elsa.Workflows.Activities;
|
||||
|
||||
namespace Elsa.Workflows.ComponentTests.Scenarios.WorkflowInstanceDeletion.Workflows;
|
||||
|
||||
/// <summary>
|
||||
/// A workflow that generates related records: execution logs, activity execution records, and bookmarks.
|
||||
/// This is used to test that all related records are properly deleted when the workflow instance is deleted.
|
||||
/// </summary>
|
||||
public class WorkflowWithRelatedRecords : WorkflowBase
|
||||
{
|
||||
public static readonly string DefinitionId = "workflow-with-related-records-test";
|
||||
|
||||
protected override void Build(IWorkflowBuilder builder)
|
||||
{
|
||||
builder.WithDefinitionId(DefinitionId);
|
||||
|
||||
builder.Root = new Sequence
|
||||
{
|
||||
Activities =
|
||||
{
|
||||
// This will create activity execution records and execution logs
|
||||
new WriteLine("Step 1: Starting workflow"),
|
||||
new WriteLine("Step 2: About to delay"),
|
||||
|
||||
// This will create a bookmark (suspends the workflow)
|
||||
new Delay(TimeSpan.FromHours(1)),
|
||||
|
||||
new WriteLine("Step 3: This should never execute")
|
||||
}
|
||||
};
|
||||
}
|
||||
}
|
||||
|
|
@ -1,52 +0,0 @@
|
|||
using Elsa.Common.Serialization;
|
||||
|
||||
namespace Elsa.Common.IntegrationTests.Serialization;
|
||||
|
||||
public class EnumerableTypeConverterTests
|
||||
{
|
||||
[Fact(DisplayName = "String variable is not serialized as JSON array")]
|
||||
public void StringVariableIsNotSerializedAsJsonArray()
|
||||
{
|
||||
var testString = "Hello World";
|
||||
AssertTypeConverterPreservesValue(testString);
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "Byte array variable is not serialized as JSON array")]
|
||||
public void ByteArrayVariableIsNotSerializedAsJsonArray()
|
||||
{
|
||||
var testByteArray = new byte[] { 0x01, 0x02, 0x03, 0x04, 0xFF };
|
||||
AssertTypeConverterPreservesValue(testByteArray);
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "String array is serialized as JSON array")]
|
||||
public void StringArrayIsSerializedAsJsonArray()
|
||||
{
|
||||
var testArray = new[] { "Element 1", "Element 2", "Element 3" };
|
||||
AssertTypeConverterSerializesToJson(testArray, "[\"Element 1\",\"Element 2\",\"Element 3\"]");
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "List is serialized as JSON array")]
|
||||
public void ListIsSerializedAsJsonArray()
|
||||
{
|
||||
var testList = new List<int> { 1, 2, 3 };
|
||||
AssertTypeConverterSerializesToJson(testList, "[1,2,3]");
|
||||
}
|
||||
|
||||
private void AssertTypeConverterPreservesValue<T>(T expectedValue)
|
||||
{
|
||||
var converter = new EnumerableTypeConverter();
|
||||
var result = converter.ConvertTo(null, null, expectedValue, typeof(string));
|
||||
|
||||
Assert.Equal(expectedValue, result);
|
||||
Assert.IsType<T>(result);
|
||||
}
|
||||
|
||||
private void AssertTypeConverterSerializesToJson<T>(T value, string expectedJson)
|
||||
{
|
||||
var converter = new EnumerableTypeConverter();
|
||||
var result = converter.ConvertTo(null, null, value, typeof(string));
|
||||
|
||||
Assert.IsType<string>(result);
|
||||
Assert.Equal(expectedJson, result);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,107 @@
|
|||
using Elsa.Expressions.Helpers;
|
||||
|
||||
namespace Elsa.Common.IntegrationTests.Serialization;
|
||||
|
||||
/// <summary>
|
||||
/// Tests for ObjectFormatter.Format() to ensure proper serialization of different types:
|
||||
/// - Strings are preserved as-is
|
||||
/// - Byte arrays are base64-encoded
|
||||
/// - Arrays and collections serialize to JSON instead of "T[] Array"
|
||||
/// Related to GitHub issue #7019.
|
||||
/// </summary>
|
||||
public class ObjectFormatterTests
|
||||
{
|
||||
[Fact(DisplayName = "String is preserved as-is")]
|
||||
public void StringIsPreservedAsIs()
|
||||
{
|
||||
var testString = "Hello World";
|
||||
var result = testString.Format();
|
||||
|
||||
Assert.Equal("Hello World", result);
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "Byte array is serialized as base64 string")]
|
||||
public void ByteArrayIsSerializedAsBase64String()
|
||||
{
|
||||
var testByteArray = new byte[] { 0x01, 0x02, 0x03, 0x04, 0xFF };
|
||||
var result = testByteArray.Format();
|
||||
|
||||
// Byte arrays are base64-encoded for serialization
|
||||
Assert.NotNull(result);
|
||||
var expectedBase64 = Convert.ToBase64String(testByteArray);
|
||||
Assert.Equal(expectedBase64, result);
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "Integer array is serialized as JSON array")]
|
||||
public void IntegerArrayIsSerializedAsJsonArray()
|
||||
{
|
||||
var testArray = new[] { 1, 2, 3, 4, 5 };
|
||||
var result = testArray.Format();
|
||||
|
||||
Assert.Equal("[1,2,3,4,5]", result);
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "String array is serialized as JSON array")]
|
||||
public void StringArrayIsSerializedAsJsonArray()
|
||||
{
|
||||
var testArray = new[] { "Hello", "World" };
|
||||
var result = testArray.Format();
|
||||
|
||||
Assert.Equal("[\"Hello\",\"World\"]", result);
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "String array with multiple elements is serialized as JSON array")]
|
||||
public void StringArrayWithMultipleElementsIsSerializedAsJsonArray()
|
||||
{
|
||||
var testArray = new[] { "Element 1", "Element 2", "Element 3" };
|
||||
var result = testArray.Format();
|
||||
|
||||
Assert.Equal("[\"Element 1\",\"Element 2\",\"Element 3\"]", result);
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "Custom class array is serialized as JSON array")]
|
||||
public void CustomClassArrayIsSerializedAsJsonArray()
|
||||
{
|
||||
var testArray = new[] { new TestClass { Name = "Item1" }, new TestClass { Name = "Item2" } };
|
||||
var result = testArray.Format();
|
||||
|
||||
// Should be JSON, not "TestClass[] Array"
|
||||
Assert.NotNull(result);
|
||||
Assert.StartsWith("[", result);
|
||||
Assert.Contains("Item1", result);
|
||||
Assert.Contains("Item2", result);
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "List of integers is serialized as JSON array")]
|
||||
public void ListOfIntegersIsSerializedAsJsonArray()
|
||||
{
|
||||
var testList = new List<int> { 1, 2, 3 };
|
||||
var result = testList.Format();
|
||||
|
||||
Assert.Equal("[1,2,3]", result);
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "List with different values is serialized as JSON array")]
|
||||
public void ListWithDifferentValuesIsSerializedAsJsonArray()
|
||||
{
|
||||
var testList = new List<int> { 10, 20, 30 };
|
||||
var result = testList.Format();
|
||||
|
||||
Assert.Equal("[10,20,30]", result);
|
||||
}
|
||||
|
||||
[Fact(DisplayName = "Null returns null")]
|
||||
public void NullReturnsNull()
|
||||
{
|
||||
object? testValue = null;
|
||||
var result = testValue.Format();
|
||||
|
||||
Assert.Null(result);
|
||||
}
|
||||
|
||||
private class TestClass
|
||||
{
|
||||
public string Name { get; set; } = string.Empty;
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -0,0 +1,18 @@
|
|||
using Elsa.Workflows.Activities;
|
||||
using Elsa.Workflows.Management;
|
||||
using Elsa.Workflows.Management.Entities;
|
||||
|
||||
namespace Elsa.Workflows.IntegrationTests.Scenarios.TriggerIndexing.Fakes;
|
||||
|
||||
/// <summary>
|
||||
/// A custom materializer that throws an exception when materialization is attempted.
|
||||
/// This simulates the "Provider not found" scenario.
|
||||
/// </summary>
|
||||
public class FailingMaterializer : IWorkflowMaterializer
|
||||
{
|
||||
public const string MaterializerName = "FailingMaterializer";
|
||||
public string Name => MaterializerName;
|
||||
|
||||
public ValueTask<Workflow> MaterializeAsync(WorkflowDefinition definition, CancellationToken cancellationToken = default) =>
|
||||
ValueTask.FromException<Workflow>(new("Provider not found"));
|
||||
}
|
||||
|
|
@ -0,0 +1,22 @@
|
|||
using Elsa.Workflows.Activities;
|
||||
using Elsa.Workflows.Management;
|
||||
using Elsa.Workflows.Management.Entities;
|
||||
|
||||
namespace Elsa.Workflows.IntegrationTests.Scenarios.TriggerIndexing.Fakes;
|
||||
|
||||
/// <summary>
|
||||
/// A custom materializer that successfully materializes workflows from a registry.
|
||||
/// </summary>
|
||||
public class WorkingMaterializer(Dictionary<string, Workflow> workflowRegistry) : IWorkflowMaterializer
|
||||
{
|
||||
public const string MaterializerName = "WorkingMaterializer";
|
||||
public string Name => MaterializerName;
|
||||
|
||||
public ValueTask<Workflow> MaterializeAsync(WorkflowDefinition definition, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (workflowRegistry.TryGetValue(definition.Id, out var workflow))
|
||||
return ValueTask.FromResult(workflow);
|
||||
|
||||
throw new InvalidOperationException($"Workflow {definition.Id} not found in registry");
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,13 @@
|
|||
namespace Elsa.Workflows.IntegrationTests.Scenarios.TriggerIndexing.TestData;
|
||||
|
||||
/// <summary>
|
||||
/// Represents a test scenario for trigger deletion with expected outcomes.
|
||||
/// </summary>
|
||||
public record TriggerDeletionTestScenario
|
||||
{
|
||||
public required string DisplayName { get; init; }
|
||||
public required WorkflowConfig[] Workflows { get; init; }
|
||||
public required string[] ExpectedRemainingTriggerIds { get; init; }
|
||||
|
||||
public override string ToString() => DisplayName;
|
||||
}
|
||||
|
|
@ -0,0 +1,57 @@
|
|||
using Elsa.Workflows.Management.Entities;
|
||||
using Elsa.Workflows.Runtime.Entities;
|
||||
|
||||
namespace Elsa.Workflows.IntegrationTests.Scenarios.TriggerIndexing.TestData;
|
||||
|
||||
/// <summary>
|
||||
/// Fluent builder for constructing test data with workflows and triggers.
|
||||
/// </summary>
|
||||
public class TriggerTestDataBuilder
|
||||
{
|
||||
private readonly List<(string workflowId, string definitionId, string materializerName)> _workflows = [];
|
||||
|
||||
public TriggerTestDataBuilder AddWorkflow(string workflowId, string definitionId, string materializerName)
|
||||
{
|
||||
_workflows.Add((workflowId, definitionId, materializerName));
|
||||
return this;
|
||||
}
|
||||
|
||||
public (WorkflowDefinition[] Workflows, StoredTrigger[] Triggers) Build()
|
||||
{
|
||||
var workflows = _workflows
|
||||
.Select(config => CreateWorkflowDefinition(config.workflowId, config.definitionId, config.materializerName))
|
||||
.ToArray();
|
||||
|
||||
var triggers = _workflows
|
||||
.Select((config, index) => CreateTrigger($"trigger{index + 1}", config.definitionId, config.workflowId, $"act{index + 1}"))
|
||||
.ToArray();
|
||||
|
||||
return (workflows, triggers);
|
||||
}
|
||||
|
||||
private static WorkflowDefinition CreateWorkflowDefinition(string id, string definitionId, string materializerName)
|
||||
{
|
||||
return new()
|
||||
{
|
||||
Id = id,
|
||||
DefinitionId = definitionId,
|
||||
Version = 1,
|
||||
IsPublished = true,
|
||||
IsLatest = true,
|
||||
MaterializerName = materializerName,
|
||||
StringData = "{\"root\":{\"type\":\"Elsa.Sequence\",\"id\":\"seq1\",\"version\":1,\"activities\":[]}}"
|
||||
};
|
||||
}
|
||||
|
||||
private static StoredTrigger CreateTrigger(string id, string definitionId, string versionId, string activityId)
|
||||
{
|
||||
return new()
|
||||
{
|
||||
Id = id,
|
||||
WorkflowDefinitionId = definitionId,
|
||||
WorkflowDefinitionVersionId = versionId,
|
||||
Name = "TestTrigger",
|
||||
ActivityId = activityId
|
||||
};
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,6 @@
|
|||
namespace Elsa.Workflows.IntegrationTests.Scenarios.TriggerIndexing.TestData;
|
||||
|
||||
/// <summary>
|
||||
/// Configuration for a workflow in a test scenario.
|
||||
/// </summary>
|
||||
public record WorkflowConfig(string Id, string DefinitionId, string MaterializerName, bool ShouldSucceed);
|
||||
|
|
@ -0,0 +1,140 @@
|
|||
using Elsa.Testing.Shared;
|
||||
using Elsa.Workflows.Activities;
|
||||
using Elsa.Workflows.IntegrationTests.Scenarios.TriggerIndexing.Fakes;
|
||||
using Elsa.Workflows.IntegrationTests.Scenarios.TriggerIndexing.TestData;
|
||||
using Elsa.Workflows.Management;
|
||||
using Elsa.Workflows.Management.Entities;
|
||||
using Elsa.Workflows.Runtime;
|
||||
using Elsa.Workflows.Runtime.Entities;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Xunit.Abstractions;
|
||||
|
||||
namespace Elsa.Workflows.IntegrationTests.Scenarios.TriggerIndexing;
|
||||
|
||||
public class Tests
|
||||
{
|
||||
private readonly IServiceProvider _services;
|
||||
private readonly ITriggerIndexer _triggerIndexer;
|
||||
private readonly ITriggerStore _triggerStore;
|
||||
private readonly IWorkflowDefinitionStore _workflowDefinitionStore;
|
||||
private readonly Dictionary<string, Workflow> _workflowRegistry = new();
|
||||
|
||||
public Tests(ITestOutputHelper testOutputHelper)
|
||||
{
|
||||
_services = new TestApplicationBuilder(testOutputHelper).ConfigureServices(services =>
|
||||
{
|
||||
services.AddSingleton<IWorkflowMaterializer, FailingMaterializer>();
|
||||
services.AddSingleton<IWorkflowMaterializer>(_ => new WorkingMaterializer(_workflowRegistry));
|
||||
})
|
||||
.Build();
|
||||
|
||||
_triggerIndexer = _services.GetRequiredService<ITriggerIndexer>();
|
||||
_triggerStore = _services.GetRequiredService<ITriggerStore>();
|
||||
_workflowDefinitionStore = _services.GetRequiredService<IWorkflowDefinitionStore>();
|
||||
}
|
||||
|
||||
[Theory(DisplayName = "DeleteTriggersAsync handles workflow materialization failures correctly")]
|
||||
[MemberData(nameof(GetTriggerDeletionTestData))]
|
||||
public async Task DeleteTriggersAsync_HandlesWorkflowFailures(TriggerDeletionTestScenario scenario)
|
||||
{
|
||||
// Arrange
|
||||
var builder = new TriggerTestDataBuilder();
|
||||
|
||||
foreach (var workflow in scenario.Workflows)
|
||||
{
|
||||
builder.AddWorkflow(workflow.Id, workflow.DefinitionId, workflow.MaterializerName);
|
||||
|
||||
// Register working workflows in the registry
|
||||
if (workflow.ShouldSucceed)
|
||||
RegisterWorkflowInRegistry(workflow.Id, workflow.DefinitionId);
|
||||
}
|
||||
|
||||
var testData = builder.Build();
|
||||
await SetupTestScenarioAsync(testData);
|
||||
|
||||
// Act
|
||||
await DeleteTriggersAsync();
|
||||
|
||||
// Assert
|
||||
await AssertRemainingTriggers(scenario.ExpectedRemainingTriggerIds);
|
||||
}
|
||||
|
||||
public static IEnumerable<object[]> GetTriggerDeletionTestData()
|
||||
{
|
||||
// Scenario 1: All workflows fail to load - all triggers should remain
|
||||
yield return
|
||||
[
|
||||
new TriggerDeletionTestScenario
|
||||
{
|
||||
DisplayName = "All workflows fail to load",
|
||||
Workflows =
|
||||
[
|
||||
new("workflow1", "def1", "Json", ShouldSucceed: false),
|
||||
new("workflow2", "def2", FailingMaterializer.MaterializerName, ShouldSucceed: false),
|
||||
new("workflow3", "def3", "Json", ShouldSucceed: false)
|
||||
],
|
||||
ExpectedRemainingTriggerIds = ["trigger1", "trigger2", "trigger3"]
|
||||
}
|
||||
];
|
||||
|
||||
// Scenario 2: One workflow fails, others succeed - only the failing workflow's trigger remains
|
||||
yield return
|
||||
[
|
||||
new TriggerDeletionTestScenario
|
||||
{
|
||||
DisplayName = "One workflow fails, others succeed",
|
||||
Workflows =
|
||||
[
|
||||
new("workflow1", "def1", WorkingMaterializer.MaterializerName, ShouldSucceed: true),
|
||||
new("workflow2", "def2", FailingMaterializer.MaterializerName, ShouldSucceed: false),
|
||||
new("workflow3", "def3", WorkingMaterializer.MaterializerName, ShouldSucceed: true)
|
||||
],
|
||||
ExpectedRemainingTriggerIds = ["trigger2"]
|
||||
}
|
||||
];
|
||||
}
|
||||
|
||||
private async Task AssertRemainingTriggers(string[] expectedTriggerIds)
|
||||
{
|
||||
var remainingTriggerIds = await GetRemainingTriggerIdsAsync();
|
||||
Assert.Equal(expectedTriggerIds.OrderBy(id => id), remainingTriggerIds);
|
||||
}
|
||||
|
||||
private async Task SaveWorkflowsAndTriggersAsync(WorkflowDefinition[] workflows, StoredTrigger[] triggers)
|
||||
{
|
||||
foreach (var workflow in workflows)
|
||||
await _workflowDefinitionStore.SaveAsync(workflow);
|
||||
|
||||
foreach (var trigger in triggers)
|
||||
await _triggerStore.SaveAsync(trigger);
|
||||
|
||||
var allTriggers = await _triggerStore.FindManyAsync(new());
|
||||
Assert.Equal(triggers.Length, allTriggers.Count());
|
||||
}
|
||||
|
||||
|
||||
private void RegisterWorkflowInRegistry(string workflowId, string definitionId)
|
||||
{
|
||||
_workflowRegistry[workflowId] = new()
|
||||
{
|
||||
Identity = new(definitionId, 1, workflowId)
|
||||
};
|
||||
}
|
||||
|
||||
private async Task SetupTestScenarioAsync((WorkflowDefinition[] Workflows, StoredTrigger[] Triggers) testData)
|
||||
{
|
||||
await _services.PopulateRegistriesAsync();
|
||||
await SaveWorkflowsAndTriggersAsync(testData.Workflows, testData.Triggers);
|
||||
}
|
||||
|
||||
private async Task DeleteTriggersAsync()
|
||||
{
|
||||
await _triggerIndexer.DeleteTriggersAsync(new());
|
||||
}
|
||||
|
||||
private async Task<string[]> GetRemainingTriggerIdsAsync()
|
||||
{
|
||||
var remainingTriggers = await _triggerStore.FindManyAsync(new());
|
||||
return remainingTriggers.Select(t => t.Id).OrderBy(id => id).ToArray();
|
||||
}
|
||||
}
|
||||
|
|
@ -103,10 +103,16 @@ public class ScheduledCronTaskTests : IDisposable
|
|||
SetupCronParser(DefaultNow, DefaultNow.AddMinutes(5)); // First: delay=0, Second: proper future time
|
||||
|
||||
// Act
|
||||
CreateScheduledTask();
|
||||
var task = CreateScheduledTask();
|
||||
|
||||
// Assert - Should call GetNextOccurrence twice (once for initial delay=0, once for retry)
|
||||
_cronParser.Received(2).GetNextOccurrence(DefaultCronExpression);
|
||||
// Dispose immediately to prevent timer from firing
|
||||
((IDisposable)task).Dispose();
|
||||
|
||||
// Assert - Should call GetNextOccurrence at least twice (initial + retry)
|
||||
// May be called more if timer fires before disposal in rare race conditions
|
||||
_cronParser.Received().GetNextOccurrence(DefaultCronExpression);
|
||||
var calls = _cronParser.ReceivedCalls().Count(c => c.GetMethodInfo().Name == nameof(_cronParser.GetNextOccurrence));
|
||||
Assert.True(calls >= 2, $"Expected at least 2 calls to GetNextOccurrence, but got {calls}");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
|
|
@ -117,10 +123,16 @@ public class ScheduledCronTaskTests : IDisposable
|
|||
SetupCronParser(DefaultNow.AddMinutes(-1), DefaultNow.AddMinutes(5)); // First: past, Second: future
|
||||
|
||||
// Act
|
||||
CreateScheduledTask();
|
||||
var task = CreateScheduledTask();
|
||||
|
||||
// Assert - Should call GetNextOccurrence twice
|
||||
_cronParser.Received(2).GetNextOccurrence(DefaultCronExpression);
|
||||
// Dispose immediately to prevent timer from firing
|
||||
((IDisposable)task).Dispose();
|
||||
Thread.Sleep(5); // Brief wait to ensure disposal completes
|
||||
|
||||
// Assert - Should call GetNextOccurrence at least twice
|
||||
_cronParser.Received().GetNextOccurrence(DefaultCronExpression);
|
||||
var calls = _cronParser.ReceivedCalls().Count(c => c.GetMethodInfo().Name == nameof(_cronParser.GetNextOccurrence));
|
||||
Assert.True(calls >= 2, $"Expected at least 2 calls to GetNextOccurrence, but got {calls}");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
|
|
@ -132,10 +144,17 @@ public class ScheduledCronTaskTests : IDisposable
|
|||
SetupCronParser(DefaultNow); // Both calls return exactly now (delay = 0)
|
||||
|
||||
// Act - This should not crash and should set up a timer with minimum delay
|
||||
CreateScheduledTask();
|
||||
var task = CreateScheduledTask();
|
||||
|
||||
// Assert - Should call GetNextOccurrence twice (initial + retry) and log warning once (on final attempt)
|
||||
_cronParser.Received(2).GetNextOccurrence(DefaultCronExpression);
|
||||
// Dispose immediately to prevent timer from firing and recursing
|
||||
((IDisposable)task).Dispose();
|
||||
Thread.Sleep(5); // Brief wait to ensure disposal completes
|
||||
|
||||
// Assert - Should call GetNextOccurrence at least twice (initial + retry)
|
||||
// May be called more if timer fires before disposal and triggers Schedule() again
|
||||
_cronParser.Received().GetNextOccurrence(DefaultCronExpression);
|
||||
var calls = _cronParser.ReceivedCalls().Count(c => c.GetMethodInfo().Name == nameof(_cronParser.GetNextOccurrence));
|
||||
Assert.True(calls >= 2, $"Expected at least 2 calls to GetNextOccurrence, but got {calls}");
|
||||
AssertWarningLogged();
|
||||
}
|
||||
|
||||
|
|
@ -148,7 +167,10 @@ public class ScheduledCronTaskTests : IDisposable
|
|||
SetupCronParser(DefaultNow.AddMilliseconds(-100), DefaultNow.AddMilliseconds(-50));
|
||||
|
||||
// Act - Should handle negative delay gracefully
|
||||
CreateScheduledTask();
|
||||
var task = CreateScheduledTask();
|
||||
|
||||
// Dispose immediately to prevent timer from firing
|
||||
((IDisposable)task).Dispose();
|
||||
|
||||
// Assert - Should log a warning and still set up timer
|
||||
AssertWarningLogged();
|
||||
|
|
@ -182,8 +204,6 @@ public class ScheduledCronTaskTests : IDisposable
|
|||
{
|
||||
((IDisposable)task).Dispose();
|
||||
}
|
||||
// Small delay to ensure any timer callbacks have completed
|
||||
Thread.Sleep(10);
|
||||
_serviceProvider.Dispose();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -96,10 +96,16 @@ public class ScheduledRecurringTaskTests : IDisposable
|
|||
var startAt = DefaultNow; // First: delay=0
|
||||
|
||||
// Act
|
||||
CreateScheduledTask(startAt: startAt);
|
||||
var task = CreateScheduledTask(startAt: startAt);
|
||||
|
||||
// Assert - System clock should be called twice (once for initial delay=0, once for retry)
|
||||
_ = _systemClock.Received(2).UtcNow;
|
||||
// Dispose immediately to prevent timer from firing
|
||||
((IDisposable)task).Dispose();
|
||||
|
||||
// Assert - System clock should be called at least twice (initial + retry)
|
||||
// May be called more if timer fires before disposal in rare race conditions
|
||||
_ = _systemClock.Received().UtcNow;
|
||||
var calls = _systemClock.ReceivedCalls().Count(c => c.GetMethodInfo().Name == "get_UtcNow");
|
||||
Assert.True(calls >= 2, $"Expected at least 2 calls to UtcNow, but got {calls}");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
|
|
@ -110,10 +116,16 @@ public class ScheduledRecurringTaskTests : IDisposable
|
|||
var startAt = DefaultNow.AddMinutes(-1); // Past time
|
||||
|
||||
// Act
|
||||
CreateScheduledTask(startAt: startAt);
|
||||
var task = CreateScheduledTask(startAt: startAt);
|
||||
|
||||
// Assert - System clock should be called twice
|
||||
_ = _systemClock.Received(2).UtcNow;
|
||||
// Dispose immediately to prevent timer from firing
|
||||
((IDisposable)task).Dispose();
|
||||
|
||||
// Assert - System clock should be called at least twice (initial + retry)
|
||||
// May be called more if timer fires before disposal in rare race conditions
|
||||
_ = _systemClock.Received().UtcNow;
|
||||
var calls = _systemClock.ReceivedCalls().Count(c => c.GetMethodInfo().Name == "get_UtcNow");
|
||||
Assert.True(calls >= 2, $"Expected at least 2 calls to UtcNow, but got {calls}");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
|
|
@ -125,10 +137,17 @@ public class ScheduledRecurringTaskTests : IDisposable
|
|||
var startAt = DefaultNow; // Both calls return exactly now (delay = 0)
|
||||
|
||||
// Act - This should not crash and should set up a timer with minimum delay
|
||||
CreateScheduledTask(startAt: startAt);
|
||||
var task = CreateScheduledTask(startAt: startAt);
|
||||
|
||||
// Assert - Should call UtcNow twice (initial + retry) and log warning
|
||||
_ = _systemClock.Received(2).UtcNow;
|
||||
// Dispose immediately to prevent timer from firing and recursing
|
||||
((IDisposable)task).Dispose();
|
||||
Thread.Sleep(5); // Brief wait to ensure disposal completes
|
||||
|
||||
// Assert - Should call UtcNow at least twice (initial + retry)
|
||||
// May be called more if timer fires before disposal
|
||||
_ = _systemClock.Received().UtcNow;
|
||||
var calls = _systemClock.ReceivedCalls().Count(c => c.GetMethodInfo().Name == "get_UtcNow");
|
||||
Assert.True(calls >= 2, $"Expected at least 2 calls to UtcNow, but got {calls}");
|
||||
AssertWarningLogged();
|
||||
}
|
||||
|
||||
|
|
@ -141,7 +160,10 @@ public class ScheduledRecurringTaskTests : IDisposable
|
|||
var startAt = DefaultNow.AddMilliseconds(-100); // Negative delay
|
||||
|
||||
// Act - Should handle negative delay gracefully
|
||||
CreateScheduledTask(startAt: startAt);
|
||||
var task = CreateScheduledTask(startAt: startAt);
|
||||
|
||||
// Dispose immediately to prevent timer from firing
|
||||
((IDisposable)task).Dispose();
|
||||
|
||||
// Assert - Should log a warning and still set up timer
|
||||
AssertWarningLogged();
|
||||
|
|
@ -170,8 +192,6 @@ public class ScheduledRecurringTaskTests : IDisposable
|
|||
{
|
||||
((IDisposable)task).Dispose();
|
||||
}
|
||||
// Small delay to ensure any timer callbacks have completed
|
||||
Thread.Sleep(10);
|
||||
_serviceProvider.Dispose();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue