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

Refactor DefaultTriggerScheduler and handle past StartAt triggers (#5463)

Browse source 
The `DefaultTriggerScheduler` received a significant refactoring, with the injection of the `ISystemClock` service and a modification of several method calls. Additionally, a check is included to avoid scheduling `StartAt` triggers if their execution time is in the past. For these triggers, an information message is logged and scheduling is skipped.
This commit is contained in:
Sipke Schoorstra 2024-05-29 10:34:44 +02:00 committed by GitHub
parent 4393d95505
commit 411151d748
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 34 additions and 32 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

31
src/modules/Elsa.Scheduling/Activities/StartAt.cs
View file

@ -28,7 +28,7 @@ public class StartAt : Trigger
public StartAt(Input<DateTimeOffset> dateTime, [CallerFilePath] string? source = default, [CallerLineNumber] int? line = default) : this(source, line) => DateTime = dateTime;
/// <inheritdoc />
public StartAt(Func<ExpressionExecutionContext, DateTimeOffset> dateTime, [CallerFilePath] string? source = default, [CallerLineNumber] int? line = default)
public StartAt(Func<ExpressionExecutionContext, DateTimeOffset> dateTime, [CallerFilePath] string? source = default, [CallerLineNumber] int? line = default)
: this(new Input<DateTimeOffset>(dateTime), source, line)
{
}
@ -41,29 +41,30 @@ public class StartAt : Trigger
}
/// <inheritdoc />
public StartAt(Func<ValueTask<DateTimeOffset>> dateTime, [CallerFilePath] string? source = default, [CallerLineNumber] int? line = default)
public StartAt(Func<ValueTask<DateTimeOffset>> dateTime, [CallerFilePath] string? source = default, [CallerLineNumber] int? line = default)
: this(new Input<DateTimeOffset>(dateTime), source, line)
{
}
/// <inheritdoc />
public StartAt(Func<DateTimeOffset> dateTime, [CallerFilePath] string? source = default, [CallerLineNumber] int? line = default)
public StartAt(Func<DateTimeOffset> dateTime, [CallerFilePath] string? source = default, [CallerLineNumber] int? line = default)
: this(new Input<DateTimeOffset>(dateTime), source, line)
{
}
/// <inheritdoc />
public StartAt(DateTimeOffset dateTime, [CallerFilePath] string? source = default, [CallerLineNumber] int? line = default) : this(source, line) =>
public StartAt(DateTimeOffset dateTime, [CallerFilePath] string? source = default, [CallerLineNumber] int? line = default) : this(source, line) =>
DateTime = new Input<DateTimeOffset>(dateTime);
/// <inheritdoc />
public StartAt(Variable<DateTimeOffset> dateTime, [CallerFilePath] string? source = default, [CallerLineNumber] int? line = default) : this(source, line) =>
public StartAt(Variable<DateTimeOffset> dateTime, [CallerFilePath] string? source = default, [CallerLineNumber] int? line = default) : this(source, line) =>
DateTime = new Input<DateTimeOffset>(dateTime);
/// <summary>
/// The timestamp at which the workflow should be triggered.
/// </summary>
[Input] public Input<DateTimeOffset> DateTime { get; set; } = default!;
[Input]
public Input<DateTimeOffset> DateTime { get; set; } = default!;
/// <inheritdoc />
protected override object GetTriggerPayload(TriggerIndexingContext context)
@ -73,27 +74,29 @@ public class StartAt : Trigger
}
/// <inheritdoc />
protected override void Execute(ActivityExecutionContext context)
protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
{
// If external input was received, it means this activity got triggered and does not need to create a bookmark.
if (context.TryGetWorkflowInput<DateTimeOffset>(InputKey, out _))
if (context.IsTriggerOfWorkflow())
{
await context.CompleteActivityAsync();
return;
// No external input received, so create a bookmark.
}
var executeAt = context.ExpressionExecutionContext.Get(DateTime);
var clock = context.ExpressionExecutionContext.GetRequiredService<ISystemClock>();
var now = clock.UtcNow;
var logger = context.GetRequiredService<ILogger<StartAt>>();
context.JournalData.Add("Executed At", now);
if (executeAt <= now)
{
logger.LogDebug("Scheduled trigger time lies in the past ('{Delta}'). Skipping scheduling", now - executeAt);
context.JournalData.Add("Executed At", now);
logger.LogDebug("Scheduled trigger time lies in the past ('{Delta}'). Completing immediately", now - executeAt);
await context.CompleteActivityAsync();
return;
}
var payload = new StartAtPayload(executeAt);
context.CreateBookmark(payload);
}

35
src/modules/Elsa.Scheduling/Services/DefaultTriggerScheduler.cs
View file

@ -1,3 +1,4 @@
using Elsa.Common.Contracts;
using Elsa.Common.Models;
using Elsa.Extensions;
using Elsa.Scheduling.Activities;
@ -12,20 +13,9 @@ namespace Elsa.Scheduling.Services;
/// <summary>
/// A default implementation of <see cref="ITriggerScheduler"/> that schedules triggers using <see cref="IWorkflowScheduler"/>.
/// </summary>
public class DefaultTriggerScheduler : ITriggerScheduler
public class DefaultTriggerScheduler(IWorkflowScheduler workflowScheduler, ISystemClock systemClock, ILogger<DefaultTriggerScheduler> logger)
: ITriggerScheduler
{
private readonly IWorkflowScheduler _workflowScheduler;
private readonly ILogger<DefaultTriggerScheduler> _logger;
/// <summary>
/// Initializes a new instance of the <see cref="DefaultTriggerScheduler"/> class.
/// </summary>
public DefaultTriggerScheduler(IWorkflowScheduler workflowScheduler, ILogger<DefaultTriggerScheduler> logger)
{
_workflowScheduler = workflowScheduler;
_logger = logger;
}
/// <inheritdoc />
public async Task ScheduleAsync(IEnumerable<StoredTrigger> triggers, CancellationToken cancellationToken = default)
{
@ -34,6 +24,7 @@ public class DefaultTriggerScheduler : ITriggerScheduler
var timerTriggers = triggerList.Filter<Activities.Timer>();
var startAtTriggers = triggerList.Filter<StartAt>();
var cronTriggers = triggerList.Filter<Cron>();
var now = systemClock.UtcNow;
// Schedule each Timer trigger.
foreach (var trigger in timerTriggers)
@ -47,13 +38,21 @@ public class DefaultTriggerScheduler : ITriggerScheduler
TriggerActivityId = trigger.ActivityId,
Input = input
};
await _workflowScheduler.ScheduleRecurringAsync(trigger.Id, request, startAt, interval, cancellationToken);
await workflowScheduler.ScheduleRecurringAsync(trigger.Id, request, startAt, interval, cancellationToken);
}
// Schedule each StartAt trigger.
foreach (var trigger in startAtTriggers)
{
var executeAt = trigger.GetPayload<StartAtPayload>().ExecuteAt;
// If the trigger is in the past, log info and skip scheduling.
if (executeAt < now)
{
logger.LogInformation("StartAt trigger is in the past. TriggerId: {TriggerId}. ExecuteAt: {ExecuteAt}. Skipping scheduling", trigger.Id, executeAt);
continue;
}
var input = new { ExecuteAt = executeAt }.ToDictionary();
var request = new DispatchWorkflowDefinitionRequest
{
@ -63,7 +62,7 @@ public class DefaultTriggerScheduler : ITriggerScheduler
Input = input
};
await _workflowScheduler.ScheduleAtAsync(trigger.Id, request, executeAt, cancellationToken);
await workflowScheduler.ScheduleAtAsync(trigger.Id, request, executeAt, cancellationToken);
}
// Schedule each Cron trigger.
@ -81,11 +80,11 @@ public class DefaultTriggerScheduler : ITriggerScheduler
};
try
{
await _workflowScheduler.ScheduleCronAsync(trigger.Id, request, cronExpression, cancellationToken);
await workflowScheduler.ScheduleCronAsync(trigger.Id, request, cronExpression, cancellationToken);
}
catch (FormatException ex)
{
_logger.LogWarning($"Cron expression format error: {ex.Message}. CronExpression: {cronExpression}");
logger.LogWarning($"Cron expression format error: {ex.Message}. CronExpression: {cronExpression}");
}
}
}
@ -109,6 +108,6 @@ public class DefaultTriggerScheduler : ITriggerScheduler
// Unschedule each trigger.
foreach (var trigger in filteredTriggers)
await _workflowScheduler.UnscheduleAsync(trigger.Id, cancellationToken);
await workflowScheduler.UnscheduleAsync(trigger.Id, cancellationToken);
}
}