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

Fix an issue with re-scheduling scopes while workflow is Suspended

Browse source 
This solves an issue when a workflow uses a fork activity where multiple branches get scheduled, but one of them contains a blocking activity, putting the workflow in the Suspended state. But as the other scheduled activities execute and "unwind" when the leaf node finished, their current scope weren't being re-scheduled.
This commit is contained in:
Sipke Schoorstra 2021-08-10 19:41:58 +02:00
parent 39712f95c5
commit 65e35dd3cb
1 changed files with 48 additions and 16 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

64
src/core/Elsa.Core/Handlers/RescheduleBranchingActivitiesAndContainers.cs
View file

@ -3,6 +3,7 @@ using System.Threading;
using System.Threading.Tasks;
using Elsa.Events;
using Elsa.Models;
using Elsa.Services.Models;
using MediatR;
namespace Elsa.Handlers
@ -16,34 +17,65 @@ namespace Elsa.Handlers
{
var activityExecutionContext = notification.ActivityExecutionContext;
var workflowExecutionContext = activityExecutionContext.WorkflowExecutionContext;
// Check to see if a suspension / completion has been instructed. If so, do nothing.
if (workflowExecutionContext.HasScheduledActivities || workflowExecutionContext.Status != WorkflowStatus.Running)
return Task.CompletedTask;
// Check if we are within a scope.
if (!workflowExecutionContext.WorkflowInstance.Scopes.Any())
return Task.CompletedTask;
// Select the scope that is within the activity's inbound trajectory. Include current activity for consideration (it might itself be a scope).
var activityId = activityExecutionContext.ActivityId;
var inboundActivityIds = new[] { activityId }.Concat(workflowExecutionContext.GetInboundActivityPath(activityId)).ToHashSet();
var workflowInstance = workflowExecutionContext.WorkflowInstance;
var scopes = workflowInstance.Scopes;
var scope = scopes.FirstOrDefault(x => inboundActivityIds.Contains(x.ActivityId));
var scope = GetScope(activityExecutionContext);
// Check to see if a suspension / completion has been instructed.
if (workflowExecutionContext.HasScheduledActivities || workflowExecutionContext.Status != WorkflowStatus.Running)
{
// If any blocking activity is within the current scope, exit.
var anyInScope = scope == null || IsAnyBlockingActivityInScope(workflowExecutionContext, scope.ActivityId);
if(anyInScope || workflowExecutionContext.Status == WorkflowStatus.Finished)
return Task.CompletedTask;
}
if (scope == null)
{
// Not scope in scope.
// No scope to reschedule.
return Task.CompletedTask;
}
// Re-schedule the current scope activity.
scopes.Remove(scope);
workflowInstance.Scopes.Remove(scope);
workflowInstance.ActivityData.GetItem(scope.ActivityId)!.SetState("Unwinding", true);
workflowExecutionContext.ScheduleActivity(scope.ActivityId);
return Task.CompletedTask;
}
private bool IsAnyBlockingActivityInScope(WorkflowExecutionContext workflowExecutionContext, string scopeActivityId)
{
// Get all blocking activity IDs.
var blockingActivityIds = workflowExecutionContext.WorkflowInstance.BlockingActivities.Select(x => x.ActivityId).ToHashSet();
// For each blocking activity, check if it is within the current scope (taking the first one).
foreach (var blockingActivityId in blockingActivityIds)
{
var inboundActivityIds = new[] { blockingActivityId }.Concat(workflowExecutionContext.GetInboundActivityPath(blockingActivityId)).ToHashSet();
if (inboundActivityIds.Contains(scopeActivityId))
return true;
}
return false;
}
private ActivityScope? GetScope(ActivityExecutionContext context)
{
var workflowExecutionContext = context.WorkflowExecutionContext;
// Check if we are within a scope.
if (!workflowExecutionContext.WorkflowInstance.Scopes.Any())
return null;
// Select the scope that is within the activity's inbound trajectory. Include current activity for consideration (it might itself be a scope).
var activityId = context.ActivityId;
var inboundActivityIds = new[] { activityId }.Concat(workflowExecutionContext.GetInboundActivityPath(activityId)).ToHashSet();
var workflowInstance = workflowExecutionContext.WorkflowInstance;
var scopes = workflowInstance.Scopes;
var scope = scopes.FirstOrDefault(x => inboundActivityIds.Contains(x.ActivityId));
return scope;
}
}
}