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

Merge pull request #6259 from elsa-workflows/bug/6251

Browse source 
Refactor recurring tasks scheduling logic.
This commit is contained in:
Sipke Schoorstra 2025-01-02 11:12:45 +01:00 committed by GitHub
parent d3af106c32 6c0f7a48e7
commit be68b4c33f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 41 additions and 42 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
src/apps/Elsa.Server.Web/Elsa.Server.Web.csproj
View file

@ -14,6 +14,7 @@
<ProjectReference Include="..\..\modules\Elsa.MassTransit.AzureServiceBus\Elsa.MassTransit.AzureServiceBus.csproj"/>
<ProjectReference Include="..\..\modules\Elsa.OpenTelemetry\Elsa.OpenTelemetry.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Quartz.EntityFrameworkCore.MySql\Elsa.Quartz.EntityFrameworkCore.MySql.csproj" Condition=" '$(TargetFramework)' != 'net9.0' " />
<ProjectReference Include="..\..\modules\Elsa.Retention\Elsa.Retention.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Secrets.Api\Elsa.Secrets.Api.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Secrets.Persistence.EntityFrameworkCore.PostgreSql\Elsa.Secrets.Persistence.EntityFrameworkCore.PostgreSql.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Secrets.Persistence.EntityFrameworkCore.Sqlite\Elsa.Secrets.Persistence.EntityFrameworkCore.Sqlite.csproj" />

16
src/apps/Elsa.Server.Web/Program.cs
View file

@ -27,6 +27,8 @@ using Elsa.MongoDb.Modules.Management;
using Elsa.MongoDb.Modules.Runtime;
using Elsa.MongoDb.Modules.Tenants;
using Elsa.OpenTelemetry.Middleware;
using Elsa.Retention.Extensions;
using Elsa.Retention.Models;
using Elsa.Secrets.Extensions;
using Elsa.Secrets.Management.Tasks;
using Elsa.Secrets.Persistence;
@ -36,6 +38,7 @@ using Elsa.Server.Web.Filters;
using Elsa.Server.Web.Messages;
using Elsa.Tenants.AspNetCore;
using Elsa.Tenants.Extensions;
using Elsa.Workflows;
using Elsa.Workflows.Api;
using Elsa.Workflows.LogPersistence;
using Elsa.Workflows.Management;
@ -513,6 +516,19 @@ services
.UseSecretsScripting()
;
}
elsa.UseRetention(r =>
{
r.SweepInterval = TimeSpan.FromHours(5);
r.AddDeletePolicy("Delete all finished workflows", sp =>
{
var filter = new RetentionWorkflowInstanceFilter
{
WorkflowStatus = WorkflowStatus.Finished
};
return filter;
});
});
if (useMultitenancy)
{

2
src/modules/Elsa.Common/Features/MultitenancyFeature.cs
View file

@ -2,7 +2,6 @@ using Elsa.Common.Multitenancy;
using Elsa.Common.Multitenancy.EventHandlers;
using Elsa.Common.Multitenancy.HostedServices;
using Elsa.Common.RecurringTasks;
using Elsa.Extensions;
using Elsa.Features.Abstractions;
using Elsa.Features.Services;
using Microsoft.Extensions.DependencyInjection;
@ -52,7 +51,6 @@ public class MultitenancyFeature(IModule module) : FeatureBase(module)
.AddSingleton<ITenantActivatedEvent, RunStartupTasks>()
.AddSingleton<RecurringTaskScheduleManager>()
.AddSingleton<TenantEventsManager>()
.AddStartupTask<ConfigureRecurringTasksScheduleStartupTask>()
.AddScoped<DefaultTenantsProvider>()
.AddScoped<DefaultTenantResolver>()
.AddScoped<ITaskExecutor, TaskExecutor>()

4
src/modules/Elsa.Common/Multitenancy/EventHandlers/StartRecurringTasks.cs
View file

@ -7,7 +7,7 @@ namespace Elsa.Common.Multitenancy.EventHandlers;
public class StartRecurringTasks(RecurringTaskScheduleManager scheduleManager, ILogger<StartRecurringTasks> logger) : ITenantActivatedEvent, ITenantDeactivatedEvent
{
private readonly ICollection<ScheduledTimer> _scheduledTimers = new List<ScheduledTimer>();
private CancellationTokenSource _cancellationTokenSource = default!;
private CancellationTokenSource _cancellationTokenSource = null!;
public async Task TenantActivatedAsync(TenantActivatedEventArgs args)
{
@ -16,7 +16,7 @@ public class StartRecurringTasks(RecurringTaskScheduleManager scheduleManager, I
var tenantScope = args.TenantScope;
var tasks = tenantScope.ServiceProvider.GetServices<IRecurringTask>().ToList();
var taskExecutor = tenantScope.ServiceProvider.GetRequiredService<ITaskExecutor>();
foreach (var task in tasks)
{
var schedule = scheduleManager.GetScheduleFor(task.GetType());

26
src/modules/Elsa.Common/RecurringTasks/ConfigureRecurringTasksScheduleStartupTask.cs
View file

@ -1,26 +0,0 @@
using Cronos;
using JetBrains.Annotations;
using Microsoft.Extensions.Options;
namespace Elsa.Common.RecurringTasks;
[UsedImplicitly]
public class ConfigureRecurringTasksScheduleStartupTask(IOptions<RecurringTaskOptions> options, ISystemClock systemClock, RecurringTaskScheduleManager recurringTaskScheduleManager) : IStartupTask
{
public Task ExecuteAsync(CancellationToken cancellationToken)
{
foreach(var entry in options.Value.Schedule.ScheduledTasks)
{
var taskType = entry.Key;
var intervalExpression = entry.Value;
var schedule = intervalExpression.Type switch
{
IntervalExpressionType.Cron => (ISchedule)new CronSchedule(systemClock, CronExpression.Parse(intervalExpression.Expression)),
IntervalExpressionType.Interval => new IntervalSchedule(TimeSpan.Parse(intervalExpression.Expression)),
_ => throw new NotSupportedException($"Interval expression type '{intervalExpression.Type}' is not supported.")
};
recurringTaskScheduleManager.ConfigureScheduledTask(taskType, schedule);
}
return Task.CompletedTask;
}
}

34
src/modules/Elsa.Common/RecurringTasks/RecurringTaskScheduleManager.cs
View file

@ -1,21 +1,31 @@
using Cronos;
using Microsoft.Extensions.Options;
namespace Elsa.Common.RecurringTasks;
public class RecurringTaskScheduleManager
public class RecurringTaskScheduleManager(IOptions<RecurringTaskOptions> options, ISystemClock systemClock)
{
public IDictionary<Type, ISchedule> ScheduledTasks { get; set; } = new Dictionary<Type, ISchedule>();
public void ConfigureScheduledTask<T>(ISchedule schedule) where T : IRecurringTask
{
ConfigureScheduledTask(typeof(T), schedule);
}
public void ConfigureScheduledTask(Type recurringTaskType, ISchedule schedule)
{
ScheduledTasks[recurringTaskType] = schedule;
}
public ISchedule GetScheduleFor(Type taskType)
{
return ScheduledTasks.TryGetValue(taskType, out var schedule) ? schedule : new IntervalSchedule(TimeSpan.FromMinutes(1));
if (!ScheduledTasks.TryGetValue(taskType, out var schedule))
{
var intervalExpression = options.Value.Schedule.ScheduledTasks.TryGetValue(taskType, out var expr) ? expr : null;
schedule = intervalExpression != null ? CreateSchedule(intervalExpression) : new IntervalSchedule(TimeSpan.FromMinutes(1));
ScheduledTasks[taskType] = schedule;
}
return schedule;
}
private ISchedule CreateSchedule(IntervalExpression intervalExpression)
{
return intervalExpression.Type switch
{
IntervalExpressionType.Cron => new CronSchedule(systemClock, CronExpression.Parse(intervalExpression.Expression)),
IntervalExpressionType.Interval => new IntervalSchedule(TimeSpan.Parse(intervalExpression.Expression)),
_ => throw new NotSupportedException($"Interval expression type '{intervalExpression.Type}' is not supported.")
};
}
}