fix(scheduling): purge orphan Delay/Timer/Cron/StartAt bookmarks on startup (#8161)
* fix(scheduling): purge orphan Delay/Timer/Cron/StartAt bookmarks on startup Reconcile stored scheduling bookmarks against the workflow-instance store during CreateSchedulesStartupTask so missing and finished instances are skipped and deleted instead of being re-scheduled on every rebuild. Co-authored-by: Sipke Schoorstra <sipkeschoorstra@outlook.com> * fix(scheduling): add missing WorkflowStatus using in bookmark reconciler Co-authored-by: Sipke Schoorstra <sipkeschoorstra@outlook.com> * fix(scheduling): revalidate orphan bookmarks immediately before purge Collect only bookmark IDs during paged rebuild, reload those rows, and re-run classification so a candidate whose instance became Running is not deleted. Still purge missing/blank/terminal bookmarks only. Co-authored-by: Sipke Schoorstra <sipkeschoorstra@outlook.com> * fix(scheduling): schedule revived bookmarks and bound reconcile batches Revalidation now schedules bookmarks whose instance became Running, chunks reload/classify/delete by StartupSchedulePageSize, and skips reconcile when IWorkflowInstanceStore or IBookmarkManager is absent. Co-authored-by: Sipke Schoorstra <sipkeschoorstra@outlook.com> --------- Co-authored-by: Cursor Agent <cursoragent@cursor.com>
This commit is contained in:
Sipke Schoorstra
GitHub
parent
7e22a12081
commit
2138f0997b
|
|
@ -0,0 +1,82 @@
|
|||
using Elsa.Workflows;
|
||||
using Elsa.Workflows.Management;
|
||||
using Elsa.Workflows.Management.Filters;
|
||||
using Elsa.Workflows.Runtime;
|
||||
using Elsa.Workflows.Runtime.Entities;
|
||||
using Elsa.Workflows.Runtime.Filters;
|
||||
|
||||
namespace Elsa.Scheduling.Services;
|
||||
|
||||
/// <summary>
|
||||
/// Classifies Delay/Timer/Cron/StartAt stored bookmarks against the workflow-instance store so startup
|
||||
/// schedule rebuild can skip and purge bookmarks whose instance is missing or finished.
|
||||
/// </summary>
|
||||
public class SchedulingBookmarkReconciler(IWorkflowInstanceStore workflowInstanceStore, IBookmarkManager? bookmarkManager)
|
||||
{
|
||||
/// <summary>
|
||||
/// Splits bookmarks into those that may be scheduled and those whose instance is missing or terminal.
|
||||
/// </summary>
|
||||
public async Task<SchedulingBookmarkClassification> ClassifyAsync(IEnumerable<StoredBookmark> bookmarks, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var bookmarkList = bookmarks as IReadOnlyList<StoredBookmark> ?? bookmarks.ToList();
|
||||
|
||||
if (bookmarkList.Count == 0)
|
||||
return new SchedulingBookmarkClassification([], []);
|
||||
|
||||
var instanceIds = bookmarkList
|
||||
.Select(x => x.WorkflowInstanceId)
|
||||
.Where(x => !string.IsNullOrWhiteSpace(x))
|
||||
.Distinct(StringComparer.Ordinal)
|
||||
.ToList();
|
||||
|
||||
var liveInstanceIds = instanceIds.Count == 0
|
||||
? new HashSet<string>(StringComparer.Ordinal)
|
||||
: (await workflowInstanceStore.FindManyIdsAsync(new WorkflowInstanceFilter
|
||||
{
|
||||
Ids = instanceIds,
|
||||
WorkflowStatus = WorkflowStatus.Running
|
||||
}, cancellationToken)).ToHashSet(StringComparer.Ordinal);
|
||||
|
||||
var schedulable = new List<StoredBookmark>(bookmarkList.Count);
|
||||
var orphans = new List<StoredBookmark>();
|
||||
|
||||
foreach (var bookmark in bookmarkList)
|
||||
{
|
||||
if (!string.IsNullOrWhiteSpace(bookmark.WorkflowInstanceId) && liveInstanceIds.Contains(bookmark.WorkflowInstanceId))
|
||||
schedulable.Add(bookmark);
|
||||
else
|
||||
orphans.Add(bookmark);
|
||||
}
|
||||
|
||||
return new SchedulingBookmarkClassification(schedulable, orphans);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Revalidates candidates against the workflow-instance store, then deletes bookmarks that are still orphans.
|
||||
/// </summary>
|
||||
public async Task PurgeAsync(IEnumerable<StoredBookmark> candidateBookmarks, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (bookmarkManager == null)
|
||||
return;
|
||||
|
||||
var remainingOrphans = (await ClassifyAsync(candidateBookmarks, cancellationToken)).Orphans;
|
||||
var bookmarkIds = remainingOrphans
|
||||
.Select(x => x.Id)
|
||||
.Where(x => !string.IsNullOrWhiteSpace(x))
|
||||
.Distinct(StringComparer.Ordinal)
|
||||
.ToList();
|
||||
|
||||
if (bookmarkIds.Count == 0)
|
||||
return;
|
||||
|
||||
await bookmarkManager.DeleteManyAsync(new BookmarkFilter
|
||||
{
|
||||
BookmarkIds = bookmarkIds
|
||||
}, cancellationToken);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The schedulable vs orphan split for a batch of scheduling bookmarks.
|
||||
/// </summary>
|
||||
public record SchedulingBookmarkClassification(IReadOnlyList<StoredBookmark> Schedulable, IReadOnlyList<StoredBookmark> Orphans);
|
||||
|
|
@ -2,6 +2,8 @@ using Elsa.Common;
|
|||
using Elsa.Common.Multitenancy;
|
||||
using Elsa.Common.Models;
|
||||
using Elsa.Scheduling.Options;
|
||||
using Elsa.Scheduling.Services;
|
||||
using Elsa.Workflows.Management;
|
||||
using Elsa.Workflows.Runtime;
|
||||
using Elsa.Workflows.Runtime.Filters;
|
||||
using Elsa.Workflows.Runtime.Tasks;
|
||||
|
|
@ -12,6 +14,7 @@ namespace Elsa.Scheduling.StartupTasks;
|
|||
|
||||
/// <summary>
|
||||
/// Enqueues schedule creation when using the default scheduler, which doesn't have its own persistence layer like Quartz or Hangfire.
|
||||
/// Scheduling bookmarks whose workflow instance is missing or finished are skipped and purged so startup does not rehydrate dead work.
|
||||
/// </summary>
|
||||
[TaskDependency(typeof(PopulateRegistriesStartupTask))]
|
||||
public class CreateSchedulesStartupTask(IServiceProvider serviceProvider, IOptions<SchedulingOptions> options) : IStartupTask
|
||||
|
|
@ -32,6 +35,10 @@ public class CreateSchedulesStartupTask(IServiceProvider serviceProvider, IOptio
|
|||
var bookmarkStore = serviceProvider.GetRequiredService<IBookmarkStore>();
|
||||
var triggerScheduler = serviceProvider.GetRequiredService<ITriggerScheduler>();
|
||||
var bookmarkScheduler = serviceProvider.GetRequiredService<IBookmarkScheduler>();
|
||||
var workflowInstanceStore = serviceProvider.GetService<IWorkflowInstanceStore>();
|
||||
var bookmarkReconciler = workflowInstanceStore == null
|
||||
? null
|
||||
: new SchedulingBookmarkReconciler(workflowInstanceStore, serviceProvider.GetService<IBookmarkManager>());
|
||||
var pageSize = Math.Max(1, options.Value.StartupSchedulePageSize);
|
||||
var stimulusNames = new[]
|
||||
{
|
||||
|
|
@ -47,7 +54,7 @@ public class CreateSchedulesStartupTask(IServiceProvider serviceProvider, IOptio
|
|||
};
|
||||
|
||||
await ScheduleTriggersAsync(triggerStore, triggerScheduler, triggerFilter, pageSize, cancellationToken);
|
||||
await ScheduleBookmarksAsync(bookmarkStore, bookmarkScheduler, bookmarkFilter, pageSize, cancellationToken);
|
||||
await ScheduleBookmarksAsync(bookmarkStore, bookmarkScheduler, bookmarkReconciler, bookmarkFilter, pageSize, cancellationToken);
|
||||
}
|
||||
|
||||
private static async Task ScheduleTriggersAsync(ITriggerStore triggerStore, ITriggerScheduler triggerScheduler, TriggerFilter triggerFilter, int pageSize, CancellationToken cancellationToken)
|
||||
|
|
@ -71,9 +78,16 @@ public class CreateSchedulesStartupTask(IServiceProvider serviceProvider, IOptio
|
|||
}
|
||||
}
|
||||
|
||||
private static async Task ScheduleBookmarksAsync(IBookmarkStore bookmarkStore, IBookmarkScheduler bookmarkScheduler, BookmarkFilter bookmarkFilter, int pageSize, CancellationToken cancellationToken)
|
||||
private static async Task ScheduleBookmarksAsync(
|
||||
IBookmarkStore bookmarkStore,
|
||||
IBookmarkScheduler bookmarkScheduler,
|
||||
SchedulingBookmarkReconciler? bookmarkReconciler,
|
||||
BookmarkFilter bookmarkFilter,
|
||||
int pageSize,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var pageArgs = PageArgs.FromRange(0, pageSize);
|
||||
var orphanBookmarkIds = new HashSet<string>(StringComparer.Ordinal);
|
||||
|
||||
while (true)
|
||||
{
|
||||
|
|
@ -82,7 +96,18 @@ public class CreateSchedulesStartupTask(IServiceProvider serviceProvider, IOptio
|
|||
if (page.Items.Count == 0)
|
||||
break;
|
||||
|
||||
await bookmarkScheduler.ScheduleAsync(page.Items, cancellationToken);
|
||||
if (bookmarkReconciler == null)
|
||||
{
|
||||
await bookmarkScheduler.ScheduleAsync(page.Items, cancellationToken);
|
||||
}
|
||||
else
|
||||
{
|
||||
var classification = await bookmarkReconciler.ClassifyAsync(page.Items, cancellationToken);
|
||||
orphanBookmarkIds.UnionWith(classification.Orphans.Select(x => x.Id).Where(id => !string.IsNullOrWhiteSpace(id)));
|
||||
|
||||
if (classification.Schedulable.Count > 0)
|
||||
await bookmarkScheduler.ScheduleAsync(classification.Schedulable, cancellationToken);
|
||||
}
|
||||
|
||||
var nextOffset = pageArgs.Offset.GetValueOrDefault() + page.Items.Count;
|
||||
if (nextOffset >= page.TotalCount)
|
||||
|
|
@ -90,5 +115,22 @@ public class CreateSchedulesStartupTask(IServiceProvider serviceProvider, IOptio
|
|||
|
||||
pageArgs = pageArgs.Next();
|
||||
}
|
||||
|
||||
if (bookmarkReconciler == null || orphanBookmarkIds.Count == 0)
|
||||
return;
|
||||
|
||||
foreach (var orphanIdBatch in orphanBookmarkIds.Chunk(pageSize))
|
||||
{
|
||||
var candidates = await bookmarkStore.FindManyAsync(new BookmarkFilter
|
||||
{
|
||||
BookmarkIds = orphanIdBatch.ToList()
|
||||
}, cancellationToken);
|
||||
var classification = await bookmarkReconciler.ClassifyAsync(candidates, cancellationToken);
|
||||
|
||||
if (classification.Schedulable.Count > 0)
|
||||
await bookmarkScheduler.ScheduleAsync(classification.Schedulable, cancellationToken);
|
||||
|
||||
await bookmarkReconciler.PurgeAsync(classification.Orphans, cancellationToken);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,134 @@
|
|||
using Elsa.Common.Services;
|
||||
using Elsa.Mediator.Contracts;
|
||||
using Elsa.Scheduling.Services;
|
||||
using Elsa.Workflows;
|
||||
using Elsa.Workflows.Management.Entities;
|
||||
using Elsa.Workflows.Management.Stores;
|
||||
using Elsa.Workflows.Runtime;
|
||||
using Elsa.Workflows.Runtime.Entities;
|
||||
using Elsa.Workflows.Runtime.Filters;
|
||||
using Elsa.Workflows.Runtime.Stores;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using NSubstitute;
|
||||
|
||||
namespace Elsa.Scheduling.UnitTests.Services;
|
||||
|
||||
public class SchedulingBookmarkReconcilerTests
|
||||
{
|
||||
[Fact]
|
||||
public async Task ClassifyAsync_SkipsMissingBlankAndFinishedInstances()
|
||||
{
|
||||
var instanceStore = CreateInstanceStore(
|
||||
Instance("running", WorkflowStatus.Running, WorkflowSubStatus.Executing),
|
||||
Instance("suspended", WorkflowStatus.Running, WorkflowSubStatus.Suspended),
|
||||
Instance("interrupted", WorkflowStatus.Running, WorkflowSubStatus.Interrupted),
|
||||
Instance("pending", WorkflowStatus.Running, WorkflowSubStatus.Pending),
|
||||
Instance("finished", WorkflowStatus.Finished, WorkflowSubStatus.Finished),
|
||||
Instance("cancelled", WorkflowStatus.Finished, WorkflowSubStatus.Cancelled),
|
||||
Instance("faulted", WorkflowStatus.Finished, WorkflowSubStatus.Faulted));
|
||||
var reconciler = new SchedulingBookmarkReconciler(instanceStore, Substitute.For<IBookmarkManager>());
|
||||
|
||||
var result = await reconciler.ClassifyAsync(
|
||||
[
|
||||
Bookmark("b-missing", "missing"),
|
||||
Bookmark("b-blank", ""),
|
||||
Bookmark("b-finished", "finished"),
|
||||
Bookmark("b-cancelled", "cancelled"),
|
||||
Bookmark("b-faulted", "faulted"),
|
||||
Bookmark("b-running", "running"),
|
||||
Bookmark("b-suspended", "suspended"),
|
||||
Bookmark("b-interrupted", "interrupted"),
|
||||
Bookmark("b-pending", "pending")
|
||||
]);
|
||||
|
||||
Assert.Equal(["b-running", "b-suspended", "b-interrupted", "b-pending"], result.Schedulable.Select(x => x.Id));
|
||||
Assert.Equal(["b-missing", "b-blank", "b-finished", "b-cancelled", "b-faulted"], result.Orphans.Select(x => x.Id));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ClassifyAsync_KeepsPastDueSuspendedDelayBookmarks()
|
||||
{
|
||||
var instanceStore = CreateInstanceStore(Instance("suspended", WorkflowStatus.Running, WorkflowSubStatus.Suspended));
|
||||
var reconciler = new SchedulingBookmarkReconciler(instanceStore, Substitute.For<IBookmarkManager>());
|
||||
|
||||
var result = await reconciler.ClassifyAsync([Bookmark("past-due-delay", "suspended")]);
|
||||
|
||||
Assert.Equal(["past-due-delay"], result.Schedulable.Select(x => x.Id));
|
||||
Assert.Empty(result.Orphans);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task PurgeAsync_DeletesOrphanBookmarksFromTheStore()
|
||||
{
|
||||
var bookmarkStore = new MemoryBookmarkStore(new MemoryStore<StoredBookmark>());
|
||||
await bookmarkStore.SaveManyAsync(
|
||||
[
|
||||
Bookmark("orphan-1", "missing"),
|
||||
Bookmark("orphan-2", "finished"),
|
||||
Bookmark("live", "suspended")
|
||||
], CancellationToken.None);
|
||||
var bookmarkManager = new DefaultBookmarkManager(bookmarkStore, Substitute.For<INotificationSender>(), NullLogger<DefaultBookmarkManager>.Instance);
|
||||
var reconciler = new SchedulingBookmarkReconciler(CreateInstanceStore(), bookmarkManager);
|
||||
|
||||
await reconciler.PurgeAsync([Bookmark("orphan-1", "missing"), Bookmark("orphan-2", "finished")]);
|
||||
|
||||
var remaining = (await bookmarkStore.FindManyAsync(new BookmarkFilter())).Select(x => x.Id).ToList();
|
||||
Assert.Equal(["live"], remaining);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task PurgeAsync_DoesNotDeleteWhenThereAreNoOrphans()
|
||||
{
|
||||
var bookmarkManager = Substitute.For<IBookmarkManager>();
|
||||
var reconciler = new SchedulingBookmarkReconciler(CreateInstanceStore(), bookmarkManager);
|
||||
|
||||
await reconciler.PurgeAsync([]);
|
||||
|
||||
await bookmarkManager.DidNotReceive().DeleteManyAsync(Arg.Any<BookmarkFilter>(), Arg.Any<CancellationToken>());
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task PurgeAsync_DoesNotDeleteWhenInstanceBecameRunning()
|
||||
{
|
||||
var bookmarkManager = Substitute.For<IBookmarkManager>();
|
||||
var reconciler = new SchedulingBookmarkReconciler(
|
||||
CreateInstanceStore(Instance("revived", WorkflowStatus.Running, WorkflowSubStatus.Suspended)),
|
||||
bookmarkManager);
|
||||
|
||||
await reconciler.PurgeAsync([Bookmark("revived-bookmark", "revived")]);
|
||||
|
||||
await bookmarkManager.DidNotReceive().DeleteManyAsync(Arg.Any<BookmarkFilter>(), Arg.Any<CancellationToken>());
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task PurgeAsync_DoesNotDeleteWhenBookmarkManagerIsMissing()
|
||||
{
|
||||
var reconciler = new SchedulingBookmarkReconciler(CreateInstanceStore(), bookmarkManager: null);
|
||||
|
||||
await reconciler.PurgeAsync([Bookmark("orphan-1", "missing")]);
|
||||
}
|
||||
|
||||
private static MemoryWorkflowInstanceStore CreateInstanceStore(params WorkflowInstance[] instances)
|
||||
{
|
||||
var store = new MemoryStore<WorkflowInstance>();
|
||||
store.AddMany(instances, x => x.Id);
|
||||
return new MemoryWorkflowInstanceStore(store);
|
||||
}
|
||||
|
||||
private static WorkflowInstance Instance(string id, WorkflowStatus status, WorkflowSubStatus subStatus) => new()
|
||||
{
|
||||
Id = id,
|
||||
DefinitionId = "definition",
|
||||
DefinitionVersionId = "version",
|
||||
Status = status,
|
||||
SubStatus = subStatus
|
||||
};
|
||||
|
||||
private static StoredBookmark Bookmark(string id, string workflowInstanceId) => new()
|
||||
{
|
||||
Id = id,
|
||||
Hash = "hash",
|
||||
Name = SchedulingStimulusNames.Delay,
|
||||
WorkflowInstanceId = workflowInstanceId
|
||||
};
|
||||
}
|
||||
|
|
@ -3,6 +3,8 @@ using Elsa.Common.Multitenancy;
|
|||
using Elsa.Common.Models;
|
||||
using Elsa.Scheduling.Options;
|
||||
using Elsa.Scheduling.StartupTasks;
|
||||
using Elsa.Workflows.Management;
|
||||
using Elsa.Workflows.Management.Filters;
|
||||
using Elsa.Workflows.Runtime;
|
||||
using Elsa.Workflows.Runtime.Entities;
|
||||
using Elsa.Workflows.Runtime.Filters;
|
||||
|
|
@ -25,6 +27,8 @@ public class CreateSchedulesStartupTaskTests
|
|||
private readonly IBookmarkStore _bookmarkStore = Substitute.For<IBookmarkStore>();
|
||||
private readonly ITriggerScheduler _triggerScheduler = Substitute.For<ITriggerScheduler>();
|
||||
private readonly IBookmarkScheduler _bookmarkScheduler = Substitute.For<IBookmarkScheduler>();
|
||||
private readonly IWorkflowInstanceStore _workflowInstanceStore = Substitute.For<IWorkflowInstanceStore>();
|
||||
private readonly IBookmarkManager _bookmarkManager = Substitute.For<IBookmarkManager>();
|
||||
private readonly SchedulingOptions _options = new() { StartupSchedulePageSize = 1000 };
|
||||
|
||||
public CreateSchedulesStartupTaskTests()
|
||||
|
|
@ -33,6 +37,8 @@ public class CreateSchedulesStartupTaskTests
|
|||
.Returns(new Page<StoredTrigger>(_triggers, _triggers.Length));
|
||||
_bookmarkStore.FindManyAsync(Arg.Any<BookmarkFilter>(), Arg.Any<PageArgs>(), Arg.Any<CancellationToken>())
|
||||
.Returns(new Page<StoredBookmark>(_bookmarks, _bookmarks.Length));
|
||||
_workflowInstanceStore.FindManyIdsAsync(Arg.Any<WorkflowInstanceFilter>(), Arg.Any<CancellationToken>())
|
||||
.Returns(call => RunningIds(call.Arg<WorkflowInstanceFilter>()));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
|
|
@ -101,14 +107,197 @@ public class CreateSchedulesStartupTaskTests
|
|||
await _bookmarkScheduler.Received(1).ScheduleAsync(Arg.Is<IEnumerable<StoredBookmark>>(x => x.SequenceEqual(secondBookmarkPage)), Arg.Any<CancellationToken>());
|
||||
}
|
||||
|
||||
private ServiceProvider CreateServiceProvider(Action<IServiceCollection>? configureServices = null)
|
||||
[Fact]
|
||||
public async Task ExecuteAsync_SkipsAndPurgesOrphanSchedulingBookmarks()
|
||||
{
|
||||
var missingInstance = Bookmark("missing-bookmark", "missing-instance");
|
||||
var finishedInstance = Bookmark("finished-bookmark", "finished-instance");
|
||||
var cancelledInstance = Bookmark("cancelled-bookmark", "cancelled-instance");
|
||||
var faultedInstance = Bookmark("faulted-bookmark", "faulted-instance");
|
||||
var emptyInstanceId = Bookmark("empty-instance-bookmark", "");
|
||||
var suspendedInstance = Bookmark("suspended-bookmark", "suspended-instance");
|
||||
var bookmarks = new[]
|
||||
{
|
||||
missingInstance, finishedInstance, cancelledInstance, faultedInstance, emptyInstanceId, suspendedInstance
|
||||
};
|
||||
BookmarkFilter? purgedFilter = null;
|
||||
_bookmarkStore.FindManyAsync(Arg.Any<BookmarkFilter>(), Arg.Any<PageArgs>(), Arg.Any<CancellationToken>())
|
||||
.Returns(new Page<StoredBookmark>(bookmarks, bookmarks.Length));
|
||||
StubBookmarkReload(bookmarks);
|
||||
_workflowInstanceStore.FindManyIdsAsync(Arg.Any<WorkflowInstanceFilter>(), Arg.Any<CancellationToken>())
|
||||
.Returns(["suspended-instance"]);
|
||||
_bookmarkManager.DeleteManyAsync(Arg.Do<BookmarkFilter>(x => purgedFilter = x), Arg.Any<CancellationToken>())
|
||||
.Returns(5);
|
||||
var task = new CreateSchedulesStartupTask(CreateServiceProvider(), OptionsFactory.Create(_options));
|
||||
|
||||
await task.ExecuteAsync(CancellationToken.None);
|
||||
|
||||
await _bookmarkScheduler.Received(1).ScheduleAsync(
|
||||
Arg.Is<IEnumerable<StoredBookmark>>(x => x.SequenceEqual(new[] { suspendedInstance })),
|
||||
Arg.Any<CancellationToken>());
|
||||
await _bookmarkManager.Received(1).DeleteManyAsync(Arg.Any<BookmarkFilter>(), Arg.Any<CancellationToken>());
|
||||
Assert.NotNull(purgedFilter);
|
||||
Assert.Equal(
|
||||
new[]
|
||||
{
|
||||
"cancelled-bookmark", "empty-instance-bookmark", "faulted-bookmark", "finished-bookmark", "missing-bookmark"
|
||||
},
|
||||
purgedFilter.BookmarkIds?.OrderBy(x => x, StringComparer.Ordinal).ToArray());
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ExecuteAsync_SchedulesOrphanWhoseInstanceBecomesRunningBeforePurge()
|
||||
{
|
||||
var revived = Bookmark("revived-bookmark", "revived-instance");
|
||||
var stillMissing = Bookmark("missing-bookmark", "missing-instance");
|
||||
var bookmarks = new[] { revived, stillMissing };
|
||||
BookmarkFilter? purgedFilter = null;
|
||||
_bookmarkStore.FindManyAsync(Arg.Any<BookmarkFilter>(), Arg.Any<PageArgs>(), Arg.Any<CancellationToken>())
|
||||
.Returns(new Page<StoredBookmark>(bookmarks, bookmarks.Length));
|
||||
StubBookmarkReload(bookmarks);
|
||||
_workflowInstanceStore.FindManyIdsAsync(Arg.Any<WorkflowInstanceFilter>(), Arg.Any<CancellationToken>())
|
||||
.Returns(_ => Array.Empty<string>(), _ => new[] { "revived-instance" });
|
||||
_bookmarkManager.DeleteManyAsync(Arg.Do<BookmarkFilter>(x => purgedFilter = x), Arg.Any<CancellationToken>())
|
||||
.Returns(1);
|
||||
var task = new CreateSchedulesStartupTask(CreateServiceProvider(), OptionsFactory.Create(_options));
|
||||
|
||||
await task.ExecuteAsync(CancellationToken.None);
|
||||
|
||||
await _bookmarkScheduler.Received(1).ScheduleAsync(
|
||||
Arg.Is<IEnumerable<StoredBookmark>>(x => x.SequenceEqual(new[] { revived })),
|
||||
Arg.Any<CancellationToken>());
|
||||
await _bookmarkManager.Received(1).DeleteManyAsync(Arg.Any<BookmarkFilter>(), Arg.Any<CancellationToken>());
|
||||
Assert.NotNull(purgedFilter);
|
||||
Assert.Equal(["missing-bookmark"], purgedFilter.BookmarkIds);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ExecuteAsync_ReconcilesOrphansInConfiguredPages()
|
||||
{
|
||||
var firstOrphan = Bookmark("orphan-1", "missing-1");
|
||||
var secondOrphan = Bookmark("orphan-2", "missing-2");
|
||||
var bookmarks = new[] { firstOrphan, secondOrphan };
|
||||
var purgedFilters = new List<BookmarkFilter>();
|
||||
var reloadedIdBatches = new List<ICollection<string>>();
|
||||
_options.StartupSchedulePageSize = 1;
|
||||
_bookmarkStore.FindManyAsync(Arg.Any<BookmarkFilter>(), Arg.Is<PageArgs>(x => x.Offset == 0 && x.Limit == 1), Arg.Any<CancellationToken>())
|
||||
.Returns(new Page<StoredBookmark>([firstOrphan], 2));
|
||||
_bookmarkStore.FindManyAsync(Arg.Any<BookmarkFilter>(), Arg.Is<PageArgs>(x => x.Offset == 1 && x.Limit == 1), Arg.Any<CancellationToken>())
|
||||
.Returns(new Page<StoredBookmark>([secondOrphan], 2));
|
||||
_bookmarkStore.FindManyAsync(Arg.Any<BookmarkFilter>(), Arg.Any<CancellationToken>())
|
||||
.Returns(call =>
|
||||
{
|
||||
var ids = call.Arg<BookmarkFilter>().BookmarkIds ?? [];
|
||||
reloadedIdBatches.Add(ids.ToList());
|
||||
return bookmarks.Where(x => ids.Contains(x.Id));
|
||||
});
|
||||
_workflowInstanceStore.FindManyIdsAsync(Arg.Any<WorkflowInstanceFilter>(), Arg.Any<CancellationToken>())
|
||||
.Returns(Array.Empty<string>());
|
||||
_bookmarkManager.DeleteManyAsync(Arg.Do<BookmarkFilter>(x => purgedFilters.Add(x)), Arg.Any<CancellationToken>())
|
||||
.Returns(1);
|
||||
var task = new CreateSchedulesStartupTask(CreateServiceProvider(), OptionsFactory.Create(_options));
|
||||
|
||||
await task.ExecuteAsync(CancellationToken.None);
|
||||
|
||||
await _bookmarkScheduler.DidNotReceive().ScheduleAsync(Arg.Any<IEnumerable<StoredBookmark>>(), Arg.Any<CancellationToken>());
|
||||
Assert.Equal(2, reloadedIdBatches.Count);
|
||||
Assert.All(reloadedIdBatches, batch => Assert.Single(batch));
|
||||
Assert.Equal(2, purgedFilters.Count);
|
||||
Assert.All(purgedFilters, filter => Assert.Single(filter.BookmarkIds!));
|
||||
Assert.Equal(["orphan-1", "orphan-2"], purgedFilters.SelectMany(x => x.BookmarkIds!).OrderBy(x => x, StringComparer.Ordinal));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ExecuteAsync_WithoutWorkflowInstanceStore_SchedulesAllBookmarks()
|
||||
{
|
||||
var missingInstance = Bookmark("missing-bookmark", "missing-instance");
|
||||
var bookmarks = new[] { missingInstance, _bookmarks[0] };
|
||||
_bookmarkStore.FindManyAsync(Arg.Any<BookmarkFilter>(), Arg.Any<PageArgs>(), Arg.Any<CancellationToken>())
|
||||
.Returns(new Page<StoredBookmark>(bookmarks, bookmarks.Length));
|
||||
var task = new CreateSchedulesStartupTask(CreateServiceProvider(registerReconcileServices: false), OptionsFactory.Create(_options));
|
||||
|
||||
await task.ExecuteAsync(CancellationToken.None);
|
||||
|
||||
await _bookmarkScheduler.Received(1).ScheduleAsync(Arg.Is<IEnumerable<StoredBookmark>>(x => x.SequenceEqual(bookmarks)), Arg.Any<CancellationToken>());
|
||||
await _bookmarkManager.DidNotReceive().DeleteManyAsync(Arg.Any<BookmarkFilter>(), Arg.Any<CancellationToken>());
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ExecuteAsync_WithoutBookmarkManager_SkipsOrphansButDoesNotPurge()
|
||||
{
|
||||
var missingInstance = Bookmark("missing-bookmark", "missing-instance");
|
||||
var suspendedInstance = Bookmark("suspended-bookmark", "suspended-instance");
|
||||
var bookmarks = new[] { missingInstance, suspendedInstance };
|
||||
_bookmarkStore.FindManyAsync(Arg.Any<BookmarkFilter>(), Arg.Any<PageArgs>(), Arg.Any<CancellationToken>())
|
||||
.Returns(new Page<StoredBookmark>(bookmarks, bookmarks.Length));
|
||||
StubBookmarkReload(bookmarks);
|
||||
_workflowInstanceStore.FindManyIdsAsync(Arg.Any<WorkflowInstanceFilter>(), Arg.Any<CancellationToken>())
|
||||
.Returns(["suspended-instance"]);
|
||||
var task = new CreateSchedulesStartupTask(
|
||||
CreateServiceProvider(services => services.AddSingleton(_workflowInstanceStore), registerReconcileServices: false),
|
||||
OptionsFactory.Create(_options));
|
||||
|
||||
await task.ExecuteAsync(CancellationToken.None);
|
||||
|
||||
await _bookmarkScheduler.Received(1).ScheduleAsync(
|
||||
Arg.Is<IEnumerable<StoredBookmark>>(x => x.SequenceEqual(new[] { suspendedInstance })),
|
||||
Arg.Any<CancellationToken>());
|
||||
await _bookmarkManager.DidNotReceive().DeleteManyAsync(Arg.Any<BookmarkFilter>(), Arg.Any<CancellationToken>());
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ExecuteAsync_DoesNotPurgeWhenEverySchedulingBookmarkIsLive()
|
||||
{
|
||||
var task = new CreateSchedulesStartupTask(CreateServiceProvider(), OptionsFactory.Create(_options));
|
||||
|
||||
await task.ExecuteAsync(CancellationToken.None);
|
||||
|
||||
await _bookmarkManager.DidNotReceive().DeleteManyAsync(Arg.Any<BookmarkFilter>(), Arg.Any<CancellationToken>());
|
||||
}
|
||||
|
||||
private ServiceProvider CreateServiceProvider(Action<IServiceCollection>? configureServices = null, bool registerReconcileServices = true)
|
||||
{
|
||||
var services = new ServiceCollection();
|
||||
services.AddSingleton(_triggerStore);
|
||||
services.AddSingleton(_bookmarkStore);
|
||||
services.AddSingleton(_triggerScheduler);
|
||||
services.AddSingleton(_bookmarkScheduler);
|
||||
|
||||
if (registerReconcileServices)
|
||||
{
|
||||
services.AddSingleton(_workflowInstanceStore);
|
||||
services.AddSingleton(_bookmarkManager);
|
||||
}
|
||||
|
||||
configureServices?.Invoke(services);
|
||||
return services.BuildServiceProvider();
|
||||
}
|
||||
|
||||
private void StubBookmarkReload(IEnumerable<StoredBookmark> bookmarks)
|
||||
{
|
||||
var bookmarkList = bookmarks.ToList();
|
||||
_bookmarkStore.FindManyAsync(Arg.Any<BookmarkFilter>(), Arg.Any<CancellationToken>())
|
||||
.Returns(call =>
|
||||
{
|
||||
var ids = call.Arg<BookmarkFilter>().BookmarkIds;
|
||||
if (ids == null)
|
||||
return bookmarkList.AsEnumerable();
|
||||
|
||||
var idSet = ids.ToHashSet(StringComparer.Ordinal);
|
||||
return bookmarkList.Where(x => idSet.Contains(x.Id));
|
||||
});
|
||||
}
|
||||
|
||||
private static IEnumerable<string> RunningIds(WorkflowInstanceFilter filter)
|
||||
{
|
||||
return (filter.Ids ?? []).Where(id => !string.IsNullOrWhiteSpace(id));
|
||||
}
|
||||
|
||||
private static StoredBookmark Bookmark(string id, string workflowInstanceId) => new()
|
||||
{
|
||||
Id = id,
|
||||
Hash = "hash",
|
||||
Name = SchedulingStimulusNames.Delay,
|
||||
WorkflowInstanceId = workflowInstanceId
|
||||
};
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue