From 6741db388baeca474a4e694b1e190d8a5c1ddb2e Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Sat, 10 Aug 2024 10:41:35 +0200 Subject: [PATCH] Fix background activity completion (#5882) * 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. --- .../appsettings.json | 3 ++- .../ActivityProviders/AgentActivityProvider.cs | 1 + .../Contracts/ICommitStateHandler.cs | 4 +++- .../Services/NoopCommitStateHandler.cs | 4 +++- .../Elsa.Workflows.Core/Services/WorkflowRunner.cs | 2 +- .../Features/DistributedRuntimeFeature.cs | 1 - .../Services/DistributedWorkflowClient.cs | 2 +- .../Services/DistributedWorkflowRuntime.cs | 2 +- .../Services/BookmarkPersister.cs | 3 +-- .../Services/BookmarkResumer.cs | 7 ++++++- .../Services/BookmarkUpdater.cs | 8 ++++++-- .../Services/StoreBookmarkQueue.cs | 14 +++++++++++++- .../Services/StoreCommitStateHandler.cs | 5 +++-- 13 files changed, 41 insertions(+), 15 deletions(-) diff --git a/samples/aspnet/Elsa.Samples.AspNet.OrchardCoreIntegration/appsettings.json b/samples/aspnet/Elsa.Samples.AspNet.OrchardCoreIntegration/appsettings.json index 8a26f369c..345470f8e 100644 --- a/samples/aspnet/Elsa.Samples.AspNet.OrchardCoreIntegration/appsettings.json +++ b/samples/aspnet/Elsa.Samples.AspNet.OrchardCoreIntegration/appsettings.json @@ -2,7 +2,8 @@ "Logging": { "LogLevel": { "Default": "Information", - "Microsoft.AspNetCore": "Warning" + "Microsoft.AspNetCore": "Warning", + "Microsoft.EntityFrameworkCore": "Warning" } }, "AllowedHosts": "*", diff --git a/src/modules/Elsa.Agents.Activities/ActivityProviders/AgentActivityProvider.cs b/src/modules/Elsa.Agents.Activities/ActivityProviders/AgentActivityProvider.cs index 836912a71..6c0b3249b 100644 --- a/src/modules/Elsa.Agents.Activities/ActivityProviders/AgentActivityProvider.cs +++ b/src/modules/Elsa.Agents.Activities/ActivityProviders/AgentActivityProvider.cs @@ -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 => { diff --git a/src/modules/Elsa.Workflows.Core/Contracts/ICommitStateHandler.cs b/src/modules/Elsa.Workflows.Core/Contracts/ICommitStateHandler.cs index 994c409f2..3d24ee6fd 100644 --- a/src/modules/Elsa.Workflows.Core/Contracts/ICommitStateHandler.cs +++ b/src/modules/Elsa.Workflows.Core/Contracts/ICommitStateHandler.cs @@ -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); } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Core/Services/NoopCommitStateHandler.cs b/src/modules/Elsa.Workflows.Core/Services/NoopCommitStateHandler.cs index 5d2f6092c..8503bddb7 100644 --- a/src/modules/Elsa.Workflows.Core/Services/NoopCommitStateHandler.cs +++ b/src/modules/Elsa.Workflows.Core/Services/NoopCommitStateHandler.cs @@ -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; } diff --git a/src/modules/Elsa.Workflows.Core/Services/WorkflowRunner.cs b/src/modules/Elsa.Workflows.Core/Services/WorkflowRunner.cs index bf0f21682..c5d54735d 100644 --- a/src/modules/Elsa.Workflows.Core/Services/WorkflowRunner.cs +++ b/src/modules/Elsa.Workflows.Core/Services/WorkflowRunner.cs @@ -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); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime.Distributed/Features/DistributedRuntimeFeature.cs b/src/modules/Elsa.Workflows.Runtime.Distributed/Features/DistributedRuntimeFeature.cs index d9c22f541..49a6526fe 100644 --- a/src/modules/Elsa.Workflows.Runtime.Distributed/Features/DistributedRuntimeFeature.cs +++ b/src/modules/Elsa.Workflows.Runtime.Distributed/Features/DistributedRuntimeFeature.cs @@ -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; diff --git a/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowClient.cs b/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowClient.cs index afb7932f0..912526509 100644 --- a/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowClient.cs +++ b/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowClient.cs @@ -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, diff --git a/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowRuntime.cs b/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowRuntime.cs index fc49cc74d..fb884fa87 100644 --- a/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowRuntime.cs +++ b/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowRuntime.cs @@ -1,7 +1,7 @@ using Elsa.Workflows.Contracts; using Microsoft.Extensions.DependencyInjection; -namespace Elsa.Workflows.Runtime.Distributed.Services; +namespace Elsa.Workflows.Runtime.Distributed; /// /// Represents a distributed workflow runtime that can create instances connected to a workflow instance. diff --git a/src/modules/Elsa.Workflows.Runtime/Services/BookmarkPersister.cs b/src/modules/Elsa.Workflows.Runtime/Services/BookmarkPersister.cs index 522068341..630eaa645 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/BookmarkPersister.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/BookmarkPersister.cs @@ -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)); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Services/BookmarkResumer.cs b/src/modules/Elsa.Workflows.Runtime/Services/BookmarkResumer.cs index 703fd6e1e..0643c221d 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/BookmarkResumer.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/BookmarkResumer.cs @@ -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; /// -public class BookmarkResumer(IWorkflowRuntime workflowRuntime, IBookmarkStore bookmarkStore, IStimulusHasher stimulusHasher) : IBookmarkResumer +public class BookmarkResumer(IWorkflowRuntime workflowRuntime, IBookmarkStore bookmarkStore, IStimulusHasher stimulusHasher, ILogger logger) : IBookmarkResumer { /// public Task ResumeAsync(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); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Services/BookmarkUpdater.cs b/src/modules/Elsa.Workflows.Runtime/Services/BookmarkUpdater.cs index 41813bd00..2a5cd0359 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/BookmarkUpdater.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/BookmarkUpdater.cs @@ -18,8 +18,12 @@ public class BookmarkUpdater(IBookmarkManager bookmarkManager, IBookmarkStore bo private async Task RemoveBookmarksAsync(string workflowInstanceId, IEnumerable 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); } diff --git a/src/modules/Elsa.Workflows.Runtime/Services/StoreBookmarkQueue.cs b/src/modules/Elsa.Workflows.Runtime/Services/StoreBookmarkQueue.cs index 096e9f4e9..003360b09 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/StoreBookmarkQueue.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/StoreBookmarkQueue.cs @@ -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 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(), diff --git a/src/modules/Elsa.Workflows.Runtime/Services/StoreCommitStateHandler.cs b/src/modules/Elsa.Workflows.Runtime/Services/StoreCommitStateHandler.cs index fb30614d5..8f3d33466 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/StoreCommitStateHandler.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/StoreCommitStateHandler.cs @@ -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(); } } \ No newline at end of file