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

Merge remote-tracking branch 'origin/develop/3.5.0'

Browse source
This commit is contained in:
Sipke Schoorstra 2025-05-21 09:55:35 +02:00
parent 787b3ed8eb ed6a20fd25
commit 6d414a79db
No known key found for this signature in database
GPG key ID: 5C10502B28A4268F
19 changed files with 137 additions and 92 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.github/workflows/packages.yml vendored
View file

@ -11,6 +11,7 @@ on:
- 'enh/*'
- 'rc/*'
- 'develop/*'
- 'codex/*'
release:
types: [ prereleased, published ]
env:

3
Directory.Build.props
View file

@ -36,4 +36,7 @@
<!-- IL trimming warnings -->
<NoWarn>$(NoWarn);IL2026;IL2046;IL2057;IL2067;IL2070;IL2072;IL2075;IL2087;IL2091</NoWarn>
</PropertyGroup>
<PropertyGroup>
<ElsaStudioVersion>3.5.0-preview.1019</ElsaStudioVersion>
</PropertyGroup>
</Project>

8
Directory.Packages.props
View file

@ -30,11 +30,11 @@
<PackageVersion Include="DistributedLock.FileSystem" Version="1.0.3"/>
<PackageVersion Include="DistributedLock.Postgres" Version="1.3.0"/>
<PackageVersion Include="DistributedLock.Redis" Version="1.0.3"/>
<PackageVersion Include="Elsa.Studio" Version="3.6.0-preview.979"/>
<PackageVersion Include="Elsa.Studio.Agents" Version="3.6.0-preview.979"/>
<PackageVersion Include="Elsa.Studio.Core.BlazorWasm" Version="3.6.0-preview.979"/>
<PackageVersion Include="Elsa.Studio.Login.BlazorWasm" Version="3.6.0-preview.979"/>
<PackageVersion Include="Elastic.Clients.Elasticsearch" Version="9.0.1"/>
<PackageVersion Include="Elsa.Studio" Version="$(ElsaStudioVersion)"/>
<PackageVersion Include="Elsa.Studio.Agents" Version="$(ElsaStudioVersion)"/>
<PackageVersion Include="Elsa.Studio.Core.BlazorWasm" Version="$(ElsaStudioVersion)"/>
<PackageVersion Include="Elsa.Studio.Login.BlazorWasm" Version="$(ElsaStudioVersion)"/>
<PackageVersion Include="FastEndpoints" Version="6.0.0"/>
<PackageVersion Include="FastEndpoints.Security" Version="6.0.0"/>
<PackageVersion Include="FastEndpoints.Swagger" Version="6.0.0"/>

16
src/modules/Elsa.EntityFrameworkCore.Common/ElsaDbContextBase.cs
View file

@ -19,7 +19,7 @@ public abstract class ElsaDbContextBase : DbContext, IElsaDbContextSchema
};
protected IServiceProvider ServiceProvider { get; }
private readonly ElsaDbContextOptions? _elsaDbContextOptions;
private readonly ElsaDbContextOptions? elsaDbContextOptions;
public string? TenantId { get; set; }
/// <summary>
@ -41,10 +41,10 @@ public abstract class ElsaDbContextBase : DbContext, IElsaDbContextSchema
protected ElsaDbContextBase(DbContextOptions options, IServiceProvider serviceProvider) : base(options)
{
ServiceProvider = serviceProvider;
_elsaDbContextOptions = options.FindExtension<ElsaDbContextOptionsExtension>()?.Options;
elsaDbContextOptions = options.FindExtension<ElsaDbContextOptionsExtension>()?.Options;
// ReSharper disable once VirtualMemberCallInConstructor
Schema = !string.IsNullOrWhiteSpace(_elsaDbContextOptions?.SchemaName) ? _elsaDbContextOptions.SchemaName : ElsaSchema;
Schema = !string.IsNullOrWhiteSpace(elsaDbContextOptions?.SchemaName) ? elsaDbContextOptions.SchemaName : ElsaSchema;
var tenantAccessor = serviceProvider.GetService<ITenantAccessor>();
var tenantId = tenantAccessor?.Tenant?.Id;
@ -63,11 +63,11 @@ public abstract class ElsaDbContextBase : DbContext, IElsaDbContextSchema
/// <inheritdoc />
protected override void OnModelCreating(ModelBuilder modelBuilder)
{
if (!string.IsNullOrWhiteSpace(Schema))
if (!string.IsNullOrWhiteSpace(Schema))
modelBuilder.HasDefaultSchema(Schema);
var additionalConfigurations = _elsaDbContextOptions?.GetModelConfigurations(this);
var additionalConfigurations = elsaDbContextOptions?.GetModelConfigurations(this);
additionalConfigurations?.Invoke(modelBuilder);
using var scope = ServiceProvider.CreateScope();
@ -75,7 +75,7 @@ public abstract class ElsaDbContextBase : DbContext, IElsaDbContextSchema
foreach (var entityType in modelBuilder.Model.GetEntityTypes().ToList())
{
foreach (var handler in entityTypeHandlers)
foreach (var handler in entityTypeHandlers)
handler.Handle(this, modelBuilder, entityType);
}
}

33
src/modules/Elsa.EntityFrameworkCore.Common/EntityHandlers/SetTenantIdFilter.cs
View file

@ -1,9 +1,7 @@
using System.Linq.Expressions;
using Elsa.Common.Entities;
using Elsa.Extensions;
using Microsoft.EntityFrameworkCore;
using Microsoft.EntityFrameworkCore.Metadata;
using Microsoft.EntityFrameworkCore.Query;
namespace Elsa.EntityFrameworkCore.EntityHandlers;
@ -15,15 +13,32 @@ public class SetTenantIdFilter : IEntityModelCreatingHandler
/// <inheritdoc />
public void Handle(ElsaDbContextBase dbContext, ModelBuilder modelBuilder, IMutableEntityType entityType)
{
if (!entityType.ClrType.IsAssignableTo(typeof(Entity)))
if (!typeof(Entity).IsAssignableFrom(entityType.ClrType))
return;
var tenantId = dbContext.TenantId.NullIfEmpty();
var parameter = Expression.Parameter(entityType.ClrType);
Expression<Func<Entity, bool>> filterExpr = entity => entity.TenantId == tenantId;
var body = ReplacingExpressionVisitor.Replace(filterExpr.Parameters[0], parameter, filterExpr.Body);
var lambdaExpression = Expression.Lambda(body, parameter);
modelBuilder
.Entity(entityType.ClrType)
.HasQueryFilter(CreateTenantFilterExpression(dbContext, entityType.ClrType));
}
entityType.SetQueryFilter(lambdaExpression);
private LambdaExpression CreateTenantFilterExpression(ElsaDbContextBase dbContext, Type clrType)
{
var parameter = Expression.Parameter(clrType, "e");
// e => EF.Property<string>(e, "TenantId") == this.TenantId
var tenantIdProperty = Expression.Call(
typeof(EF),
nameof(EF.Property),
[typeof(string)],
parameter,
Expression.Constant("TenantId"));
var tenantIdOnContext = Expression.Property(
Expression.Constant(dbContext),
nameof(ElsaDbContextBase.TenantId));
var body = Expression.Equal(tenantIdProperty, tenantIdOnContext);
return Expression.Lambda(body, parameter);
}
}

4
src/modules/Elsa.Http/Activities/HttpEndpoint.cs
View file

@ -163,7 +163,7 @@ public class HttpEndpoint : Trigger<HttpRequest>
{
var path = Path.Get(context);
var methods = SupportedMethods.GetOrDefault(context) ?? new List<string> { HttpMethods.Get };
context.WaitForHttpRequest(path, methods, OnResumeAsync);
await context.WaitForHttpRequestAsync(path, methods, OnResumeAsync);
}
private async ValueTask OnResumeAsync(ActivityExecutionContext context)
@ -497,4 +497,4 @@ public class HttpEndpoint : Trigger<HttpRequest>
return routeData;
}
}
}

4
src/modules/Elsa.Http/Activities/HttpEndpointBase.cs
View file

@ -20,10 +20,10 @@ public abstract class HttpEndpointBase<TResult> : Trigger<TResult>
{
}
protected override void Execute(ActivityExecutionContext context)
protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
{
var options = GetOptions();
context.WaitForHttpRequest(options, HttpRequestReceivedAsync);
await context.WaitForHttpRequestAsync(options, HttpRequestReceivedAsync);
}
protected override IEnumerable<object> GetTriggerPayloads(TriggerIndexingContext context)

39
src/modules/Elsa.Http/Extensions/HttpEndpointActivityExecutionContextExtensions.cs
View file

@ -9,27 +9,27 @@ namespace Elsa.Http.Extensions;
public static class HttpEndpointActivityExecutionContextExtensions
{
public static void WaitForHttpRequest(this ActivityExecutionContext context, string path, string method, ExecuteActivityDelegate? callback = null)
public static async ValueTask WaitForHttpRequestAsync(this ActivityExecutionContext context, string path, string method, ExecuteActivityDelegate? callback = null)
{
var options = new HttpEndpointOptions
{
var options = new HttpEndpointOptions
{
Path = path,
Methods = [method]
};
WaitForHttpRequest(context, options, callback);
}
Path = path,
Methods = [method]
};
await WaitForHttpRequestAsync(context, options, callback);
}
public static void WaitForHttpRequest(this ActivityExecutionContext context, string path, IEnumerable<string> methods, ExecuteActivityDelegate? callback = null)
public static async ValueTask WaitForHttpRequestAsync(this ActivityExecutionContext context, string path, IEnumerable<string> methods, ExecuteActivityDelegate? callback = null)
{
var options = new HttpEndpointOptions
{
var options = new HttpEndpointOptions
{
Path = path,
Methods = methods.ToList()
};
WaitForHttpRequest(context, options, callback);
}
Path = path,
Methods = methods.ToList()
};
await WaitForHttpRequestAsync(context, options, callback);
}
public static void WaitForHttpRequest(this ActivityExecutionContext context, HttpEndpointOptions options, ExecuteActivityDelegate? callback = null)
public static async ValueTask WaitForHttpRequestAsync(this ActivityExecutionContext context, HttpEndpointOptions options, ExecuteActivityDelegate? callback = null)
{
var path = options.Path;
if (path.Contains("//"))
@ -42,7 +42,8 @@ public static class HttpEndpointActivityExecutionContextExtensions
return;
}
callback?.Invoke(context);
if (callback is not null)
await callback(context);
}
public static IEnumerable<object> GetHttpEndpointStimuli(this TriggerIndexingContext context, string path, string method)
@ -91,4 +92,4 @@ public static class HttpEndpointActivityExecutionContextExtensions
};
context.CreateBookmark(bookmarkOptions);
}
}
}

10
src/modules/Elsa.Workflows.Runtime/Activities/BulkDispatchWorkflows.cs
View file

@ -78,12 +78,12 @@ public class BulkDispatchWorkflows : Activity
Description = "Wait for the dispatched workflows to complete before completing this activity.",
DefaultValue = true)]
public Input<bool> WaitForCompletion { get; set; } = new(true);
/// <summary>
/// Indicates whether a new trace context should be started for the workflow execution.
/// </summary>
[Input(Description = "Start a new trace context when using Open Telemetry.", Category = "Open Telemetry")]
public Input<bool> StartNewTrace { get; set; }
public Input<bool> StartNewTrace { get; set; } = new(false);
/// <summary>
/// The channel to dispatch the workflow to.
@ -238,17 +238,17 @@ public class BulkDispatchWorkflows : Activity
await context.ScheduleActivityAsync(ChildCompleted, options);
return;
default:
await CheckIfCompletedAsync(context);
await AttemptToCompleteAsync(context);
break;
}
}
private async ValueTask OnChildFinishedCompletedAsync(ActivityCompletedContext context)
{
await CheckIfCompletedAsync(context.TargetContext);
await AttemptToCompleteAsync(context.TargetContext);
}
private async ValueTask CheckIfCompletedAsync(ActivityExecutionContext context)
private async ValueTask AttemptToCompleteAsync(ActivityExecutionContext context)
{
var dispatchedInstancesCount = context.GetProperty<long>(DispatchedInstancesCountKey);
var finishedInstancesCount = context.GetProperty<long>(CompletedInstancesCountKey);

6
src/modules/Elsa.Workflows.Runtime/Exceptions/WorkflowInstanceNotFoundException.cs Normal file
View file

@ -0,0 +1,6 @@
namespace Elsa.Workflows.Runtime.Exceptions;
public class WorkflowInstanceNotFoundException(string message, string instanceId) : Exception(message)
{
public string InstanceId { get; } = instanceId;
}

8
src/modules/Elsa.Workflows.Runtime/Handlers/SignalBookmarkQueueWorker.cs
View file

@ -1,4 +1,5 @@
using Elsa.Mediator.Contracts;
using Elsa.Workflows.Management.Notifications;
using Elsa.Workflows.Runtime.Notifications;
using JetBrains.Annotations;
@ -8,7 +9,7 @@ namespace Elsa.Workflows.Runtime.Handlers;
/// Signals the bookmark queue worker to process any queued work.
/// </summary>
[UsedImplicitly]
public class SignalBookmarkQueueWorker(IBookmarkQueueSignaler signaler) : INotificationHandler<WorkflowBookmarksIndexed>, INotificationHandler<BookmarkSaved>
public class SignalBookmarkQueueWorker(IBookmarkQueueSignaler signaler) : INotificationHandler<WorkflowBookmarksIndexed>, INotificationHandler<BookmarkSaved>, INotificationHandler<WorkflowInstanceSaved>
{
public Task HandleAsync(BookmarkSaved notification, CancellationToken cancellationToken)
{
@ -20,6 +21,11 @@ public class SignalBookmarkQueueWorker(IBookmarkQueueSignaler signaler) : INotif
return Trigger();
}
public Task HandleAsync(WorkflowInstanceSaved notification, CancellationToken cancellationToken)
{
return Trigger();
}
private async Task Trigger()
{
await signaler.TriggerAsync();

2
src/modules/Elsa.Workflows.Runtime/Messages/RunWorkflowInstanceResponse.cs
View file

@ -10,7 +10,7 @@ public record RunWorkflowInstanceResponse
/// <summary>
/// The ID of the workflow instance.
/// </summary>
public string WorkflowInstanceId { get; set; } = default!;
public string WorkflowInstanceId { get; set; } = null!;
/// <summary>
/// The status of the workflow instance.

43
src/modules/Elsa.Workflows.Runtime/Services/BookmarkQueueSignaler.cs
View file

@ -1,43 +1,30 @@
using System.Threading.Channels;
namespace Elsa.Workflows.Runtime;
public class BookmarkQueueSignaler : IBookmarkQueueSignaler
{
private readonly object _lock = new();
private TaskCompletionSource<object?> _tcs = new();
private readonly Channel<object?> _channel;
public async Task AwaitAsync(CancellationToken cancellationToken)
public BookmarkQueueSignaler()
{
Task waitTask;
lock (_lock)
var options = new BoundedChannelOptions(1)
{
// Capture the current TCS and await it
waitTask = _tcs.Task;
}
SingleReader = true,
SingleWriter = false,
AllowSynchronousContinuations = false
};
_channel = Channel.CreateBounded<object?>(options);
}
await WaitAndResetAsync(waitTask);
public Task AwaitAsync(CancellationToken cancellationToken)
{
return _channel.Reader.ReadAsync(cancellationToken).AsTask();
}
public Task TriggerAsync(CancellationToken cancellationToken)
{
lock (_lock)
{
// If TCS is already in a completed state, no need to set it again.
if (!_tcs.Task.IsCompleted)
{
_tcs.SetResult(null);
}
}
_channel.Writer.TryWrite(null);
return Task.CompletedTask;
}
private async Task WaitAndResetAsync(Task waitTask)
{
await waitTask;
lock (_lock)
{
// Reset the TCS for the next wait
_tcs = new();
}
}
}

19
src/modules/Elsa.Workflows.Runtime/Services/BookmarkQueueWorker.cs
View file

@ -7,7 +7,7 @@ namespace Elsa.Workflows.Runtime;
public class BookmarkQueueWorker : IBookmarkQueueWorker
{
private readonly RateLimitedFunc<CancellationToken, Task> _rateLimitedProcessAsync;
private CancellationTokenSource _cts = default!;
private CancellationTokenSource _cts = null!;
private bool _running;
private readonly IBookmarkQueueSignaler _signaler;
private readonly IServiceScopeFactory _scopeFactory;
@ -18,7 +18,7 @@ public class BookmarkQueueWorker : IBookmarkQueueWorker
_signaler = signaler;
_scopeFactory = scopeFactory;
_logger = logger;
_rateLimitedProcessAsync = Debouncer.Debounce<CancellationToken, Task>(ProcessAsync, TimeSpan.FromMilliseconds(500));
_rateLimitedProcessAsync = Throttler.Throttle<CancellationToken, Task>(ProcessAsync, TimeSpan.FromMilliseconds(500));
}
public void Start()
@ -47,8 +47,19 @@ public class BookmarkQueueWorker : IBookmarkQueueWorker
{
while (!_cts.IsCancellationRequested)
{
await _signaler.AwaitAsync(_cts.Token);
await _rateLimitedProcessAsync.InvokeAsync(_cts.Token);
try
{
await _signaler.AwaitAsync(_cts.Token);
await _rateLimitedProcessAsync.InvokeAsync(_cts.Token);
}
catch (OperationCanceledException)
{
break; // Stop() was called
}
catch (Exception ex)
{
_logger.LogError(ex, "BookmarkQueueWorker error – continuing loop");
}
}
}

19
src/modules/Elsa.Workflows.Runtime/Services/BookmarkResumer.cs
View file

@ -1,4 +1,5 @@
using Elsa.Workflows.Helpers;
using Elsa.Workflows.Runtime.Exceptions;
using Elsa.Workflows.Runtime.Filters;
using Elsa.Workflows.Runtime.Messages;
using Elsa.Workflows.Runtime.Options;
@ -64,7 +65,7 @@ public class BookmarkResumer(IWorkflowRuntime workflowRuntime, IBookmarkStore bo
ActivityHandle = request.ActivityHandle,
BookmarkId = request.BookmarkId
};
var workflowInstanceId = request.WorkflowInstanceId;
var workflowClient = await workflowRuntime.CreateClientAsync(workflowInstanceId, cancellationToken);
var response = await workflowClient.RunInstanceAsync(runRequest, cancellationToken);
@ -89,8 +90,18 @@ public class BookmarkResumer(IWorkflowRuntime workflowRuntime, IBookmarkStore bo
Properties = options?.Properties,
BookmarkId = bookmark.Id
};
var response = await workflowClient.RunInstanceAsync(runRequest, cancellationToken);
logger.LogDebug("Resumed workflow instance {WorkflowInstanceId} with bookmark {BookmarkId}", bookmark.WorkflowInstanceId, bookmark.Id);
return ResumeBookmarkResult.Found(response);
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();
}
}
}

5
src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowClient.cs
View file

@ -4,6 +4,7 @@ using Elsa.Workflows.Management.Mappers;
using Elsa.Workflows.Management.Options;
using Elsa.Workflows.Models;
using Elsa.Workflows.Options;
using Elsa.Workflows.Runtime.Exceptions;
using Elsa.Workflows.Runtime.Messages;
using Elsa.Workflows.State;
using Microsoft.Extensions.Logging;
@ -166,7 +167,7 @@ public class LocalWorkflowClient(
private async Task<WorkflowInstance> GetWorkflowInstanceAsync(CancellationToken cancellationToken)
{
var workflowInstance = await workflowInstanceManager.FindByIdAsync(WorkflowInstanceId, cancellationToken);
if (workflowInstance == null) throw new InvalidOperationException($"Workflow instance {WorkflowInstanceId} not found.");
if (workflowInstance == null) throw new WorkflowInstanceNotFoundException($"Workflow instance not found.", WorkflowInstanceId);
return workflowInstance;
}
@ -179,7 +180,7 @@ public class LocalWorkflowClient(
private async Task<WorkflowGraph> GetWorkflowGraphAsync(WorkflowDefinitionHandle definitionHandle, CancellationToken cancellationToken)
{
var workflowGraph = await workflowDefinitionService.FindWorkflowGraphAsync(definitionHandle, cancellationToken);
if (workflowGraph == null) throw new InvalidOperationException($"Workflow graph with handle {definitionHandle} not found.");
if (workflowGraph == null) throw new WorkflowGraphNotFoundException($"Workflow graph not found.", definitionHandle);
return workflowGraph;
}
}

2
src/modules/Elsa.Workflows.Runtime/Services/StoreBookmarkQueue.cs
View file

@ -32,7 +32,7 @@ public class StoreBookmarkQueue(
return;
}
// There was no matching bookmark yet. Store the queue item for the system to pick up whenever the bookmark becomes present.
// There was no matching bookmark yet, or the associated workflow instance hasn't been stored in the DB yet. Store the queue item for the system to pick up whenever the bookmark or workflow instance becomes present.
logger.LogDebug("No bookmark with ID {BookmarkId} found for workflow {WorkflowInstance} for activity type {ActivityType}. Adding the request to the bookmark queue", item.BookmarkId, item.WorkflowInstanceId, item.ActivityTypeName);
var entity = new BookmarkQueueItem

5
src/modules/Elsa/Features/ElsaFeature.cs
View file

@ -40,7 +40,10 @@ public class ElsaFeature : FeatureBase
.UseWorkflowManagement(management =>
{
if (!DisableAutomaticActivityRegistration)
management.AddActivitiesFrom<WriteLine>();
management
.AddActivitiesFrom<WriteLine>()
.RemoveActivity<ReadLine>() // ReadLine is not commonly used and can cause "hanging" containers when awaiting user input. Better to opt-in explicitly.
;
});
}
}

2
test/component/Elsa.Workflows.ComponentTests/Scenarios/LogPersistenceModes/InputOutputLoggingTests.cs
View file

@ -49,7 +49,7 @@ public class InputOutputLoggingTests(App app) : AppComponentTest(app)
Assert.True(output2IsIncluded);
}
[Fact(Skip = "Although the scenario works reliably, for some reason the test fails, most of the time, when run from the CLI and not using the IDE (Rider).")]
[Fact(Skip = "Although this functionality works in practice, the component test fails from time to time for no clear reason (yet).")]
public async Task WorkflowAsActivityInternal_ShouldHonorSettings_WhenExecuting()
{
await ExecuteWorkflowAsync("input-output-logging-3");