SemianiakaVY/elsa-core
1
0
Fork 0
Code Issues Pull requests Actions 1 Packages Projects Releases Wiki Activity

Refactor workflow scheduling with deferred task execution

Browse source 
Reorder pipeline steps to ensure deferred tasks are used correctly in `WorkflowExecutionPipelineBuilderExtensions.cs`. Add `workflowInstanceManager` to `ScheduleBackgroundActivitiesMiddleware` and defer background activity scheduling to execute after state is saved for better consistency.
This commit is contained in:
Sipke Schoorstra 2024-07-07 23:37:58 +02:00
parent 2a31f598a7
commit 0816443dd6
2 changed files with 47 additions and 34 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
src/modules/Elsa.Workflows.Runtime/Extensions/WorkflowExecutionPipelineBuilderExtensions.cs
View file

@ -13,8 +13,8 @@ public static class WorkflowExecutionPipelineBuilderExtensions
public static IWorkflowExecutionPipelineBuilder UseDefaultPipeline(this IWorkflowExecutionPipelineBuilder pipelineBuilder) =>
pipelineBuilder
.Reset()
.UseBackgroundActivities()
.UseDeferredActivityTasks()
.UseBackgroundActivities()
.UseBookmarkPersistence()
.UseActivityExecutionLogPersistence()
.UseWorkflowExecutionLogPersistence()

79
src/modules/Elsa.Workflows.Runtime/Middleware/Workflows/ScheduleBackgroundActivitiesMiddleware.cs
View file

@ -1,5 +1,6 @@
using Elsa.Extensions;
using Elsa.Workflows.Contracts;
using Elsa.Workflows.Management;
using Elsa.Workflows.Pipelines.WorkflowExecution;
using Elsa.Workflows.Runtime.Entities;
using Elsa.Workflows.Runtime.Middleware.Activities;
@ -14,7 +15,8 @@ public class ScheduleBackgroundActivitiesMiddleware(
WorkflowMiddlewareDelegate next,
IBackgroundActivityScheduler backgroundActivityScheduler,
IStimulusHasher stimulusHasher,
IBookmarkStore bookmarkStore)
IBookmarkStore bookmarkStore,
IWorkflowInstanceManager workflowInstanceManager)
: WorkflowExecutionMiddleware(next)
{
/// <inheritdoc />
@ -29,42 +31,53 @@ public class ScheduleBackgroundActivitiesMiddleware(
var scheduledBackgroundActivities = workflowExecutionContext
.TransientProperties
.GetOrAdd(BackgroundActivityInvokerMiddleware.BackgroundActivitySchedulesKey, () => new List<ScheduledBackgroundActivity>());
if (scheduledBackgroundActivities.Count == 0)
return;
foreach (var scheduledBackgroundActivity in scheduledBackgroundActivities)
context.DeferTask(async () =>
{
// Schedule the background activity.
var jobId = await backgroundActivityScheduler.ScheduleAsync(scheduledBackgroundActivity, cancellationToken);
// Select the bookmark associated with the background activity.
var bookmark = workflowExecutionContext.Bookmarks.First(x => x.Id == scheduledBackgroundActivity.BookmarkId);
var stimulus = bookmark.GetPayload<BackgroundActivityStimulus>();
// Store the created job ID.
workflowExecutionContext.Bookmarks.Remove(bookmark);
stimulus.JobId = jobId;
bookmark = bookmark with
// Commit state.
await workflowInstanceManager.SaveAsync(context, cancellationToken);
foreach (var scheduledBackgroundActivity in scheduledBackgroundActivities)
{
Payload = bookmark.Payload,
Hash = stimulusHasher.Hash(bookmark.Name, stimulus)
};
workflowExecutionContext.Bookmarks.Add(bookmark);
// Schedule the background activity.
var jobId = await backgroundActivityScheduler.ScheduleAsync(scheduledBackgroundActivity, cancellationToken);
// Update the bookmark.
var storedBookmark = new StoredBookmark
{
Id = bookmark.Id,
TenantId = tenantId,
ActivityInstanceId = bookmark.ActivityInstanceId,
ActivityTypeName = bookmark.Name,
Hash = bookmark.Hash,
WorkflowInstanceId = workflowExecutionContext.Id,
CreatedAt = bookmark.CreatedAt,
CorrelationId = workflowExecutionContext.CorrelationId,
Payload = bookmark.Payload,
Metadata = bookmark.Metadata,
};
// Select the bookmark associated with the background activity.
var bookmark = workflowExecutionContext.Bookmarks.First(x => x.Id == scheduledBackgroundActivity.BookmarkId);
var stimulus = bookmark.GetPayload<BackgroundActivityStimulus>();
// Store the created job ID.
workflowExecutionContext.Bookmarks.Remove(bookmark);
stimulus.JobId = jobId;
bookmark = bookmark with
{
Payload = bookmark.Payload,
Hash = stimulusHasher.Hash(bookmark.Name, stimulus)
};
workflowExecutionContext.Bookmarks.Add(bookmark);
// Update the bookmark.
var storedBookmark = new StoredBookmark
{
Id = bookmark.Id,
TenantId = tenantId,
ActivityInstanceId = bookmark.ActivityInstanceId,
ActivityTypeName = bookmark.Name,
Hash = bookmark.Hash,
WorkflowInstanceId = workflowExecutionContext.Id,
CreatedAt = bookmark.CreatedAt,
CorrelationId = workflowExecutionContext.CorrelationId,
Payload = bookmark.Payload,
Metadata = bookmark.Metadata,
};
await bookmarkStore.SaveAsync(storedBookmark, cancellationToken);
}
await bookmarkStore.SaveAsync(storedBookmark, cancellationToken);
}
});
}
}