More correlation fixes
This commit is contained in:
Sipke Schoorstra
parent
26bf218da6
commit
ea40ad1121
|
|
@ -11,14 +11,12 @@ namespace Elsa.Activities.AzureServiceBus.Bookmarks
|
|||
{
|
||||
}
|
||||
|
||||
public QueueMessageReceivedBookmark(string queueName, string? correlationId = default)
|
||||
public QueueMessageReceivedBookmark(string queueName)
|
||||
{
|
||||
QueueName = queueName;
|
||||
CorrelationId = correlationId;
|
||||
}
|
||||
|
||||
public string QueueName { get; set; } = default!;
|
||||
public string? CorrelationId { get; set; }
|
||||
}
|
||||
|
||||
public class QueueMessageReceivedBookmarkProvider : BookmarkProvider<QueueMessageReceivedBookmark, AzureServiceBusQueueMessageReceived>
|
||||
|
|
@ -28,8 +26,7 @@ namespace Elsa.Activities.AzureServiceBus.Bookmarks
|
|||
{
|
||||
Result(new QueueMessageReceivedBookmark
|
||||
{
|
||||
QueueName = (await context.ReadActivityPropertyAsync(x => x.QueueName, cancellationToken))!,
|
||||
CorrelationId = context.ActivityExecutionContext.WorkflowExecutionContext.CorrelationId
|
||||
QueueName = (await context.ReadActivityPropertyAsync(x => x.QueueName, cancellationToken))!
|
||||
})
|
||||
};
|
||||
}
|
||||
|
|
|
|||
|
|
@ -3,7 +3,6 @@ using System.Threading.Tasks;
|
|||
using Elsa.Activities.AzureServiceBus.Bookmarks;
|
||||
using Elsa.Activities.AzureServiceBus.Options;
|
||||
using Elsa.Bookmarks;
|
||||
using Elsa.Dispatch;
|
||||
using Elsa.Services;
|
||||
using Microsoft.Azure.ServiceBus;
|
||||
using Microsoft.Azure.ServiceBus.Core;
|
||||
|
|
@ -25,7 +24,7 @@ namespace Elsa.Activities.AzureServiceBus.Services
|
|||
|
||||
protected override string ActivityType => nameof(AzureServiceBusQueueMessageReceived);
|
||||
|
||||
protected override IBookmark CreateBookmark(Message message) => new QueueMessageReceivedBookmark(ReceiverClient.Path, message.CorrelationId);
|
||||
protected override IBookmark CreateBookmark(Message message) => new QueueMessageReceivedBookmark(ReceiverClient.Path);
|
||||
protected override IBookmark CreateTrigger(Message message) => new QueueMessageReceivedBookmark(ReceiverClient.Path);
|
||||
}
|
||||
}
|
||||
|
|
@ -30,7 +30,7 @@ namespace Elsa.Activities.Conductor
|
|||
DefaultSyntax = SyntaxNames.Json,
|
||||
SupportedSyntaxes = new[] { SyntaxNames.Json }
|
||||
)]
|
||||
public ISet<string> Outcomes { get; set; } = new HashSet<string>();
|
||||
public ISet<string> OutcomeNames { get; set; } = new HashSet<string>();
|
||||
|
||||
[ActivityOutput(Hint = "Any input that was sent along with the event from your application.")]
|
||||
public object? Payload { get; set; }
|
||||
|
|
@ -44,7 +44,7 @@ namespace Elsa.Activities.Conductor
|
|||
var outcomes = eventModel.Outcomes;
|
||||
|
||||
if (outcomes?.Any() == false)
|
||||
outcomes = new[] { OutcomeNames.Done };
|
||||
outcomes = new[] { Elsa.OutcomeNames.Done };
|
||||
|
||||
Payload = eventModel.Payload;
|
||||
return base.Outcomes(outcomes!);
|
||||
|
|
|
|||
|
|
@ -1,4 +1,3 @@
|
|||
using System;
|
||||
using System.Collections.Generic;
|
||||
using Elsa.Comparers;
|
||||
using Newtonsoft.Json.Linq;
|
||||
|
|
@ -21,7 +20,7 @@ namespace Elsa.Models
|
|||
public string? TenantId { get; set; }
|
||||
public int Version { get; set; }
|
||||
public WorkflowStatus WorkflowStatus { get; set; }
|
||||
public string? CorrelationId { get; set; }
|
||||
public string CorrelationId { get; set; } = default!;
|
||||
public string? ContextType { get; set; }
|
||||
public string? ContextId { get; set; }
|
||||
public string? Name { get; set; }
|
||||
|
|
|
|||
|
|
@ -10,8 +10,6 @@ namespace Elsa.Services
|
|||
Task ResumeWorkflowsAsync(
|
||||
IEnumerable<BookmarkFinderResult> results,
|
||||
object? input = default,
|
||||
string? correlationId = default,
|
||||
string? contextId = default,
|
||||
CancellationToken cancellationToken = default);
|
||||
}
|
||||
}
|
||||
|
|
@ -90,8 +90,8 @@ namespace Elsa.Activities.Workflows
|
|||
return Outcome("Not Found");
|
||||
|
||||
var result = await _startsWorkflow.StartWorkflowAsync(workflowBlueprint!, TenantId, Input, CorrelationId, ContextId, cancellationToken);
|
||||
var workflowInstance = result.WorkflowInstance;
|
||||
var workflowStatus = result.WorkflowInstance.WorkflowStatus;
|
||||
var workflowInstance = result.WorkflowInstance!;
|
||||
var workflowStatus = result.WorkflowInstance!.WorkflowStatus;
|
||||
|
||||
ChildWorkflowInstanceId = workflowInstance.Id;
|
||||
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
using System;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using Elsa.Models;
|
||||
|
|
@ -30,7 +31,7 @@ namespace Elsa.Services
|
|||
TenantId = workflowBlueprint.TenantId,
|
||||
Version = workflowBlueprint.Version,
|
||||
WorkflowStatus = WorkflowStatus.Idle,
|
||||
CorrelationId = correlationId,
|
||||
CorrelationId = correlationId ?? Guid.NewGuid().ToString("N"),
|
||||
ContextId = contextId,
|
||||
CreatedAt = _clock.GetCurrentInstant(),
|
||||
Variables = new Variables(workflowBlueprint.Variables),
|
||||
|
|
|
|||
|
|
@ -274,9 +274,8 @@ namespace Elsa.Services
|
|||
private async Task<IEnumerable<PendingWorkflow>> CollectResumableOrStartableCorrelatedWorkflowsAsync(CollectWorkflowsContext context, CancellationToken cancellationToken)
|
||||
{
|
||||
var correlationId = context.CorrelationId!;
|
||||
var lockKey = correlationId;
|
||||
|
||||
await using var handle = await AcquireLockAsync(lockKey, cancellationToken);
|
||||
await using var handle = await AcquireLockAsync(correlationId, cancellationToken);
|
||||
var correlatedWorkflowInstanceCount = !string.IsNullOrWhiteSpace(correlationId)
|
||||
? await _workflowInstanceStore.CountAsync(new CorrelationIdSpecification<WorkflowInstance>(correlationId).And(new WorkflowUnfinishedStatusSpecification()), cancellationToken)
|
||||
: 0;
|
||||
|
|
|
|||
|
|
@ -39,10 +39,10 @@ namespace Elsa.Services
|
|||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var results = await _bookmarkFinder.FindBookmarksAsync(activityType, bookmark, correlationId, tenantId, cancellationToken).ToList();
|
||||
await ResumeWorkflowsAsync(results, input, correlationId, contextId, cancellationToken);
|
||||
await ResumeWorkflowsAsync(results, input, cancellationToken);
|
||||
}
|
||||
|
||||
public async Task ResumeWorkflowsAsync(IEnumerable<BookmarkFinderResult> results, object? input = default, string? correlationId = default, string? contextId = default, CancellationToken cancellationToken = default)
|
||||
public async Task ResumeWorkflowsAsync(IEnumerable<BookmarkFinderResult> results, object? input = default, CancellationToken cancellationToken = default)
|
||||
{
|
||||
foreach (var result in results)
|
||||
{
|
||||
|
|
|
|||
|
|
@ -6,9 +6,11 @@ using Elsa.Models;
|
|||
using Elsa.Persistence;
|
||||
using Elsa.Persistence.Specifications.WorkflowInstances;
|
||||
using Elsa.Services;
|
||||
using Medallion.Threading;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using NodaTime;
|
||||
using Open.Linq.AsyncExtensions;
|
||||
using IDistributedLockProvider = Elsa.Services.IDistributedLockProvider;
|
||||
|
||||
namespace Elsa.StartupTasks
|
||||
{
|
||||
|
|
@ -21,17 +23,20 @@ namespace Elsa.StartupTasks
|
|||
private readonly IWorkflowInstanceStore _workflowInstanceStore;
|
||||
private readonly IWorkflowInstanceDispatcher _workflowInstanceDispatcher;
|
||||
private readonly IDistributedLockProvider _distributedLockProvider;
|
||||
private readonly ElsaOptions _elsaOptions;
|
||||
private readonly ILogger _logger;
|
||||
|
||||
public ContinueRunningWorkflows(
|
||||
IWorkflowInstanceStore workflowInstanceStore,
|
||||
IWorkflowInstanceDispatcher workflowInstanceDispatcher,
|
||||
IDistributedLockProvider distributedLockProvider,
|
||||
ElsaOptions elsaOptions,
|
||||
ILogger<ContinueRunningWorkflows> logger)
|
||||
{
|
||||
_workflowInstanceStore = workflowInstanceStore;
|
||||
_workflowInstanceDispatcher = workflowInstanceDispatcher;
|
||||
_distributedLockProvider = distributedLockProvider;
|
||||
_elsaOptions = elsaOptions;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
|
|
@ -39,9 +44,7 @@ namespace Elsa.StartupTasks
|
|||
|
||||
public async Task ExecuteAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
var lockKey = GetType().Name;
|
||||
|
||||
await using var handle = await _distributedLockProvider.AcquireLockAsync(lockKey, Duration.FromSeconds(10), cancellationToken);
|
||||
await using var handle = await _distributedLockProvider.AcquireLockAsync(GetType().Name, _elsaOptions.DistributedLockTimeout, cancellationToken);
|
||||
|
||||
if (handle == null)
|
||||
return;
|
||||
|
|
@ -55,6 +58,14 @@ namespace Elsa.StartupTasks
|
|||
|
||||
foreach (var instance in instances)
|
||||
{
|
||||
await using var correlationLockHandle = await _distributedLockProvider.AcquireLockAsync(instance.CorrelationId, _elsaOptions.DistributedLockTimeout, cancellationToken);
|
||||
|
||||
if(handle == null)
|
||||
{
|
||||
_logger.LogWarning("Failed to acquire lock on correlation {CorrelationId} for workflow instance {WorkflowInstanceId}", instance.CorrelationId, instance.Id);
|
||||
continue;
|
||||
}
|
||||
|
||||
_logger.LogInformation("Resuming {WorkflowInstanceId}", instance.Id);
|
||||
var scheduledActivities = instance.ScheduledActivities;
|
||||
|
||||
|
|
@ -65,6 +76,7 @@ namespace Elsa.StartupTasks
|
|||
_logger.LogWarning(
|
||||
"Workflow '{WorkflowInstanceId}' was in the Running state, but has no scheduled activities not has a currently executing one. However, it does have blocking activities, so switching to Suspended status",
|
||||
instance.Id);
|
||||
|
||||
instance.WorkflowStatus = WorkflowStatus.Suspended;
|
||||
await _workflowInstanceStore.SaveAsync(instance, cancellationToken);
|
||||
continue;
|
||||
|
|
|
|||
Loading…
Reference in a new issue