| | | 1 | | using System.Reflection; |
| | | 2 | | using Elsa.Common.Helpers; |
| | | 3 | | using Elsa.Common.RecurringTasks; |
| | | 4 | | using Microsoft.Extensions.DependencyInjection; |
| | | 5 | | using Microsoft.Extensions.Logging; |
| | | 6 | | |
| | | 7 | | namespace Elsa.Common.Multitenancy.EventHandlers; |
| | | 8 | | |
| | | 9 | | /// <summary> |
| | | 10 | | /// Manages the lifecycle of startup, background, and recurring tasks for tenants. |
| | | 11 | | /// Executes tasks in the proper sequence: startup tasks first, then background tasks, then recurring tasks. |
| | | 12 | | /// </summary> |
| | 3 | 13 | | public class TenantTaskManager(RecurringTaskScheduleManager scheduleManager, ILogger<TenantTaskManager> logger) : ITenan |
| | | 14 | | { |
| | 3 | 15 | | private readonly ICollection<Task> _runningBackgroundTasks = new List<Task>(); |
| | 3 | 16 | | private readonly ICollection<ScheduledTimer> _scheduledTimers = new List<ScheduledTimer>(); |
| | | 17 | | private CancellationTokenSource _cancellationTokenSource = null!; |
| | | 18 | | |
| | | 19 | | public async Task TenantActivatedAsync(TenantActivatedEventArgs args) |
| | | 20 | | { |
| | 12 | 21 | | var cancellationToken = args.CancellationToken; |
| | 12 | 22 | | var tenantScope = args.TenantScope; |
| | 12 | 23 | | var taskExecutor = tenantScope.ServiceProvider.GetRequiredService<ITaskExecutor>(); |
| | | 24 | | |
| | 12 | 25 | | _cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); |
| | | 26 | | |
| | | 27 | | // Step 1: Run startup tasks (with dependency ordering) |
| | 12 | 28 | | await RunStartupTasksAsync(tenantScope, taskExecutor, cancellationToken); |
| | | 29 | | |
| | | 30 | | // Step 2: Run background tasks |
| | 12 | 31 | | await RunBackgroundTasksAsync(tenantScope, taskExecutor, cancellationToken); |
| | | 32 | | |
| | | 33 | | // Step 3: Start recurring tasks |
| | 12 | 34 | | await StartRecurringTasksAsync(tenantScope, taskExecutor, cancellationToken); |
| | 12 | 35 | | } |
| | | 36 | | |
| | | 37 | | public async Task TenantDeactivatedAsync(TenantDeactivatedEventArgs args) |
| | | 38 | | { |
| | 12 | 39 | | var tenantScope = args.TenantScope; |
| | | 40 | | |
| | | 41 | | // Cancel all running tasks |
| | 12 | 42 | | _cancellationTokenSource.Cancel(); |
| | | 43 | | |
| | | 44 | | // Wait for background tasks to complete (with cancellation they should finish quickly) |
| | 12 | 45 | | if (_runningBackgroundTasks.Any()) |
| | | 46 | | { |
| | | 47 | | try |
| | | 48 | | { |
| | 3 | 49 | | await Task.WhenAll(_runningBackgroundTasks); |
| | 3 | 50 | | } |
| | 0 | 51 | | catch (OperationCanceledException) |
| | | 52 | | { |
| | | 53 | | // Expected when tasks are cancelled |
| | 0 | 54 | | } |
| | 3 | 55 | | _runningBackgroundTasks.Clear(); |
| | | 56 | | } |
| | | 57 | | |
| | | 58 | | // Stop all recurring task timers |
| | 96 | 59 | | foreach (var timer in _scheduledTimers) |
| | 36 | 60 | | await timer.DisposeAsync(); |
| | 12 | 61 | | _scheduledTimers.Clear(); |
| | | 62 | | |
| | | 63 | | // Stop recurring tasks |
| | 12 | 64 | | var recurringTasks = tenantScope.ServiceProvider.GetServices<IRecurringTask>(); |
| | 96 | 65 | | foreach (var task in recurringTasks) |
| | 36 | 66 | | await task.StopAsync(args.CancellationToken); |
| | 12 | 67 | | } |
| | | 68 | | |
| | | 69 | | private async Task RunStartupTasksAsync(ITenantScope tenantScope, ITaskExecutor taskExecutor, CancellationToken canc |
| | | 70 | | { |
| | 12 | 71 | | var startupTasks = tenantScope.ServiceProvider.GetServices<IStartupTask>() |
| | 108 | 72 | | .OrderBy(x => x.GetType().GetCustomAttribute<OrderAttribute>()?.Order ?? 0f) |
| | 12 | 73 | | .ToList(); |
| | | 74 | | |
| | | 75 | | // First apply OrderAttribute to determine a base order, then perform topological sorting. |
| | | 76 | | // The topological sort is the final ordering step to ensure dependency constraints are respected. |
| | 12 | 77 | | var sortedTasks = TopologicalTaskSorter.Sort(startupTasks).ToList(); |
| | 240 | 78 | | foreach (var task in sortedTasks) |
| | 108 | 79 | | await taskExecutor.ExecuteTaskAsync(task, cancellationToken); |
| | 12 | 80 | | } |
| | | 81 | | |
| | | 82 | | private Task RunBackgroundTasksAsync(ITenantScope tenantScope, ITaskExecutor taskExecutor, CancellationToken cancell |
| | | 83 | | { |
| | 12 | 84 | | var backgroundTasks = tenantScope.ServiceProvider.GetServices<IBackgroundTask>(); |
| | 12 | 85 | | var backgroundTaskStarter = tenantScope.ServiceProvider.GetRequiredService<IBackgroundTaskStarter>(); |
| | | 86 | | |
| | 48 | 87 | | foreach (var backgroundTask in backgroundTasks) |
| | | 88 | | { |
| | 12 | 89 | | var task = backgroundTaskStarter |
| | 12 | 90 | | .StartAsync(backgroundTask, _cancellationTokenSource.Token) |
| | 12 | 91 | | .ContinueWith(t => taskExecutor.ExecuteTaskAsync(backgroundTask, _cancellationTokenSource.Token), |
| | 12 | 92 | | cancellationToken, |
| | 12 | 93 | | TaskContinuationOptions.RunContinuationsAsynchronously, |
| | 12 | 94 | | TaskScheduler.Default) |
| | 12 | 95 | | .Unwrap(); |
| | | 96 | | |
| | 12 | 97 | | if (!task.IsCompleted) |
| | 12 | 98 | | _runningBackgroundTasks.Add(task); |
| | | 99 | | } |
| | | 100 | | |
| | 12 | 101 | | return Task.CompletedTask; |
| | | 102 | | } |
| | | 103 | | |
| | | 104 | | private async Task StartRecurringTasksAsync(ITenantScope tenantScope, ITaskExecutor taskExecutor, CancellationToken |
| | | 105 | | { |
| | 12 | 106 | | var recurringTasks = tenantScope.ServiceProvider.GetServices<IRecurringTask>().ToList(); |
| | | 107 | | |
| | 96 | 108 | | foreach (var task in recurringTasks) |
| | | 109 | | { |
| | 36 | 110 | | var schedule = scheduleManager.GetScheduleFor(task.GetType()); |
| | 36 | 111 | | var timer = schedule.CreateTimer(async () => |
| | 36 | 112 | | { |
| | 36 | 113 | | try |
| | 36 | 114 | | { |
| | 76 | 115 | | await taskExecutor.ExecuteTaskAsync(task, _cancellationTokenSource.Token); |
| | 76 | 116 | | } |
| | 0 | 117 | | catch (OperationCanceledException e) |
| | 36 | 118 | | { |
| | 0 | 119 | | logger.LogInformation(e, "Recurring task {TaskType} was cancelled", task.GetType().Name); |
| | 0 | 120 | | } |
| | 0 | 121 | | catch (Exception e) |
| | 36 | 122 | | { |
| | 36 | 123 | | // Log but don't rethrow - recurring tasks should not crash the host |
| | 0 | 124 | | logger.LogError(e, "Recurring task {TaskType} failed with an error", task.GetType().Name); |
| | 0 | 125 | | } |
| | 112 | 126 | | }, logger); |
| | | 127 | | |
| | 36 | 128 | | _scheduledTimers.Add(timer); |
| | 36 | 129 | | await task.StartAsync(cancellationToken); |
| | | 130 | | } |
| | 12 | 131 | | } |
| | | 132 | | } |