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

Fix background activity completion (#5882)

Browse source 
* Enhance logging and state handling in workflows

Added debug logging in `BookmarkResumer` and `StoreBookmarkQueue` to track workflow resumptions and queue additions. Updated `ICommitStateHandler` to include `WorkflowState` parameter and modified related implementations. Adjusted logging configuration in sample app and removed unused service import.

* Add CustomProperties entry for RootType in AgentActivity

This change ensures that the `RootType` property is set to `AgentActivity` inside the `AgentActivityProvider`. It enhances the descriptor's metadata, enabling clearer classification and potentially improving integration with other components.

* Update RemoveBookmarksAsync to use BookmarkIds instead of Hashes

Modified RemoveBookmarksAsync to filter bookmarks by BookmarkIds rather than Hashes for improved accuracy. Updated filter creation logic to accommodate the new identifier field.

* Fix commit state handler call in WorkflowRunner

Add workflowState as an argument to commitStateHandler.CommitAsync. This ensures that the commitStateHandler has the necessary context to commit properly and maintains consistency in workflow state changes.
This commit is contained in:
Sipke Schoorstra 2024-08-10 10:41:35 +02:00 committed by GitHub
parent 58e4523406
commit 6741db388b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
13 changed files with 41 additions and 15 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
samples/aspnet/Elsa.Samples.AspNet.OrchardCoreIntegration/appsettings.json
View file

@ -2,7 +2,8 @@
"Logging": {
"LogLevel": {
"Default": "Information",
"Microsoft.AspNetCore": "Warning"
"Microsoft.AspNetCore": "Warning",
"Microsoft.EntityFrameworkCore": "Warning"
}
},
"AllowedHosts": "*",

1
src/modules/Elsa.Agents.Activities/ActivityProviders/AgentActivityProvider.cs
View file

@ -46,6 +46,7 @@ public class AgentActivityProvider(AgentManager agentManager, KernelConfig kerne
activityDescriptor.IsBrowsable = true;
activityDescriptor.Category = "Agent Skills";
activityDescriptor.Kind = ActivityKind.Task;
activityDescriptor.CustomProperties["RootType"] = nameof(AgentActivity);
activityDescriptor.Constructor = context =>
{

4
src/modules/Elsa.Workflows.Core/Contracts/ICommitStateHandler.cs
View file

@ -1,6 +1,8 @@
using Elsa.Workflows.State;
namespace Elsa.Workflows;
public interface ICommitStateHandler
{
Task CommitAsync(WorkflowExecutionContext workflowExecutionContext, CancellationToken cancellationToken = default);
Task CommitAsync(WorkflowExecutionContext workflowExecutionContext, WorkflowState workflowState, CancellationToken cancellationToken = default);
}

4
src/modules/Elsa.Workflows.Core/Services/NoopCommitStateHandler.cs
View file

@ -1,8 +1,10 @@
using Elsa.Workflows.State;
namespace Elsa.Workflows;
public class NoopCommitStateHandler : ICommitStateHandler
{
public Task CommitAsync(WorkflowExecutionContext workflowExecutionContext, CancellationToken cancellationToken = default)
public Task CommitAsync(WorkflowExecutionContext workflowExecutionContext, WorkflowState workflowState, CancellationToken cancellationToken = default)
{
return Task.CompletedTask;
}

2
src/modules/Elsa.Workflows.Core/Services/WorkflowRunner.cs
View file

@ -195,7 +195,7 @@ public class WorkflowRunner(
var result = workflow.ResultVariable?.Get(workflowExecutionContext.MemoryRegister);
await notificationSender.SendAsync(new WorkflowExecuted(workflow, workflowState, workflowExecutionContext), cancellationToken);
await commitStateHandler.CommitAsync(workflowExecutionContext, cancellationToken);
await commitStateHandler.CommitAsync(workflowExecutionContext, workflowState, cancellationToken);
return new RunWorkflowResult(workflowState, workflowExecutionContext.Workflow, result);
}
}

1
src/modules/Elsa.Workflows.Runtime.Distributed/Features/DistributedRuntimeFeature.cs
View file

@ -3,7 +3,6 @@ using Elsa.Features.Abstractions;
using Elsa.Features.Attributes;
using Elsa.Features.Services;
using Elsa.Workflows.Runtime.Distributed.Handlers;
using Elsa.Workflows.Runtime.Distributed.Services;
using Elsa.Workflows.Runtime.Features;
using Microsoft.Extensions.DependencyInjection;

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

@ -5,7 +5,7 @@ using Medallion.Threading;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Options;
namespace Elsa.Workflows.Runtime.Distributed.Services;
namespace Elsa.Workflows.Runtime.Distributed;
public class DistributedWorkflowClient(
string workflowInstanceId,

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

@ -1,7 +1,7 @@
using Elsa.Workflows.Contracts;
using Microsoft.Extensions.DependencyInjection;
namespace Elsa.Workflows.Runtime.Distributed.Services;
namespace Elsa.Workflows.Runtime.Distributed;
/// <summary>
/// Represents a distributed workflow runtime that can create <see cref="IWorkflowClient"/> instances connected to a workflow instance.

3
src/modules/Elsa.Workflows.Runtime/Services/BookmarkPersister.cs
View file

@ -1,4 +1,3 @@
using Elsa.Mediator;
using Elsa.Mediator.Contracts;
using Elsa.Workflows.Runtime.Notifications;
using Elsa.Workflows.Runtime.Requests;
@ -17,6 +16,6 @@ public class BookmarksPersister(IBookmarkUpdater bookmarkUpdater, INotificationS
await notificationSender.SendAsync(new WorkflowBookmarksIndexed(new IndexedWorkflowBookmarks(updateBookmarksRequest.WorkflowInstanceId, updateBookmarksRequest.Diff.Added, updateBookmarksRequest.Diff.Removed, updateBookmarksRequest.Diff.Unchanged)));
// Publish domain event.
await notificationSender.SendAsync(new WorkflowBookmarksPersisted(updateBookmarksRequest.Diff), NotificationStrategy.Background);
await notificationSender.SendAsync(new WorkflowBookmarksPersisted(updateBookmarksRequest.Diff));
}
}

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

@ -3,11 +3,12 @@ using Elsa.Workflows.Helpers;
using Elsa.Workflows.Runtime.Filters;
using Elsa.Workflows.Runtime.Messages;
using Elsa.Workflows.Runtime.Options;
using Microsoft.Extensions.Logging;
namespace Elsa.Workflows.Runtime;
/// <inheritdoc />
public class BookmarkResumer(IWorkflowRuntime workflowRuntime, IBookmarkStore bookmarkStore, IStimulusHasher stimulusHasher) : IBookmarkResumer
public class BookmarkResumer(IWorkflowRuntime workflowRuntime, IBookmarkStore bookmarkStore, IStimulusHasher stimulusHasher, ILogger<BookmarkResumer> logger) : IBookmarkResumer
{
/// <inheritdoc />
public Task<ResumeBookmarkResult> ResumeAsync<TActivity>(object stimulus, ResumeBookmarkOptions? options, CancellationToken cancellationToken = default) where TActivity : IActivity
@ -35,7 +36,10 @@ public class BookmarkResumer(IWorkflowRuntime workflowRuntime, IBookmarkStore bo
var bookmark = await bookmarkStore.FindAsync(filter, cancellationToken);
if (bookmark == null)
{
logger.LogDebug("Bookmark not found in store for filter {@Filter}", filter);
return ResumeBookmarkResult.NotFound();
}
var workflowClient = await workflowRuntime.CreateClientAsync(bookmark.WorkflowInstanceId, cancellationToken);
var runRequest = new RunWorkflowInstanceRequest
@ -45,6 +49,7 @@ public class BookmarkResumer(IWorkflowRuntime workflowRuntime, IBookmarkStore bo
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);
}
}

8
src/modules/Elsa.Workflows.Runtime/Services/BookmarkUpdater.cs
View file

@ -18,8 +18,12 @@ public class BookmarkUpdater(IBookmarkManager bookmarkManager, IBookmarkStore bo
private async Task RemoveBookmarksAsync(string workflowInstanceId, IEnumerable<Bookmark> bookmarks, CancellationToken cancellationToken)
{
var matchingHashes = bookmarks.Select(x => x.Hash).ToList();
var filter = new BookmarkFilter { Hashes = matchingHashes, WorkflowInstanceId = workflowInstanceId };
var matchingIds = bookmarks.Select(x => x.Id).ToList();
var filter = new BookmarkFilter
{
BookmarkIds = matchingIds,
WorkflowInstanceId = workflowInstanceId
};
await bookmarkManager.DeleteManyAsync(filter, cancellationToken);
}

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

@ -2,10 +2,17 @@ using Elsa.Common.Contracts;
using Elsa.Workflows.Contracts;
using Elsa.Workflows.Runtime.Entities;
using Elsa.Workflows.Runtime.Filters;
using Microsoft.Extensions.Logging;
namespace Elsa.Workflows.Runtime;
public class StoreBookmarkQueue(IBookmarkQueueStore store, IBookmarkResumer resumer, IBookmarkQueueSignaler bookmarkQueueSignaler, ISystemClock systemClock, IIdentityGenerator identityGenerator) : IBookmarkQueue
public class StoreBookmarkQueue(
IBookmarkQueueStore store,
IBookmarkResumer resumer,
IBookmarkQueueSignaler bookmarkQueueSignaler,
ISystemClock systemClock,
IIdentityGenerator identityGenerator,
ILogger<StoreBookmarkQueue> logger) : IBookmarkQueue
{
public async Task EnqueueAsync(NewBookmarkQueueItem item, CancellationToken cancellationToken = default)
{
@ -20,9 +27,14 @@ public class StoreBookmarkQueue(IBookmarkQueueStore store, IBookmarkResumer resu
var result = await resumer.ResumeAsync(filter, item.Options, cancellationToken);
if (result.Matched)
{
logger.LogDebug("Successfully resumed workflow instance {WorkflowInstance} using bookmark {BookmarkId}", item.WorkflowInstanceId, item.BookmarkId);
return;
}
// There was no matching bookmark yet. Store the queue item for the system to pick up whenever the bookmark becomes present.
logger.LogDebug("No bookmark with ID {BookmarkId} found for workflow {WorkflowInstance}. Adding the request to the bookmark queue", item.BookmarkId, item.WorkflowInstanceId);
var entity = new BookmarkQueueItem
{
Id = identityGenerator.GenerateId(),

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

@ -1,12 +1,13 @@
using Elsa.Workflows.Management;
using Elsa.Workflows.State;
namespace Elsa.Workflows.Runtime;
public class StoreCommitStateHandler(IWorkflowInstanceManager workflowInstanceManager) : ICommitStateHandler
{
public async Task CommitAsync(WorkflowExecutionContext workflowExecutionContext, CancellationToken cancellationToken = default)
public async Task CommitAsync(WorkflowExecutionContext workflowExecutionContext, WorkflowState workflowState, CancellationToken cancellationToken = default)
{
await workflowInstanceManager.SaveAsync(workflowExecutionContext, cancellationToken);
await workflowInstanceManager.SaveAsync(workflowState, cancellationToken);
await workflowExecutionContext.ExecuteDeferredTasksAsync();
}
}