From 95346c98a3aaf5ed447e4fbbfeac0183368cbdae Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Tue, 5 Aug 2025 08:31:53 +0200 Subject: [PATCH 1/5] Fix RunMigrations Propagation Issue (#6838) * Introduce `MigrationOptions` for configurable EF Core migration execution Added `MigrationOptions` to control migration execution on a per-DbContext basis. Updated `RunMigrationsStartupTask` to respect these options, and refactored configurations in `PersistenceFeatureBase`. This fixes an issue where the `RunMigrations` setting is not properly propagated to dependency features. Fixes #6912 * Propagate `DbContextOptionsBuilder`, `UseContextPooling`, and `RunMigrations` settings to dependency features in `WorkflowManagementPersistenceFeature`. * Update src/modules/Elsa.EntityFrameworkCore.Common/RunMigrationsStartupTask.cs Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --------- Co-authored-by: Frans van Ek Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- .../MigrationOptions.cs | 14 ++++++++++++++ .../PersistenceFeatureBase.cs | 8 ++++++-- .../RunMigrationsStartupTask.cs | 10 ++++++++-- .../WorkflowManagementPersistenceFeature.cs | 4 ++-- 4 files changed, 30 insertions(+), 6 deletions(-) create mode 100644 src/modules/Elsa.EntityFrameworkCore.Common/MigrationOptions.cs diff --git a/src/modules/Elsa.EntityFrameworkCore.Common/MigrationOptions.cs b/src/modules/Elsa.EntityFrameworkCore.Common/MigrationOptions.cs new file mode 100644 index 000000000..0ff4cb70f --- /dev/null +++ b/src/modules/Elsa.EntityFrameworkCore.Common/MigrationOptions.cs @@ -0,0 +1,14 @@ +namespace Elsa.EntityFrameworkCore; + +public class MigrationOptions +{ + /// + /// Gets or sets a collection that determines whether Entity Framework Core migrations + /// should be executed for specific DbContext types. + /// + /// + /// The key is the DbContext type, and the value is a boolean indicating whether migrations + /// should be applied for that specific context (true to apply, false to skip). + /// + public IDictionary RunMigrations { get; set; } = new Dictionary(); +} \ No newline at end of file diff --git a/src/modules/Elsa.EntityFrameworkCore.Common/PersistenceFeatureBase.cs b/src/modules/Elsa.EntityFrameworkCore.Common/PersistenceFeatureBase.cs index 1fd3352c0..c6620abe7 100644 --- a/src/modules/Elsa.EntityFrameworkCore.Common/PersistenceFeatureBase.cs +++ b/src/modules/Elsa.EntityFrameworkCore.Common/PersistenceFeatureBase.cs @@ -40,8 +40,7 @@ public abstract class PersistenceFeatureBase(IModule modul public override void ConfigureHostedServices() { - if (RunMigrations) - ConfigureMigrations(); + ConfigureMigrations(); } /// @@ -62,6 +61,11 @@ public abstract class PersistenceFeatureBase(IModule modul Services.AddDbContextFactory(setup, DbContextFactoryLifetime); Services.Decorate, TenantAwareDbContextFactory>(); + + Services.Configure(options => + { + options.RunMigrations[typeof(TDbContext)] = RunMigrations; + }); } protected virtual void ConfigureMigrations() diff --git a/src/modules/Elsa.EntityFrameworkCore.Common/RunMigrationsStartupTask.cs b/src/modules/Elsa.EntityFrameworkCore.Common/RunMigrationsStartupTask.cs index 45f9c4a5e..0a37bc581 100644 --- a/src/modules/Elsa.EntityFrameworkCore.Common/RunMigrationsStartupTask.cs +++ b/src/modules/Elsa.EntityFrameworkCore.Common/RunMigrationsStartupTask.cs @@ -2,6 +2,7 @@ using Elsa.Common.RecurringTasks; using JetBrains.Annotations; using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.Options; namespace Elsa.EntityFrameworkCore; @@ -11,11 +12,16 @@ namespace Elsa.EntityFrameworkCore; [UsedImplicitly] [SingleNodeTask] [Order(-100)] -public class RunMigrationsStartupTask(IDbContextFactory dbContextFactory) : IStartupTask where TDbContext : DbContext +public class RunMigrationsStartupTask(IDbContextFactory dbContextFactory, IOptions options) : IStartupTask where TDbContext : DbContext { - /// public async Task ExecuteAsync(CancellationToken cancellationToken) { + bool shouldRunMigrations = false; + options.Value.RunMigrations.TryGetValue(typeof(TDbContext), out shouldRunMigrations); + if (!shouldRunMigrations) + return; + var dbContext = await dbContextFactory.CreateDbContextAsync(cancellationToken); await dbContext.Database.MigrateAsync(cancellationToken); } diff --git a/src/modules/Elsa.EntityFrameworkCore/Modules/Management/WorkflowManagementPersistenceFeature.cs b/src/modules/Elsa.EntityFrameworkCore/Modules/Management/WorkflowManagementPersistenceFeature.cs index e89f9a299..b741625f4 100644 --- a/src/modules/Elsa.EntityFrameworkCore/Modules/Management/WorkflowManagementPersistenceFeature.cs +++ b/src/modules/Elsa.EntityFrameworkCore/Modules/Management/WorkflowManagementPersistenceFeature.cs @@ -27,7 +27,7 @@ public class WorkflowManagementPersistenceFeature(IModule module) : PersistenceF public override bool UseContextPooling { - get => base.UseContextPooling; + get => base.UseContextPooling; set { base.UseContextPooling = value; @@ -38,7 +38,7 @@ public class WorkflowManagementPersistenceFeature(IModule module) : PersistenceF public override bool RunMigrations { - get => base.RunMigrations; + get => base.RunMigrations; set { base.RunMigrations = value; From adad649fa5bc3a752022f147c1901fdc7cd804f2 Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Tue, 5 Aug 2025 22:16:25 +0200 Subject: [PATCH 2/5] Update tenant HTTP prefix in appsettings.json for consistency --- src/apps/Elsa.Server.Web/appsettings.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/apps/Elsa.Server.Web/appsettings.json b/src/apps/Elsa.Server.Web/appsettings.json index 76b543702..bf9d5c532 100644 --- a/src/apps/Elsa.Server.Web/appsettings.json +++ b/src/apps/Elsa.Server.Web/appsettings.json @@ -44,7 +44,7 @@ "Name": "Tenant 1", "Configuration": { "Http": { - "Prefix": "/tenant-1", + "Prefix": "tenant-1", "Host": "localhost:5001" }, "ConnectionStrings": { From fc5612687d718c081214c34c2d9624480318d286 Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Tue, 5 Aug 2025 22:22:13 +0200 Subject: [PATCH 3/5] Refactors notification system for context (#6821) * Fix `CancellationToken` usage in `BackgroundCommandSenderHostedService` Corrected the `CancellationToken` parameter to use `commandContext.CancellationToken` instead of the method's cancellation token, ensuring proper propagation and handling within the command sender. Fixes #6449 * Refactor notifications system to use `NotificationContext` across channels and middleware * Update src/common/Elsa.Mediator/Extensions/HandlerExtensions.cs Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Simplify `InvokeAsync` method call by replacing braces with brackets in argument array. * Refactor `NotificationContext` usage in mediator strategies and handler extensions Replaced direct references to `Notification` and `CancellationToken` with `NotificationContext` across mediator strategies to align with updated context structure. Simplified handler invocations by passing `CommandContext` and `NotificationContext` where applicable. * Refactor notification extraction in publishing strategies Standardize notification retrieval by introducing `notificationContext.Notification` in `SequentialProcessingStrategy` and `ParallelProcessingStrategy` for improved clarity and consistency. * Refactor hosted services to improve worker channel management and cancellation handling Implemented better round-robin distribution and added linked cancellation token sources in `BackgroundCommandSender`, `JobRunner`, and `BackgroundEventPublisher` hosted services. Enhanced logging and comments for clarity. * Update src/common/Elsa.Mediator/HostedServices/BackgroundCommandSenderHostedService.cs Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --------- Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- .../Channels/NotificationsChannel.cs | 3 +- .../CommandStrategies/DefaultStrategy.cs | 3 +- .../Contexts/NotificationStrategyContext.cs | 5 ++-- .../Contracts/INotificationsChannel.cs | 9 +++--- .../Extensions/HandlerExtensions.cs | 15 ++++++---- .../BackgroundCommandSenderHostedService.cs | 30 +++++++++++++++---- .../BackgroundEventPublisherHostedService.cs | 29 ++++++++++++++---- .../HostedServices/JobRunnerHostedService.cs | 21 +++++++++---- .../NotificationHandlerInvokerMiddleware.cs | 2 +- .../BackgroundProcessingStrategy.cs | 2 +- .../ParallelProcessingStrategy.cs | 6 ++-- .../SequentialProcessingStrategy.cs | 6 ++-- 12 files changed, 92 insertions(+), 39 deletions(-) diff --git a/src/common/Elsa.Mediator/Channels/NotificationsChannel.cs b/src/common/Elsa.Mediator/Channels/NotificationsChannel.cs index 5480c3a60..676673630 100644 --- a/src/common/Elsa.Mediator/Channels/NotificationsChannel.cs +++ b/src/common/Elsa.Mediator/Channels/NotificationsChannel.cs @@ -1,9 +1,10 @@ using Elsa.Mediator.Abstractions; using Elsa.Mediator.Contracts; +using Elsa.Mediator.Middleware.Notification; namespace Elsa.Mediator.Channels; /// -public class NotificationsChannel : ChannelBase, INotificationsChannel +public class NotificationsChannel : ChannelBase, INotificationsChannel { } \ No newline at end of file diff --git a/src/common/Elsa.Mediator/CommandStrategies/DefaultStrategy.cs b/src/common/Elsa.Mediator/CommandStrategies/DefaultStrategy.cs index a1e26aa79..12bf6e2a0 100644 --- a/src/common/Elsa.Mediator/CommandStrategies/DefaultStrategy.cs +++ b/src/common/Elsa.Mediator/CommandStrategies/DefaultStrategy.cs @@ -14,11 +14,10 @@ public class DefaultStrategy : ICommandStrategy { var commandContext = context.CommandContext; var command = commandContext.Command; - var cancellationToken = context.CancellationToken; var commandType = command.GetType(); var handleMethod = commandType.GetCommandHandlerMethod(); var handler = context.Handler; - return await handler.InvokeAsync(handleMethod, command, cancellationToken); + return await handler.InvokeAsync(handleMethod, commandContext); } } \ No newline at end of file diff --git a/src/common/Elsa.Mediator/Contexts/NotificationStrategyContext.cs b/src/common/Elsa.Mediator/Contexts/NotificationStrategyContext.cs index b07339f82..7872e1a83 100644 --- a/src/common/Elsa.Mediator/Contexts/NotificationStrategyContext.cs +++ b/src/common/Elsa.Mediator/Contexts/NotificationStrategyContext.cs @@ -1,4 +1,5 @@ using Elsa.Mediator.Contracts; +using Elsa.Mediator.Middleware.Notification; using Microsoft.Extensions.Logging; namespace Elsa.Mediator.Contexts; @@ -6,9 +7,9 @@ namespace Elsa.Mediator.Contexts; /// /// Represents a context for publishing events. /// -/// The notification to publish. +/// The notification to publish. /// The handlers to publish the notification to. /// The logger. /// The service provider to resolve services from. /// The cancellation token. -public record NotificationStrategyContext(INotification Notification, INotificationHandler[] Handlers, ILogger Logger, IServiceProvider ServiceProvider, CancellationToken CancellationToken = default); \ No newline at end of file +public record NotificationStrategyContext(NotificationContext NotificationContext, INotificationHandler[] Handlers, ILogger Logger, IServiceProvider ServiceProvider, CancellationToken CancellationToken = default); \ No newline at end of file diff --git a/src/common/Elsa.Mediator/Contracts/INotificationsChannel.cs b/src/common/Elsa.Mediator/Contracts/INotificationsChannel.cs index 7ee0240f7..59f25449a 100644 --- a/src/common/Elsa.Mediator/Contracts/INotificationsChannel.cs +++ b/src/common/Elsa.Mediator/Contracts/INotificationsChannel.cs @@ -1,4 +1,5 @@ using System.Threading.Channels; +using Elsa.Mediator.Middleware.Notification; namespace Elsa.Mediator.Contracts; @@ -8,12 +9,12 @@ namespace Elsa.Mediator.Contracts; public interface INotificationsChannel { /// - /// Gets the writer for the notifications queue. + /// Gets the writer for the notification queue. /// - ChannelWriter Writer { get; } + ChannelWriter Writer { get; } /// - /// Gets the reader for the notifications queue. + /// Gets the reader for the notification queue. /// - ChannelReader Reader { get; } + ChannelReader Reader { get; } } \ No newline at end of file diff --git a/src/common/Elsa.Mediator/Extensions/HandlerExtensions.cs b/src/common/Elsa.Mediator/Extensions/HandlerExtensions.cs index cc5b55fd7..5ac381cb0 100644 --- a/src/common/Elsa.Mediator/Extensions/HandlerExtensions.cs +++ b/src/common/Elsa.Mediator/Extensions/HandlerExtensions.cs @@ -1,6 +1,8 @@ using System.Diagnostics.CodeAnalysis; using System.Reflection; using Elsa.Mediator.Contracts; +using Elsa.Mediator.Middleware.Command; +using Elsa.Mediator.Middleware.Notification; using Elsa.Mediator.Models; namespace Elsa.Mediator.Extensions; @@ -40,10 +42,12 @@ public static class HandlerExtensions /// /// The handler to invoke. /// The handle method. - /// The notification to handle. + /// The notification to handle. /// The cancellation token. - public static Task InvokeAsync(this INotificationHandler handler, MethodBase handleMethod, INotification notification, CancellationToken cancellationToken) + public static Task InvokeAsync(this INotificationHandler handler, MethodBase handleMethod, NotificationContext notificationContext) { + var notification = notificationContext.Notification; + var cancellationToken = notificationContext.CancellationToken; return (Task)handleMethod.Invoke(handler, [notification, cancellationToken])!; } @@ -52,10 +56,11 @@ public static class HandlerExtensions /// /// The handler to invoke. /// The handle method. - /// The command to handle. - /// The cancellation token. - public static Task InvokeAsync(this ICommandHandler handler, MethodBase handleMethod, ICommand command, CancellationToken cancellationToken) + /// The command to handle. + public static Task InvokeAsync(this ICommandHandler handler, MethodBase handleMethod, CommandContext commandContext) { + var command = commandContext.Command; + var cancellationToken = commandContext.CancellationToken; var task = (Task)handleMethod.Invoke(handler, [command, cancellationToken])!; return task; } diff --git a/src/common/Elsa.Mediator/HostedServices/BackgroundCommandSenderHostedService.cs b/src/common/Elsa.Mediator/HostedServices/BackgroundCommandSenderHostedService.cs index 47770bab5..f0606a1f5 100644 --- a/src/common/Elsa.Mediator/HostedServices/BackgroundCommandSenderHostedService.cs +++ b/src/common/Elsa.Mediator/HostedServices/BackgroundCommandSenderHostedService.cs @@ -24,49 +24,67 @@ public class BackgroundCommandSenderHostedService : BackgroundService public BackgroundCommandSenderHostedService(IOptions options, ICommandsChannel commandsChannel, IServiceScopeFactory scopeFactory, ILogger logger) { _workerCount = options.Value.CommandWorkerCount; - _commandsChannel = commandsChannel; + _commandsChannel = commandsChannel; // The shared input channel for all commands _scopeFactory = scopeFactory; _logger = logger; - _outputs = new(_workerCount); + _outputs = new(_workerCount); // Prepare a list to hold worker-specific channels } /// protected override async Task ExecuteAsync(CancellationToken cancellationToken) { - var index = 0; + var index = 0; // Used for round-robin distribution of work + // Set up worker channels and start background tasks for each worker for (var i = 0; i < _workerCount; i++) { var output = Channel.CreateUnbounded(); _outputs.Add(output); + // Start a background task that processes commands from this worker's channel _ = ReadOutputAsync(output, cancellationToken); } + // Main dispatcher loop: read from the input channel and distribute to worker channels await foreach (var commandContext in _commandsChannel.Reader.ReadAllAsync(cancellationToken)) { var output = _outputs[index]; await output.Writer.WriteAsync(commandContext, cancellationToken); + // Round-robin distribution - move to next worker index = (index + 1) % _workerCount; } - foreach (var output in _outputs) + // If the input channel is completed, complete all worker channels + foreach (var output in _outputs) output.Writer.Complete(); } private async Task ReadOutputAsync(Channel output, CancellationToken cancellationToken) { + // Worker task: process commands from the worker's channel await foreach (var commandContext in output.Reader.ReadAllAsync(cancellationToken)) { try { + // Create a fresh scope for each command to ensure proper service lifetime using var scope = _scopeFactory.CreateScope(); var commandSender = scope.ServiceProvider.GetRequiredService(); - await commandSender.SendAsync(commandContext.Command, CommandStrategy.Default, commandContext.Headers, cancellationToken); + // Link the service cancellation token with the command's token to ensure proper cancellation + using var linkedTokenSource = CancellationTokenSource.CreateLinkedTokenSource( + cancellationToken, + commandContext.CancellationToken); + + // Process the command using the command sender service with the linked token + await commandSender.SendAsync( + commandContext.Command, + CommandStrategy.Default, + commandContext.Headers, + linkedTokenSource.Token); } catch (Exception e) { - _logger.LogError(e, "An unhandled exception occured while processing the queue"); + // Log errors but continue processing other commands + _logger.LogError(e, "An unhandled exception occurred while processing the queue"); } } } diff --git a/src/common/Elsa.Mediator/HostedServices/BackgroundEventPublisherHostedService.cs b/src/common/Elsa.Mediator/HostedServices/BackgroundEventPublisherHostedService.cs index 67e43ddfb..384b09160 100644 --- a/src/common/Elsa.Mediator/HostedServices/BackgroundEventPublisherHostedService.cs +++ b/src/common/Elsa.Mediator/HostedServices/BackgroundEventPublisherHostedService.cs @@ -1,5 +1,6 @@ using System.Threading.Channels; using Elsa.Mediator.Contracts; +using Elsa.Mediator.Middleware.Notification; using Elsa.Mediator.Options; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; @@ -16,7 +17,7 @@ public class BackgroundEventPublisherHostedService : BackgroundService private readonly int _workerCount; private readonly INotificationsChannel _notificationsChannel; private readonly IServiceScopeFactory _scopeFactory; - private readonly List> _outputs; + private readonly List> _outputs; private readonly ILogger _logger; /// @@ -26,46 +27,62 @@ public class BackgroundEventPublisherHostedService : BackgroundService _notificationsChannel = notificationsChannel; _scopeFactory = scopeFactory; _logger = logger; - _outputs = new List>(_workerCount); + _outputs = new(_workerCount); } /// protected override async Task ExecuteAsync(CancellationToken cancellationToken) { + // Index to round-robin distribute notifications across worker channels var index = 0; using var scope = _scopeFactory.CreateScope(); var notificationSender = scope.ServiceProvider.GetRequiredService(); + // Create multiple output channels and start worker tasks for parallel processing for (var i = 0; i < _workerCount; i++) { - var output = Channel.CreateUnbounded(); + var output = Channel.CreateUnbounded(); _outputs.Add(output); + // Start a background task to process notifications from this output channel _ = ReadOutputAsync(output, notificationSender, cancellationToken); } var channelReader = _notificationsChannel.Reader; + // Continuously read notifications from the input channel and distribute them to worker channels + // using round-robin distribution for load balancing await foreach (var notification in channelReader.ReadAllAsync(cancellationToken)) { var output = _outputs[index]; await output.Writer.WriteAsync(notification, cancellationToken); + // Move to the next worker in a circular fashion index = (index + 1) % _workerCount; } + // When the input channel is completed, complete all output channels foreach (var output in _outputs) { output.Writer.Complete(); } } - private async Task ReadOutputAsync(Channel output, INotificationSender notificationSender, CancellationToken cancellationToken) + /// + /// Processes notifications from an output channel asynchronously. + /// + /// The channel to read notifications from + /// The service used to send notifications + /// Cancellation token from the hosted service + private async Task ReadOutputAsync(Channel output, INotificationSender notificationSender, CancellationToken cancellationToken) { - await foreach (var notification in output.Reader.ReadAllAsync(cancellationToken)) + await foreach (var notificationContext in output.Reader.ReadAllAsync(cancellationToken)) { try { - await notificationSender.SendAsync(notification, NotificationStrategy.Sequential, cancellationToken); + var notification = notificationContext.Notification; + // Link the cancellation tokens so that cancellation can happen from either source + using var linkedTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, notificationContext.CancellationToken); + await notificationSender.SendAsync(notification, NotificationStrategy.Sequential, linkedTokenSource.Token); } catch (OperationCanceledException e) { diff --git a/src/common/Elsa.Mediator/HostedServices/JobRunnerHostedService.cs b/src/common/Elsa.Mediator/HostedServices/JobRunnerHostedService.cs index 4eeec9061..14af18b4b 100644 --- a/src/common/Elsa.Mediator/HostedServices/JobRunnerHostedService.cs +++ b/src/common/Elsa.Mediator/HostedServices/JobRunnerHostedService.cs @@ -17,7 +17,7 @@ public class JobRunnerHostedService : BackgroundService /// public JobRunnerHostedService(IOptions options, IJobsChannel jobsChannel, ILogger logger) - { + { _workerCount = options.Value.JobWorkerCount; _jobsChannel = jobsChannel; _logger = logger; @@ -26,21 +26,31 @@ public class JobRunnerHostedService : BackgroundService /// protected override async Task ExecuteAsync(CancellationToken stoppingToken) { - var workers = new Task[_workerCount]; - - for (var i = 0; i < _workerCount; i++) + // Create an array of worker tasks to process jobs in parallel + var workers = new Task[_workerCount]; + + // Start multiple workers (tasks) to process jobs concurrently + for (var i = 0; i < _workerCount; i++) workers[i] = ProcessJobsAsync(stoppingToken); + // Wait for all worker tasks to complete (typically when the application is shutting down) await Task.WhenAll(workers); } + /// + /// Continuously processes jobs from the job channel until cancellation is requested. + /// + /// Cancellation token from the hosted service private async Task ProcessJobsAsync(CancellationToken stoppingToken) { + // Process all jobs from the channel until it's completed or cancellation is requested await foreach (var jobItem in _jobsChannel.Reader.ReadAllAsync(stoppingToken)) { try { - await jobItem.Action(jobItem.CancellationTokenSource.Token); + // Link the cancellation tokens so that cancellation can happen from either source + using var linkedTokenSource = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken, jobItem.CancellationTokenSource.Token); + await jobItem.Action(linkedTokenSource.Token); _logger.LogInformation("Worker {CurrentTaskId} processed job {JobId}", Task.CurrentId, jobItem.JobId); } catch (OperationCanceledException) @@ -53,6 +63,7 @@ public class JobRunnerHostedService : BackgroundService } finally { + // Notify that the job has completed (whether successfully or not) jobItem.OnJobCompleted(jobItem.JobId); } } diff --git a/src/common/Elsa.Mediator/Middleware/Notification/Components/NotificationHandlerInvokerMiddleware.cs b/src/common/Elsa.Mediator/Middleware/Notification/Components/NotificationHandlerInvokerMiddleware.cs index 874c64f17..e7e22125c 100644 --- a/src/common/Elsa.Mediator/Middleware/Notification/Components/NotificationHandlerInvokerMiddleware.cs +++ b/src/common/Elsa.Mediator/Middleware/Notification/Components/NotificationHandlerInvokerMiddleware.cs @@ -24,7 +24,7 @@ public class NotificationHandlerInvokerMiddleware( var serviceProvider = context.ServiceProvider; var notificationHandlers = serviceProvider.GetServices(); var handlers = notificationHandlers.Where(x => handlerType.IsInstanceOfType(x)).DistinctBy(x => x.GetType()).ToArray(); - var strategyContext = new NotificationStrategyContext(notification, handlers, logger, serviceProvider, context.CancellationToken); + var strategyContext = new NotificationStrategyContext(context, handlers, logger, serviceProvider, context.CancellationToken); await context.NotificationStrategy.PublishAsync(strategyContext); diff --git a/src/common/Elsa.Mediator/PublishingStrategies/BackgroundProcessingStrategy.cs b/src/common/Elsa.Mediator/PublishingStrategies/BackgroundProcessingStrategy.cs index b228c24a2..e960050f5 100644 --- a/src/common/Elsa.Mediator/PublishingStrategies/BackgroundProcessingStrategy.cs +++ b/src/common/Elsa.Mediator/PublishingStrategies/BackgroundProcessingStrategy.cs @@ -14,6 +14,6 @@ public class BackgroundProcessingStrategy : IEventPublishingStrategy { var notificationsChannel = context.ServiceProvider.GetRequiredService(); - await notificationsChannel.Writer.WriteAsync(context.Notification, context.CancellationToken); + await notificationsChannel.Writer.WriteAsync(context.NotificationContext, context.CancellationToken); } } \ No newline at end of file diff --git a/src/common/Elsa.Mediator/PublishingStrategies/ParallelProcessingStrategy.cs b/src/common/Elsa.Mediator/PublishingStrategies/ParallelProcessingStrategy.cs index fd16c8436..8eec00643 100644 --- a/src/common/Elsa.Mediator/PublishingStrategies/ParallelProcessingStrategy.cs +++ b/src/common/Elsa.Mediator/PublishingStrategies/ParallelProcessingStrategy.cs @@ -12,11 +12,11 @@ public class ParallelProcessingStrategy : IEventPublishingStrategy /// public async Task PublishAsync(NotificationStrategyContext context) { - var notification = context.Notification; - var cancellationToken = context.CancellationToken; + var notificationContext = context.NotificationContext; + var notification = notificationContext.Notification; var notificationType = notification.GetType(); var handleMethod = notificationType.GetNotificationHandlerMethod(); - var tasks = context.Handlers.Select(handler => handler.InvokeAsync(handleMethod, notification, cancellationToken)).ToList(); + var tasks = context.Handlers.Select(handler => handler.InvokeAsync(handleMethod, context.NotificationContext)).ToList(); await Task.WhenAll(tasks); } diff --git a/src/common/Elsa.Mediator/PublishingStrategies/SequentialProcessingStrategy.cs b/src/common/Elsa.Mediator/PublishingStrategies/SequentialProcessingStrategy.cs index ee7510445..494b9de2b 100644 --- a/src/common/Elsa.Mediator/PublishingStrategies/SequentialProcessingStrategy.cs +++ b/src/common/Elsa.Mediator/PublishingStrategies/SequentialProcessingStrategy.cs @@ -12,12 +12,12 @@ public class SequentialProcessingStrategy : IEventPublishingStrategy /// public async Task PublishAsync(NotificationStrategyContext context) { - var notification = context.Notification; - var cancellationToken = context.CancellationToken; + var notificationContext = context.NotificationContext; + var notification = notificationContext.Notification; var notificationType = notification.GetType(); var handleMethod = notificationType.GetNotificationHandlerMethod(); foreach (var handler in context.Handlers) - await handler.InvokeAsync(handleMethod, notification, cancellationToken); + await handler.InvokeAsync(handleMethod, context.NotificationContext); } } \ No newline at end of file From 133e8dfc63fb81aa06f6a5c9d9491f6047bd3ddb Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Wed, 6 Aug 2025 08:43:39 +0200 Subject: [PATCH 4/5] Introduce `TenantDeleted` event to handle tenant cleanup (#6843) * Introduce `TenantDeleted` event to handle tenant cleanup Added a `TenantDeleted` event to differentiate between tenant deactivation and deletion. Updated event handlers and services to support unregistering resources only during tenant deletion, ensuring clearer separation of responsibilities. Included ADR documentation for the new event. * Remove unnecessary blank line in tenant deactivation logic --- Elsa.sln | 1 + doc/adr/0005-tenant-deleted-event.md | 23 +++++++++ doc/adr/graph.dot | 18 ++++--- doc/adr/toc.md | 3 +- .../Contracts/ITenantDeletedEvent.cs | 6 +++ .../EventArgs/TenantDeletedEventArgs.cs | 3 ++ .../Implementations/DefaultTenantService.cs | 11 ++-- .../Implementations/TenantEventsManager.cs | 50 +++++++++++++------ .../Features/SchedulingFeature.cs | 2 +- .../Handlers/UpdateTenantSchedules.cs | 4 +- 10 files changed, 91 insertions(+), 30 deletions(-) create mode 100644 doc/adr/0005-tenant-deleted-event.md create mode 100644 src/modules/Elsa.Common/Multitenancy/Contracts/ITenantDeletedEvent.cs create mode 100644 src/modules/Elsa.Common/Multitenancy/EventArgs/TenantDeletedEventArgs.cs diff --git a/Elsa.sln b/Elsa.sln index a5cad563a..29feedd62 100644 --- a/Elsa.sln +++ b/Elsa.sln @@ -392,6 +392,7 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "adr", "adr", "{0A04B1FD-06C doc\adr\graph.dot = doc\adr\graph.dot doc\adr\0003-direct-bookmark-management-in-workflowexecutioncontext.md = doc\adr\0003-direct-bookmark-management-in-workflowexecutioncontext.md doc\adr\0004-activity-execution-snapshots.md = doc\adr\0004-activity-execution-snapshots.md + doc\adr\0005-tenant-deleted-event.md = doc\adr\0005-tenant-deleted-event.md EndProjectSection EndProject Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "bounty", "bounty", "{9B80A705-2E31-4012-964A-83963DCDB384}" diff --git a/doc/adr/0005-tenant-deleted-event.md b/doc/adr/0005-tenant-deleted-event.md new file mode 100644 index 000000000..d0d2fa9f1 --- /dev/null +++ b/doc/adr/0005-tenant-deleted-event.md @@ -0,0 +1,23 @@ +# 5. Tenant Deleted Event + +Date: 2025-08-05 + +## Status + +Accepted + +## Context + +As outlined in issue [#6661](https://github.com/elsa-workflows/elsa-core/issues/6661), there is a need to differentiate between **Tenant Deactivating** and **Tenant Deleting** events. + +Currently, the `TenantDeactivated` event is used to unregister timer-based triggers. However, this causes the triggers to be unregistered when the application host shuts down, which is not the desired behavior. Instead, we want the triggers to remain registered until the tenant is explicitly deleted. + +## Decision + +To address this, we will introduce a new event called `TenantDeleted`. This event will be raised when a tenant is deleted and will be responsible for unregistering timer-based triggers. This ensures that the triggers remain active until the tenant is explicitly deleted. + +## Consequences + +- Timer-based triggers will no longer be unregistered during tenant deactivation. Instead, they will remain active until the tenant is deleted. +- The `TenantDeactivated` event will continue to be used for deactivating tenants without affecting the registration of timer-based triggers. +- The new `TenantDeleted` event will specifically handle the cleanup of resources associated with a tenant when it is deleted, ensuring a clear separation of responsibilities. diff --git a/doc/adr/graph.dot b/doc/adr/graph.dot index 61e1e6f7c..4818df2a2 100644 --- a/doc/adr/graph.dot +++ b/doc/adr/graph.dot @@ -1,12 +1,14 @@ digraph { -node [shape = plaintext]; +node [shape=plaintext]; subgraph { -_1 [label = "1. Record architecture decisions"; URL = "0001-record-architecture-decisions.html"]; -_2 [label = "2. Fault Propagation from Child to Parent Activities"; URL ="0002-fault-propagation-from-child-to-parent-activities.html"]; -_1 -> _2 [style= "dotted", weight = 1]; -_3 [label = "3. Direct Bookmark Management in WorkflowExecutionContext"; URL = "0003-direct-bookmark-management-in-workflowexecutioncontext.html"]; -_2 -> _3 [style = "dotted", weight = 1]; -_4 [label ="4. Activity Execution Snapshots"; URL = "0004-activity-execution-snapshots.html"]; -_3 -> _4 [style = "dotted", weight = 1]; +_1 [label="1. Record architecture decisions"; URL="0001-record-architecture-decisions.html"]; +_2 [label="2. Fault Propagation from Child to Parent Activities"; URL="0002-fault-propagation-from-child-to-parent-activities.html"]; +_1 -> _2 [style="dotted", weight=1]; +_3 [label="3. Direct Bookmark Management in WorkflowExecutionContext"; URL="0003-direct-bookmark-management-in-workflowexecutioncontext.html"]; +_2 -> _3 [style="dotted", weight=1]; +_4 [label="4. Activity Execution Snapshots"; URL="0004-activity-execution-snapshots.html"]; +_3 -> _4 [style="dotted", weight=1]; +_5 [label="5. Tenant Deleted Event"; URL="0005-tenant-deleted-event.html"]; +_4 -> _5 [style="dotted", weight=1]; } } \ No newline at end of file diff --git a/doc/adr/toc.md b/doc/adr/toc.md index 7eecf8af6..26b15683d 100644 --- a/doc/adr/toc.md +++ b/doc/adr/toc.md @@ -3,4 +3,5 @@ * [1. Record architecture decisions](0001-record-architecture-decisions.md) * [2. Fault Propagation from Child to Parent Activities](0002-fault-propagation-from-child-to-parent-activities.md) * [3. Direct Bookmark Management in WorkflowExecutionContext](0003-direct-bookmark-management-in-workflowexecutioncontext.md) -* [4. Activity Execution Snapshots](0004-activity-execution-snapshots.md) \ No newline at end of file +* [4. Activity Execution Snapshots](0004-activity-execution-snapshots.md) +* [5. Tenant Deleted Event](0005-tenant-deleted-event.md) \ No newline at end of file diff --git a/src/modules/Elsa.Common/Multitenancy/Contracts/ITenantDeletedEvent.cs b/src/modules/Elsa.Common/Multitenancy/Contracts/ITenantDeletedEvent.cs new file mode 100644 index 000000000..f651cdf70 --- /dev/null +++ b/src/modules/Elsa.Common/Multitenancy/Contracts/ITenantDeletedEvent.cs @@ -0,0 +1,6 @@ +namespace Elsa.Common.Multitenancy; + +public interface ITenantDeletedEvent +{ + Task TenantDeletedAsync(TenantDeletedEventArgs args); +} \ No newline at end of file diff --git a/src/modules/Elsa.Common/Multitenancy/EventArgs/TenantDeletedEventArgs.cs b/src/modules/Elsa.Common/Multitenancy/EventArgs/TenantDeletedEventArgs.cs new file mode 100644 index 000000000..07f6f1136 --- /dev/null +++ b/src/modules/Elsa.Common/Multitenancy/EventArgs/TenantDeletedEventArgs.cs @@ -0,0 +1,3 @@ +namespace Elsa.Common.Multitenancy; + +public record TenantDeletedEventArgs(Tenant Tenant, TenantScope TenantScope, CancellationToken CancellationToken) : TenantEventArgs(Tenant, TenantScope, CancellationToken); \ No newline at end of file diff --git a/src/modules/Elsa.Common/Multitenancy/Implementations/DefaultTenantService.cs b/src/modules/Elsa.Common/Multitenancy/Implementations/DefaultTenantService.cs index 89298e5ac..6a1f9cb98 100644 --- a/src/modules/Elsa.Common/Multitenancy/Implementations/DefaultTenantService.cs +++ b/src/modules/Elsa.Common/Multitenancy/Implementations/DefaultTenantService.cs @@ -64,7 +64,7 @@ public class DefaultTenantService(IServiceScopeFactory scopeFactory, ITenantScop var tenants = dictionary.Values.ToArray(); foreach (var tenant in tenants) - await UnregisterTenantAsync(tenant, cancellationToken); + await UnregisterTenantAsync(tenant, false, cancellationToken); } public async Task RefreshAsync(CancellationToken cancellationToken = default) @@ -85,7 +85,7 @@ public class DefaultTenantService(IServiceScopeFactory scopeFactory, ITenantScop foreach (var removedTenantId in removedTenantIds) { var removedTenant = currentTenants[removedTenantId]; - await UnregisterTenantAsync(removedTenant, cancellationToken); + await UnregisterTenantAsync(removedTenant, true, cancellationToken); } foreach (var addedTenantId in addedTenantIds) @@ -137,14 +137,19 @@ public class DefaultTenantService(IServiceScopeFactory scopeFactory, ITenantScop await tenantEvents.TenantActivatedAsync(new(tenant, scope, cancellationToken)); } - private async Task UnregisterTenantAsync(Tenant tenant, CancellationToken cancellationToken = default) + private async Task UnregisterTenantAsync(Tenant tenant, bool isDeleted, CancellationToken cancellationToken = default) { if (_tenantScopesDictionary!.Remove(tenant, out var scope)) { _tenantsDictionary!.Remove(tenant.Id.EmptyIfNull(), out _); using (tenantAccessor.PushContext(tenant)) + { await tenantEvents.TenantDeactivatedAsync(new(tenant, scope, cancellationToken)); + + if (isDeleted) + await tenantEvents.TenantDeletedAsync(new(tenant, scope, cancellationToken)); + } } } } \ No newline at end of file diff --git a/src/modules/Elsa.Common/Multitenancy/Implementations/TenantEventsManager.cs b/src/modules/Elsa.Common/Multitenancy/Implementations/TenantEventsManager.cs index 1c8e85a85..c8b97639a 100644 --- a/src/modules/Elsa.Common/Multitenancy/Implementations/TenantEventsManager.cs +++ b/src/modules/Elsa.Common/Multitenancy/Implementations/TenantEventsManager.cs @@ -2,34 +2,54 @@ using Microsoft.Extensions.Logging; namespace Elsa.Common.Multitenancy; -public class TenantEventsManager(IEnumerable tenantActivatedEvents, IEnumerable tenantDeactivatedEvents, ILogger logger) +public class TenantEventsManager( + IEnumerable tenantActivatedEvents, + IEnumerable tenantDeactivatedEvents, + IEnumerable tenantDeletedEvents, + ILogger logger) { public async Task TenantActivatedAsync(TenantActivatedEventArgs args) { - foreach (var tenantActivatedEvent in tenantActivatedEvents) - { - try - { - await tenantActivatedEvent.TenantActivatedAsync(args); - } - catch (Exception e) - { - logger.LogError(e, "Error occurred while processing tenant activated event."); - } - } + await ExecuteEventHandlersAsync( + tenantActivatedEvents, + (handler, eventArgs) => handler.TenantActivatedAsync(eventArgs), + args, + "activated"); } public async Task TenantDeactivatedAsync(TenantDeactivatedEventArgs args) { - foreach (var tenantDeactivatedEvent in tenantDeactivatedEvents) + await ExecuteEventHandlersAsync( + tenantDeactivatedEvents, + (handler, eventArgs) => handler.TenantDeactivatedAsync(eventArgs), + args, + "deactivated"); + } + + public async Task TenantDeletedAsync(TenantDeletedEventArgs args) + { + await ExecuteEventHandlersAsync( + tenantDeletedEvents, + (handler, eventArgs) => handler.TenantDeletedAsync(eventArgs), + args, + "deleted"); + } + + private async Task ExecuteEventHandlersAsync( + IEnumerable handlers, + Func handlerAction, + TArgs args, + string eventType) + { + foreach (var handler in handlers) { try { - await tenantDeactivatedEvent.TenantDeactivatedAsync(args); + await handlerAction(handler, args); } catch (Exception e) { - logger.LogError(e, "Error occurred while processing tenant deactivated event."); + logger.LogError(e, "Error occurred while processing tenant {EventType} event.", eventType); } } } diff --git a/src/modules/Elsa.Scheduling/Features/SchedulingFeature.cs b/src/modules/Elsa.Scheduling/Features/SchedulingFeature.cs index 7ef3dfdec..33bf2eff4 100644 --- a/src/modules/Elsa.Scheduling/Features/SchedulingFeature.cs +++ b/src/modules/Elsa.Scheduling/Features/SchedulingFeature.cs @@ -39,7 +39,7 @@ public class SchedulingFeature : FeatureBase Services .AddSingleton() .AddSingleton(sp => sp.GetRequiredService()) - .AddSingleton(sp => sp.GetRequiredService()) + .AddSingleton(sp => sp.GetRequiredService()) .AddSingleton() .AddSingleton() .AddSingleton(CronParser) diff --git a/src/modules/Elsa.Scheduling/Handlers/UpdateTenantSchedules.cs b/src/modules/Elsa.Scheduling/Handlers/UpdateTenantSchedules.cs index bdeecad26..46456cb9b 100644 --- a/src/modules/Elsa.Scheduling/Handlers/UpdateTenantSchedules.cs +++ b/src/modules/Elsa.Scheduling/Handlers/UpdateTenantSchedules.cs @@ -9,7 +9,7 @@ using Timer = Elsa.Scheduling.Activities.Timer; namespace Elsa.Scheduling.Handlers; -public class UpdateTenantSchedules : ITenantActivatedEvent, ITenantDeactivatedEvent +public class UpdateTenantSchedules : ITenantActivatedEvent, ITenantDeletedEvent { private static readonly string[] ActivityTypeNames = [ @@ -31,7 +31,7 @@ public class UpdateTenantSchedules : ITenantActivatedEvent, ITenantDeactivatedEv await bookmarkScheduler.ScheduleAsync(bookmarks, args.CancellationToken); } - public async Task TenantDeactivatedAsync(TenantDeactivatedEventArgs args) + public async Task TenantDeletedAsync(TenantDeletedEventArgs args) { var serviceProvider = args.TenantScope.ServiceProvider; var cancellationToken = args.CancellationToken; From 1570cc929eef5f60ff45876d626066c59c69bdff Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Wed, 6 Aug 2025 08:47:23 +0200 Subject: [PATCH 5/5] Send `ActivityCompleted` notification from middleware and remove redundant notification logic in activity execution context. (#6842) * Send `ActivityCompleted` notification from middleware and remove redundant notification logic in activity execution context. * Update src/modules/Elsa.Workflows.Core/Middleware/Activities/DefaultActivityInvokerMiddleware.cs Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Update src/modules/Elsa.Workflows.Core/Middleware/Activities/DefaultActivityInvokerMiddleware.cs Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --------- Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- .../Contexts/ActivityExecutionContext.Complete.cs | 3 --- .../Activities/DefaultActivityInvokerMiddleware.cs | 12 ++++++++++++ 2 files changed, 12 insertions(+), 3 deletions(-) diff --git a/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.Complete.cs b/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.Complete.cs index faa250d04..7e5f87c59 100644 --- a/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.Complete.cs +++ b/src/modules/Elsa.Workflows.Core/Contexts/ActivityExecutionContext.Complete.cs @@ -63,9 +63,6 @@ public partial class ActivityExecutionContext // Update the completed at timestamp. CompletedAt = WorkflowExecutionContext.SystemClock.UtcNow; - - var mediator = GetRequiredService(); - await mediator.SendAsync(new Notifications.ActivityCompleted(this), CancellationToken); } /// diff --git a/src/modules/Elsa.Workflows.Core/Middleware/Activities/DefaultActivityInvokerMiddleware.cs b/src/modules/Elsa.Workflows.Core/Middleware/Activities/DefaultActivityInvokerMiddleware.cs index 6731bb8a6..24ffb4939 100644 --- a/src/modules/Elsa.Workflows.Core/Middleware/Activities/DefaultActivityInvokerMiddleware.cs +++ b/src/modules/Elsa.Workflows.Core/Middleware/Activities/DefaultActivityInvokerMiddleware.cs @@ -1,5 +1,6 @@ using System.Reflection; using Elsa.Extensions; +using Elsa.Mediator.Contracts; using Elsa.Workflows.Activities; using Elsa.Workflows.CommitStates; using Elsa.Workflows.Pipelines.ActivityExecution; @@ -59,10 +60,14 @@ public class DefaultActivityInvokerMiddleware(ActivityMiddlewareDelegate next, I if (ShouldCommit(context, ActivityLifetimeEvent.ActivityExecuting)) await context.WorkflowExecutionContext.CommitAsync(); + var previousActivityStatus = context.Status; context.TransitionTo(ActivityStatus.Running); // Execute activity. await ExecuteActivityAsync(context); + + var currentActivityStatus = context.Status; + var activityDidComplete = previousActivityStatus != ActivityStatus.Completed && currentActivityStatus == ActivityStatus.Completed; // Reset execute delegate. workflowExecutionContext.ExecuteDelegate = null; @@ -81,6 +86,13 @@ public class DefaultActivityInvokerMiddleware(ActivityMiddlewareDelegate next, I // Invoke next middleware. await next(context); + + // If the activity completed, send a notification. + if (activityDidComplete) + { + var mediator = context.GetRequiredService(); + await mediator.SendAsync(new Notifications.ActivityCompleted(context), context.CancellationToken); + } // Conditionally commit the workflow state. if (ShouldCommit(context, ActivityLifetimeEvent.ActivityExecuted))