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

BREAKING DB: Add DefinitonVersionId ans optimize workflow registry perf

Browse source
This commit is contained in:
Sipke Schoorstra 2021-12-15 11:58:28 +01:00
parent 941b7dd6c9
commit b6eb9a91e4
11 changed files with 108 additions and 50 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
src/core/Elsa.Abstractions/Models/WorkflowInstance.cs
View file

@ -18,8 +18,10 @@ namespace Elsa.Models
}
public string DefinitionId { get; set; } = default!;
public string DefinitionVersionId { get; set; } = default!;
public string? TenantId { get; set; }
public int Version { get; set; }
public WorkflowStatus WorkflowStatus { get; set; }
public string CorrelationId { get; set; } = default!;
public string? ContextType { get; set; }

1
src/core/Elsa.Abstractions/Services/Models/IWorkflowBlueprint.cs
View file

@ -4,6 +4,7 @@ namespace Elsa.Services.Models
{
public interface IWorkflowBlueprint : ICompositeActivityBlueprint
{
string VersionId { get; }
int Version { get; }
string? TenantId { get; }
bool IsSingleton { get; }

23
src/core/Elsa.Abstractions/Services/Models/WorkflowBlueprint.cs
View file

@ -16,6 +16,7 @@ namespace Elsa.Services.Models
public WorkflowBlueprint(
string id,
int version,
string versionId,
string? tenantId,
bool isSingleton,
string? name,
@ -33,21 +34,22 @@ namespace Elsa.Services.Models
IEnumerable<IActivityBlueprint> activities,
IEnumerable<IConnection> connections,
IActivityPropertyProviders activityPropertyValueProviders) : base(
id,
default,
name,
displayName,
description,
id,
true,
false,
false,
id,
default,
name,
displayName,
description,
id,
true,
false,
false,
new Dictionary<string, string>(),
default)
{
Id = id;
Parent = this;
Version = version;
VersionId = versionId;
TenantId = tenantId;
IsSingleton = isSingleton;
IsLatest = isLatest;
@ -66,6 +68,7 @@ namespace Elsa.Services.Models
}
public int Version { get; set; }
public string VersionId { get; set; }
public string? TenantId { get; set; }
public bool IsSingleton { get; set; }
public bool IsPublished { get; set; }
@ -77,7 +80,7 @@ namespace Elsa.Services.Models
/// The channel, or queue, to place workflow instances of this workflow blueprint in. Channels can be used by the workflow dispatcher to prioritize workflows.
/// </summary>
public string? Channel { get; }
public Variables Variables { get; set; }
public WorkflowContextOptions? ContextOptions { get; set; }
public WorkflowPersistenceBehavior PersistenceBehavior { get; set; }

17
src/core/Elsa.Core/Builders/WorkflowBuilder.cs
View file

@ -24,7 +24,7 @@ namespace Elsa.Builders
PropertyValueProviders = new Dictionary<string, IActivityPropertyValueProvider>();
PersistenceBehavior = WorkflowPersistenceBehavior.WorkflowBurst;
}
public int Version { get; private set; }
public bool IsLatest { get; private set; }
public bool IsPublished { get; private set; }
@ -45,25 +45,25 @@ namespace Elsa.Builders
ActivityId = value!;
return this;
}
public IWorkflowBuilder ForTenantId(string? value)
{
TenantId = value;
return this;
}
public new IWorkflowBuilder WithDisplayName(string value)
{
DisplayName = value;
return this;
}
public new IWorkflowBuilder WithDescription(string? value)
{
Description = value;
return this;
}
public IWorkflowBuilder WithChannel(string? value)
{
Channel = value;
@ -93,7 +93,7 @@ namespace Elsa.Builders
IsPublished = isPublished;
return this;
}
public IWorkflowBuilder WithTag(string value)
{
Tag = value;
@ -148,7 +148,7 @@ namespace Elsa.Builders
Name ??= workflowTypeName;
DisplayName ??= workflowTypeName;
WithId(workflowTypeName);
workflow.Build(this);
return BuildBlueprint(activityIdPrefix);
@ -165,10 +165,11 @@ namespace Elsa.Builders
public IWorkflowBlueprint BuildBlueprint(string activityIdPrefix)
{
var compositeRoot = Build(activityIdPrefix);
return new WorkflowBlueprint(
ActivityId,
Version,
$"{ActivityId}v{Version}",
TenantId,
IsSingleton,
Name,

39
src/core/Elsa.Core/Decorators/CachingWorkflowRegistry.cs
View file

@ -26,9 +26,9 @@ namespace Elsa.Decorators
private readonly IWorkflowInstanceStore _workflowInstanceStore;
public CachingWorkflowRegistry(
IWorkflowRegistry workflowRegistry,
IMemoryCache memoryCache,
ICacheSignal cacheSignal,
IWorkflowRegistry workflowRegistry,
IMemoryCache memoryCache,
ICacheSignal cacheSignal,
IWorkflowInstanceStore workflowInstanceStore)
{
_workflowRegistry = workflowRegistry;
@ -63,25 +63,28 @@ namespace Elsa.Decorators
return await _workflowRegistry.ListAsync(cancellationToken).ToList();
});
}
private async IAsyncEnumerable<IWorkflowBlueprint> ListActiveInternalAsync([EnumeratorCancellation] CancellationToken cancellationToken)
{
var workflows = await ListInternalAsync(cancellationToken);
foreach (var workflow in workflows)
{
// If a workflow is not published, only consider it for processing if it has at least one non-ended workflow instance.
if (!workflow.IsPublished && !await WorkflowHasUnfinishedWorkflowsAsync(workflow, cancellationToken))
continue;
var publishedWorkflows = workflows.Where(x => x.IsPublished);
yield return workflow;
}
}
private async Task<bool> WorkflowHasUnfinishedWorkflowsAsync(IWorkflowBlueprint workflowBlueprint, CancellationToken cancellationToken)
{
var count = await _workflowInstanceStore.CountAsync(new UnfinishedWorkflowSpecification().WithWorkflowDefinition(workflowBlueprint.Id), cancellationToken);
return count > 0;
foreach (var publishedWorkflow in publishedWorkflows)
yield return publishedWorkflow;
// We also need to consider unpublished workflows for inclusion in case they still have associated active workflow instances.
var unpublishedWorkflows = workflows.Where(x => !x.IsPublished).ToDictionary(x => x.VersionId);
var unpublishedWorkflowIds = unpublishedWorkflows.Keys;
if (!unpublishedWorkflowIds.Any())
yield break;
var activeWorkflowInstances = await _workflowInstanceStore.FindManyAsync(new UnfinishedWorkflowSpecification().WithWorkflowDefinitionVersionIds(unpublishedWorkflowIds), cancellationToken: cancellationToken).ToList();
var activeUnpublishedWorkflowVersionIds = activeWorkflowInstances.Select(x => x.DefinitionVersionId).Distinct().ToList();
var activeUnpublishedWorkflowVersions = unpublishedWorkflows.Where(x => activeUnpublishedWorkflowVersionIds.Contains(x.Key)).Select(x => x.Value);
foreach (var unpublishedWorkflow in activeUnpublishedWorkflowVersions)
yield return unpublishedWorkflow;
}
Task INotificationHandler<WorkflowDefinitionSaved>.Handle(WorkflowDefinitionSaved notification, CancellationToken cancellationToken)

14
src/core/Elsa.Core/Persistence/Specifications/WorkflowInstances/WorkflowDefinitionVersionIdsSpecification.cs Normal file
View file

@ -0,0 +1,14 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Linq.Expressions;
using Elsa.Models;
namespace Elsa.Persistence.Specifications.WorkflowInstances;
public class WorkflowDefinitionVersionIdsSpecification : Specification<WorkflowInstance>
{
public ICollection<string> WorkflowDefinitionVersionIds { get; set; }
public WorkflowDefinitionVersionIdsSpecification(IEnumerable<string> workflowDefinitionVersionIds) => WorkflowDefinitionVersionIds = workflowDefinitionVersionIds.ToList();
public override Expression<Func<WorkflowInstance, bool>> ToExpression() => x => WorkflowDefinitionVersionIds.Contains(x.DefinitionVersionId);
}

24
src/core/Elsa.Core/Persistence/Specifications/WorkflowInstances/WorkflowInstanceSpecificationExtensions.cs
View file

@ -1,21 +1,33 @@
using System;
using System.Collections.Generic;
using Elsa.Models;
namespace Elsa.Persistence.Specifications.WorkflowInstances
{
public static class WorkflowInstanceSpecificationExtensions
{
public static ISpecification<WorkflowInstance> WithWorkflowDefinition(this ISpecification<WorkflowInstance> specification, string workflowDefinitionId) => specification.And(new WorkflowDefinitionIdSpecification(workflowDefinitionId));
public static ISpecification<WorkflowInstance> WithWorkflowDefinition(this ISpecification<WorkflowInstance> specification, string workflowDefinitionId) =>
specification.And(new WorkflowDefinitionIdSpecification(workflowDefinitionId));
public static ISpecification<WorkflowInstance> WithWorkflowDefinitionVersionIds(this ISpecification<WorkflowInstance> specification, IEnumerable<string> workflowDefinitionVersionIds) =>
specification.And(new WorkflowDefinitionVersionIdsSpecification(workflowDefinitionVersionIds));
public static ISpecification<WorkflowInstance> WithWorkflowName(this ISpecification<WorkflowInstance> specification, string name) => specification.And(new WorkflowInstanceNameMatchSpecification(name));
public static ISpecification<WorkflowInstance> WithCorrelationId(this ISpecification<WorkflowInstance> specification, string correlationId) => specification.And(new CorrelationIdSpecification<WorkflowInstance>(correlationId));
public static ISpecification<WorkflowInstance> WithContextId(this ISpecification<WorkflowInstance> specification, string contextType, string contextId) => specification.And(new WorkflowInstanceContextIdMatchSpecification(contextType, contextId));
public static ISpecification<WorkflowInstance> WithContextId(this ISpecification<WorkflowInstance> specification, Type contextType, string contextId) => specification.And(new WorkflowInstanceContextIdMatchSpecification(contextType, contextId));
public static ISpecification<WorkflowInstance> WithContextId<TContextType>(this ISpecification<WorkflowInstance> specification, string contextId) => specification.And(new WorkflowInstanceContextIdMatchSpecification(typeof(TContextType), contextId));
public static ISpecification<WorkflowInstance> WithContextId(this ISpecification<WorkflowInstance> specification, string contextType, string contextId) =>
specification.And(new WorkflowInstanceContextIdMatchSpecification(contextType, contextId));
public static ISpecification<WorkflowInstance> WithContextId(this ISpecification<WorkflowInstance> specification, Type contextType, string contextId) =>
specification.And(new WorkflowInstanceContextIdMatchSpecification(contextType, contextId));
public static ISpecification<WorkflowInstance> WithContextId<TContextType>(this ISpecification<WorkflowInstance> specification, string contextId) =>
specification.And(new WorkflowInstanceContextIdMatchSpecification(typeof(TContextType), contextId));
public static ISpecification<WorkflowInstance> WithContext(this ISpecification<WorkflowInstance> specification, string contextType) => specification.And(new WorkflowInstanceContextMatchSpecification(contextType));
public static ISpecification<WorkflowInstance> WithContext(this ISpecification<WorkflowInstance> specification, Type contextType) => specification.And(new WorkflowInstanceContextMatchSpecification(contextType));
public static ISpecification<WorkflowInstance> WithContextId<TContextType>(this ISpecification<WorkflowInstance> specification) => specification.And(new WorkflowInstanceContextMatchSpecification(typeof(TContextType)));
public static ISpecification<WorkflowInstance> WithStatus(this ISpecification<WorkflowInstance> specification, WorkflowStatus status) => specification.And(new WorkflowStatusSpecification(status));
public static ISpecification<WorkflowInstance> WithSearchTerm(this ISpecification<WorkflowInstance> specification, string searchTerm) => specification.And(new WorkflowSearchTermSpecification(searchTerm));
}
}
}

1
src/core/Elsa.Core/Services/Workflows/WorkflowBlueprintMaterializer.cs
View file

@ -46,6 +46,7 @@ namespace Elsa.Services.Workflows
var workflowBlueprint = new WorkflowBlueprint(
workflowDefinition.DefinitionId,
workflowDefinition.Version,
workflowDefinition.VersionId,
workflowDefinition.TenantId,
workflowDefinition.IsSingleton,
workflowDefinition.Name,

1
src/core/Elsa.Core/Services/Workflows/WorkflowFactory.cs
View file

@ -31,6 +31,7 @@ namespace Elsa.Services.Workflows
DefinitionId = workflowBlueprint.Id,
TenantId = tenantId ?? workflowBlueprint.TenantId,
Version = workflowBlueprint.Version,
DefinitionVersionId = workflowBlueprint.VersionId,
WorkflowStatus = WorkflowStatus.Idle,
CorrelationId = correlationId ?? Guid.NewGuid().ToString("N"),
ContextId = contextId,

25
src/core/Elsa.Core/Services/Workflows/WorkflowRegistry.cs
View file

@ -46,16 +46,25 @@ namespace Elsa.Services.Workflows
private async IAsyncEnumerable<IWorkflowBlueprint> ListActiveInternalAsync([EnumeratorCancellation] CancellationToken cancellationToken)
{
var workflows = await ListAsync(cancellationToken);
var workflows = await ListInternalAsync(cancellationToken).ToListAsync(cancellationToken);
var publishedWorkflows = workflows.Where(x => x.IsPublished);
foreach (var workflow in workflows)
{
// If a workflow is not published, only consider it for processing if it has at least one non-ended workflow instance.
if (!workflow.IsPublished && !await WorkflowHasUnfinishedWorkflowsAsync(workflow, cancellationToken))
continue;
foreach (var publishedWorkflow in publishedWorkflows)
yield return publishedWorkflow;
yield return workflow;
}
// We also need to consider unpublished workflows for inclusion in case they still have associated active workflow instances.
var unpublishedWorkflows = workflows.Where(x => !x.IsPublished).ToDictionary(x => x.VersionId);
var unpublishedWorkflowIds = unpublishedWorkflows.Keys;
if (!unpublishedWorkflowIds.Any())
yield break;
var activeWorkflowInstances = await _workflowInstanceStore.FindManyAsync(new UnfinishedWorkflowSpecification().WithWorkflowDefinitionVersionIds(unpublishedWorkflowIds), cancellationToken: cancellationToken).ToList();
var activeUnpublishedWorkflowVersionIds = activeWorkflowInstances.Select(x => x.DefinitionVersionId).Distinct().ToList();
var activeUnpublishedWorkflowVersions = unpublishedWorkflows.Where(x => activeUnpublishedWorkflowVersionIds.Contains(x.Key)).Select(x => x.Value);
foreach (var unpublishedWorkflow in activeUnpublishedWorkflowVersions)
yield return unpublishedWorkflow;
}
private async IAsyncEnumerable<IWorkflowBlueprint> ListInternalAsync([EnumeratorCancellation] CancellationToken cancellationToken)

11
src/samples/server/Elsa.Samples.Server.Host/SampleActivity.cs Normal file
View file

@ -0,0 +1,11 @@
using System.Collections.Generic;
using Elsa.Attributes;
using Elsa.Expressions;
using Elsa.Services;
namespace Elsa.Samples.Server.Host;
public class SampleActivity : Activity
{
[ActivityInput(SupportedSyntaxes = new[]{SyntaxNames.JavaScript, SyntaxNames.Liquid})] public ICollection<string> MyList { get; set; }
}