elsa-core/src/modules/Elsa.Workflows.Runtime/Services/BookmarkResumer.cs
Sipke Schoorstra c549f49dfb
Fix race condition when sending same stimuli (#6895)
* Introduce `WorkflowResumer` service and deprecate `BookmarkResumer`.

- Adds `IWorkflowResumer` and its implementation for workflow resumption.
- Marks `BookmarkResumer` and related interfaces as obsolete.
- Refactors dependent services to use `WorkflowResumer`.
- Enhances `ResumeBookmarkRequest` to include `ActivityInstanceId`.
- Updates logging and queue handling logic to align with the new resumption approach.

* Update lock key prefix in `WorkflowResumer` for consistency with service naming.

* Add exception handling for distributed lock acquisition in `WorkflowResumer`

- Wrap distributed lock logic with `try-catch` to handle `TimeoutException`.
- Improve error message when lock acquisition fails due to timeout.
- Preserve existing workflow resumption behavior and logging.

* Update src/modules/Elsa.Workflows.Runtime/Filters/BookmarkFilter.cs

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Optimize `BookmarkFilter` hashing logic for improved performance and readability.

* Merge remote-tracking branch 'origin/enh/locked-bookmark-resumption-2' into enh/locked-bookmark-resumption-2

* Remove unused variable and redundant line breaks for cleaner code.

* Clean up logging configuration by removing unused debug log levels.

* Update src/modules/Elsa.Workflows.Runtime/Services/WorkflowResumer.cs

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Handle collections in `BookmarkFilter` hashing to ensure determinism and improve compatibility.

* Refactor `BookmarkFilter` hashing logic for clarity and consistency.

* Improve `TimeoutException` handling with a more descriptive message in `WorkflowResumer`.

* Update src/modules/Elsa.Workflows.Runtime/Filters/BookmarkFilter.cs

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Simplify `BookmarkFilter` by utilizing `using` directives and refining type references.

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
2025-09-11 22:30:25 +02:00

108 lines
4.6 KiB
C#

using Elsa.Workflows.Helpers;
using Elsa.Workflows.Runtime.Exceptions;
using Elsa.Workflows.Runtime.Filters;
using Elsa.Workflows.Runtime.Messages;
using Elsa.Workflows.Runtime.Options;
using Microsoft.Extensions.Logging;
namespace Elsa.Workflows.Runtime;
/// <inheritdoc />
[Obsolete("Use WorkflowResumer instead.")]
public class BookmarkResumer(IWorkflowRuntime workflowRuntime, IBookmarkStore bookmarkStore, IStimulusHasher stimulusHasher, ILogger<BookmarkResumer> logger) : IBookmarkResumer
{
/// <inheritdoc />
public Task<ResumeBookmarkResult> ResumeAsync<TActivity>(object stimulus, ResumeBookmarkOptions? options = null, CancellationToken cancellationToken = default) where TActivity : IActivity
{
return ResumeAsync<TActivity>(stimulus, null, options, cancellationToken);
}
/// <inheritdoc />
public async Task<ResumeBookmarkResult> ResumeAsync<TActivity>(object stimulus, string? workflowInstanceId = null, ResumeBookmarkOptions? options = null, CancellationToken cancellationToken = default) where TActivity : IActivity
{
var activityTypeName = ActivityTypeNameHelper.GenerateTypeName<TActivity>();
var stimulusHash = stimulusHasher.Hash(activityTypeName, stimulus);
var bookmarkFilter = new BookmarkFilter
{
Name = activityTypeName,
WorkflowInstanceId = workflowInstanceId,
Hash = stimulusHash,
};
return await ResumeAsync(bookmarkFilter, options, cancellationToken);
}
/// <inheritdoc />
public async Task<ResumeBookmarkResult> ResumeAsync(string bookmarkId, IDictionary<string, object> input, CancellationToken cancellationToken = default)
{
var bookmarkFilter = new BookmarkFilter
{
BookmarkId = bookmarkId
};
var options = new ResumeBookmarkOptions
{
Input = input
};
return await ResumeAsync(bookmarkFilter, options, cancellationToken);
}
/// <inheritdoc />
public async Task<ResumeBookmarkResult> ResumeAsync<TActivity>(string bookmarkId, ResumeBookmarkOptions? options = null, CancellationToken cancellationToken = default) where TActivity : IActivity
{
var activityTypeName = ActivityTypeNameHelper.GenerateTypeName<TActivity>();
var bookmarkFilter = new BookmarkFilter
{
Name = activityTypeName,
BookmarkId = bookmarkId
};
return await ResumeAsync(bookmarkFilter, options, cancellationToken);
}
public async Task<ResumeBookmarkResult> ResumeAsync(ResumeBookmarkRequest request, CancellationToken cancellationToken = default)
{
var runRequest = new RunWorkflowInstanceRequest
{
Input = request.Input,
Properties = request.Properties,
ActivityHandle = request.ActivityHandle,
BookmarkId = request.BookmarkId
};
var workflowInstanceId = request.WorkflowInstanceId;
var workflowClient = await workflowRuntime.CreateClientAsync(workflowInstanceId, cancellationToken);
var response = await workflowClient.RunInstanceAsync(runRequest, cancellationToken);
return ResumeBookmarkResult.Found(response);
}
/// <inheritdoc />
public async Task<ResumeBookmarkResult> ResumeAsync(BookmarkFilter filter, ResumeBookmarkOptions? options = null, CancellationToken cancellationToken = default)
{
var bookmark = await bookmarkStore.FindAsync(filter, cancellationToken);
if (bookmark == null)
{
logger.LogDebug("Bookmark not found in store for filter {@Filter}", filter);
return ResumeBookmarkResult.NotFound();
}
var workflowClient = await workflowRuntime.CreateClientAsync(bookmark.WorkflowInstanceId, cancellationToken);
var runRequest = new RunWorkflowInstanceRequest
{
Input = options?.Input,
Properties = options?.Properties,
BookmarkId = bookmark.Id
};
try
{
var response = await workflowClient.RunInstanceAsync(runRequest, cancellationToken);
logger.LogDebug("Resumed workflow instance {WorkflowInstanceId} with bookmark {BookmarkId}", bookmark.WorkflowInstanceId, bookmark.Id);
return ResumeBookmarkResult.Found(response);
}
catch (WorkflowInstanceNotFoundException)
{
// The workflow instance does not (yet) exist in the DB.
logger.LogDebug("No workflow instance with ID {WorkflowInstanceId} found for bookmark {BookmarkId} at this time.", bookmark.WorkflowInstanceId, bookmark.Id);
return ResumeBookmarkResult.NotFound();
}
}
}