From bc95a11fa1944bd9b597d0e8cacc1198472fc1ce Mon Sep 17 00:00:00 2001 From: KrzysztofPajak Date: Wed, 14 Oct 2026 20:40:36 +0200 Subject: [PATCH 1/2] Keep scheduled task runs from overlapping across instances A claimed run now holds a lease (LeasedUntilUtc) that the running instance renews, so another instance cannot start the same task while it still runs. Tasks pinned with LeasedByMachineName can also match Application:InstanceName, since container hostnames change on every deployment. The loop sleeps for the exact remaining time instead of whole minutes, and dates read back from LiteDB as local time are compared in UTC. --- docs/installation.md | 15 ++ .../ScheduleTasks/IScheduleTaskService.cs | 15 +- .../Configuration/AppConfig.cs | 8 + .../BackgroundServices/ScheduleTaskService.cs | 35 ++++- .../ScheduleTaskServiceTests.cs | 139 ++++++++++++++++- .../BackgroundServiceTaskTests.cs | 140 +++++++++++++++--- .../Grand.Web.Admin/App_Data/appsettings.json | 4 + .../Infrastructure/BackgroundServiceTask.cs | 102 +++++++++++-- .../Grand.Web.Store/App_Data/appsettings.json | 4 + .../App_Data/appsettings.json | 4 + src/Web/Grand.Web/App_Data/appsettings.json | 4 + 11 files changed, 421 insertions(+), 49 deletions(-) diff --git a/docs/installation.md b/docs/installation.md index 3e3c5d9207..1c0a9cc953 100644 --- a/docs/installation.md +++ b/docs/installation.md @@ -47,6 +47,21 @@ instance serves its own copy of the sitemap until the sitemap task runs again, a the push notification settings are saved again. A file left behind in `wwwroot` keeps being served after it is deleted in the admin, because the shop only deletes on the media path. Thumbnails are regenerated on the media path. +Scheduled tasks need no setup: every instance runs the scheduler, and each due run is claimed in the database by +exactly one of them. The claiming instance holds a lease on the task while it runs (renewed every minute, five +minutes long), so a run that takes longer than the task's interval is not started again elsewhere, and a run whose +instance died is picked up once the lease runs out. + +To keep a task on particular instances, set **Leased by machine name** on the task in the admin. In a container the +machine name is the hostname, which changes on every deployment, so give those instances a stable name with +`Application:InstanceName` and enter that name instead. Several instances may share it; one of them runs each due +task: + +```bash +docker run -d --name grandnode2-worker --link mongodb:mongo \ + -e Application__InstanceName=worker grandnode/grandnode2 +``` + ## Open locally in an IDE Extract the source code package downloaded from the Releases tab to a folder (or clone the repository), and open diff --git a/src/Business/Grand.Business.Core/Interfaces/System/ScheduleTasks/IScheduleTaskService.cs b/src/Business/Grand.Business.Core/Interfaces/System/ScheduleTasks/IScheduleTaskService.cs index 8c78591b98..8356080650 100644 --- a/src/Business/Grand.Business.Core/Interfaces/System/ScheduleTasks/IScheduleTaskService.cs +++ b/src/Business/Grand.Business.Core/Interfaces/System/ScheduleTasks/IScheduleTaskService.cs @@ -40,13 +40,26 @@ public interface IScheduleTaskService /// Atomically claims a single run of the task, so that only one application instance /// executes it. The claim succeeds only if LastStartUtc still has the value the caller /// read (compare-and-set) - a concurrent claim by another instance makes it fail. + /// It also fails while LeasedUntilUtc of a previous run is still in the future, + /// so a run that outlasts the interval is not started a second time elsewhere. /// /// Task identifier /// LastStartUtc value read before the claim /// New LastStartUtc value (start of the claimed run) + /// Until when the claimed run blocks other instances, unless renewed /// Unique identifier of the claiming application instance /// true if this instance won the claim and should execute the task - Task TryClaimTaskRun(string taskId, DateTime? expectedLastStartUtc, DateTime runStartUtc, string instanceId); + Task TryClaimTaskRun(string taskId, DateTime? expectedLastStartUtc, DateTime runStartUtc, + DateTime leasedUntilUtc, string instanceId); + + /// + /// Extends the lease of a running task. Does nothing when the instance no longer holds + /// the lease (released, or taken over by another instance after it ran out). + /// + /// Task identifier + /// Unique identifier of the application instance running the task + /// New end of the lease + Task RenewTaskLease(string taskId, string instanceId, DateTime leasedUntilUtc); /// /// Delete the task diff --git a/src/Core/Grand.Infrastructure/Configuration/AppConfig.cs b/src/Core/Grand.Infrastructure/Configuration/AppConfig.cs index 1267750921..3573ac8d8e 100644 --- a/src/Core/Grand.Infrastructure/Configuration/AppConfig.cs +++ b/src/Core/Grand.Infrastructure/Configuration/AppConfig.cs @@ -46,4 +46,12 @@ public class AppConfig /// instance at the same shared volume. A relative path is resolved against the content root. /// public string MediaPath { get; set; } + + /// + /// Gets or sets the name this instance answers to when a scheduled task is pinned to a machine + /// ("Leased by machine name"). Empty uses only the machine name, which in a container is the + /// hostname and changes on every deployment. Several instances may share the name - one of them + /// runs each due task. + /// + public string InstanceName { get; set; } } \ No newline at end of file diff --git a/src/Modules/Grand.Module.ScheduledTasks/BackgroundServices/ScheduleTaskService.cs b/src/Modules/Grand.Module.ScheduledTasks/BackgroundServices/ScheduleTaskService.cs index 8a5ad31c04..7ce0d320b0 100644 --- a/src/Modules/Grand.Module.ScheduledTasks/BackgroundServices/ScheduleTaskService.cs +++ b/src/Modules/Grand.Module.ScheduledTasks/BackgroundServices/ScheduleTaskService.cs @@ -79,26 +79,47 @@ public virtual async Task UpdateTask(ScheduleTask task) } /// - /// Atomically claims a single run of the task (compare-and-set on LastStartUtc) + /// Atomically claims a single run of the task (compare-and-set on LastStartUtc, + /// refused while the lease of a previous run has not run out) /// public virtual async Task TryClaimTaskRun(string taskId, DateTime? expectedLastStartUtc, - DateTime runStartUtc, string instanceId) + DateTime runStartUtc, DateTime leasedUntilUtc, string instanceId) { //truncate to milliseconds so the value survives the database round-trip unchanged - runStartUtc = new DateTime(runStartUtc.Ticks - runStartUtc.Ticks % TimeSpan.TicksPerMillisecond, - DateTimeKind.Utc); + runStartUtc = TruncateToMilliseconds(runStartUtc); //conditional update - matches only when no other instance has claimed the run //in the meantime; on MongoDB a single update is atomic, so exactly one instance wins await _taskRepository.UpdateOneAsync( - x => x.Id == taskId && x.LastStartUtc == expectedLastStartUtc, + x => x.Id == taskId && x.LastStartUtc == expectedLastStartUtc && + (x.LeasedUntilUtc == null || x.LeasedUntilUtc < runStartUtc), UpdateBuilder.Create() .Set(x => x.LastStartUtc, runStartUtc) + .Set(x => x.LeasedUntilUtc, TruncateToMilliseconds(leasedUntilUtc)) .Set(x => x.LeasedByInstance, instanceId)); - //UpdateOneAsync does not report whether the filter matched - read back the winner + //UpdateOneAsync does not report whether the filter matched - read back the winner. + //LiteDB returns dates as local time, so compare in UTC - otherwise no claim ever + //succeeds on a host outside the UTC time zone var task = await GetTaskById(taskId); - return task != null && task.LeasedByInstance == instanceId && task.LastStartUtc == runStartUtc; + return task != null && task.LeasedByInstance == instanceId && + task.LastStartUtc?.ToUniversalTime() == runStartUtc; + } + + /// + /// Extends the lease of a run, only while the given instance still holds it + /// + public virtual async Task RenewTaskLease(string taskId, string instanceId, DateTime leasedUntilUtc) + { + await _taskRepository.UpdateOneAsync( + x => x.Id == taskId && x.LeasedByInstance == instanceId && x.LeasedUntilUtc != null, + UpdateBuilder.Create() + .Set(x => x.LeasedUntilUtc, TruncateToMilliseconds(leasedUntilUtc))); + } + + private static DateTime TruncateToMilliseconds(DateTime value) + { + return new DateTime(value.Ticks - value.Ticks % TimeSpan.TicksPerMillisecond, DateTimeKind.Utc); } /// diff --git a/src/Tests/Grand.Modules.Tests/Services/BackgroundService/ScheduleTaskServiceTests.cs b/src/Tests/Grand.Modules.Tests/Services/BackgroundService/ScheduleTaskServiceTests.cs index 7663112f34..cf4ffe9aef 100644 --- a/src/Tests/Grand.Modules.Tests/Services/BackgroundService/ScheduleTaskServiceTests.cs +++ b/src/Tests/Grand.Modules.Tests/Services/BackgroundService/ScheduleTaskServiceTests.cs @@ -1,3 +1,4 @@ +using Grand.Data.Tests.LiteDb; using Grand.Data.Tests.MongoDb; using Grand.Domain.Tasks; using Grand.Module.ScheduledTasks.BackgroundServices; @@ -11,6 +12,8 @@ public class ScheduleTaskServiceTests private MongoDBRepositoryTest _repository; private ScheduleTaskService _service; + private static readonly DateTime LeasedUntilUtc = DateTime.UtcNow.AddMinutes(5); + [TestInitialize] public void Init() { @@ -18,13 +21,16 @@ public void Init() _service = new ScheduleTaskService(_repository); } - private async Task InsertTask(DateTime? lastStartUtc = null) + private async Task InsertTask(DateTime? lastStartUtc = null, DateTime? leasedUntilUtc = null, + string leasedByInstance = null) { var task = new ScheduleTask { ScheduleTaskName = "TestTask", Enabled = true, TimeInterval = 1, - LastStartUtc = lastStartUtc + LastStartUtc = lastStartUtc, + LeasedUntilUtc = leasedUntilUtc, + LeasedByInstance = leasedByInstance }; await _service.InsertTask(task); //read back so LastStartUtc has the value as stored in the database (ms precision) @@ -37,7 +43,7 @@ public async Task TryClaimTaskRun_FirstClaim_ReturnsTrueAndSetsLease() var task = await InsertTask(); var runStartUtc = DateTime.UtcNow; - var claimed = await _service.TryClaimTaskRun(task.Id, task.LastStartUtc, runStartUtc, "instance-a"); + var claimed = await _service.TryClaimTaskRun(task.Id, task.LastStartUtc, runStartUtc, LeasedUntilUtc, "instance-a"); Assert.IsTrue(claimed); var updated = await _service.GetTaskById(task.Id); @@ -51,9 +57,9 @@ public async Task TryClaimTaskRun_StaleExpectedValue_ReturnsFalseAndKeepsWinner( var task = await InsertTask(DateTime.UtcNow.AddMinutes(-10)); //instance A claims the run - var claimedByA = await _service.TryClaimTaskRun(task.Id, task.LastStartUtc, DateTime.UtcNow, "instance-a"); + var claimedByA = await _service.TryClaimTaskRun(task.Id, task.LastStartUtc, DateTime.UtcNow, LeasedUntilUtc, "instance-a"); //instance B still holds the old LastStartUtc value - its claim must fail - var claimedByB = await _service.TryClaimTaskRun(task.Id, task.LastStartUtc, DateTime.UtcNow, "instance-b"); + var claimedByB = await _service.TryClaimTaskRun(task.Id, task.LastStartUtc, DateTime.UtcNow, LeasedUntilUtc, "instance-b"); Assert.IsTrue(claimedByA); Assert.IsFalse(claimedByB); @@ -67,8 +73,8 @@ public async Task TryClaimTaskRun_ConcurrentClaims_ExactlyOneWins() var task = await InsertTask(DateTime.UtcNow.AddMinutes(-10)); var claims = await Task.WhenAll( - _service.TryClaimTaskRun(task.Id, task.LastStartUtc, DateTime.UtcNow, "instance-1"), - _service.TryClaimTaskRun(task.Id, task.LastStartUtc, DateTime.UtcNow, "instance-2")); + _service.TryClaimTaskRun(task.Id, task.LastStartUtc, DateTime.UtcNow, LeasedUntilUtc, "instance-1"), + _service.TryClaimTaskRun(task.Id, task.LastStartUtc, DateTime.UtcNow, LeasedUntilUtc, "instance-2")); Assert.AreEqual(1, claims.Count(x => x)); } @@ -76,8 +82,125 @@ public async Task TryClaimTaskRun_ConcurrentClaims_ExactlyOneWins() [TestMethod] public async Task TryClaimTaskRun_NonExistingTask_ReturnsFalse() { - var claimed = await _service.TryClaimTaskRun("000000000000000000000000", null, DateTime.UtcNow, "instance-a"); + var claimed = await _service.TryClaimTaskRun("000000000000000000000000", null, DateTime.UtcNow, LeasedUntilUtc, "instance-a"); Assert.IsFalse(claimed); } + + [TestMethod] + public async Task TryClaimTaskRun_PreviousRunStillLeased_ReturnsFalseAndKeepsRunningInstance() + { + //a long run on instance A outlasts the interval - its lease is still in the future + var task = await InsertTask(DateTime.UtcNow.AddMinutes(-10), DateTime.UtcNow.AddMinutes(3), "instance-a"); + + var claimed = await _service.TryClaimTaskRun(task.Id, task.LastStartUtc, DateTime.UtcNow, LeasedUntilUtc, + "instance-b"); + + Assert.IsFalse(claimed); + var updated = await _service.GetTaskById(task.Id); + Assert.AreEqual("instance-a", updated.LeasedByInstance); + } + + [TestMethod] + public async Task TryClaimTaskRun_PreviousLeaseRanOut_ReturnsTrue() + { + //instance A died during the run and stopped renewing its lease + var task = await InsertTask(DateTime.UtcNow.AddMinutes(-10), DateTime.UtcNow.AddMinutes(-4), "instance-a"); + + var claimed = await _service.TryClaimTaskRun(task.Id, task.LastStartUtc, DateTime.UtcNow, LeasedUntilUtc, + "instance-b"); + + Assert.IsTrue(claimed); + var updated = await _service.GetTaskById(task.Id); + Assert.AreEqual("instance-b", updated.LeasedByInstance); + Assert.IsNotNull(updated.LeasedUntilUtc); + } + + [TestMethod] + public async Task RenewTaskLease_OwnLease_ExtendsIt() + { + var task = await InsertTask(DateTime.UtcNow, DateTime.UtcNow.AddMinutes(1), "instance-a"); + + await _service.RenewTaskLease(task.Id, "instance-a", DateTime.UtcNow.AddMinutes(5)); + + var updated = await _service.GetTaskById(task.Id); + Assert.IsTrue(updated.LeasedUntilUtc > DateTime.UtcNow.AddMinutes(4)); + } + + [TestMethod] + public async Task RenewTaskLease_LeaseTakenOverByAnotherInstance_DoesNothing() + { + var task = await InsertTask(DateTime.UtcNow, DateTime.UtcNow.AddMinutes(5), "instance-b"); + + await _service.RenewTaskLease(task.Id, "instance-a", DateTime.UtcNow.AddMinutes(30)); + + var updated = await _service.GetTaskById(task.Id); + Assert.AreEqual("instance-b", updated.LeasedByInstance); + Assert.IsTrue(updated.LeasedUntilUtc < DateTime.UtcNow.AddMinutes(6)); + } + + [TestMethod] + public async Task RenewTaskLease_LeaseReleased_DoesNotRecreateIt() + { + //the run has finished and released the lease before a late renewal arrived + var task = await InsertTask(DateTime.UtcNow, null, "instance-a"); + + await _service.RenewTaskLease(task.Id, "instance-a", DateTime.UtcNow.AddMinutes(5)); + + var updated = await _service.GetTaskById(task.Id); + Assert.IsNull(updated.LeasedUntilUtc); + } +} + +/// +/// The lease filter compares nullable dates - checked on LiteDB too, which translates +/// the same expression with its own LINQ provider +/// +[TestClass] +public class ScheduleTaskServiceLiteDbTests +{ + private ScheduleTaskService _service; + + [TestInitialize] + public void Init() + { + _service = new ScheduleTaskService(new LiteDBRepositoryMock()); + } + + private async Task InsertTask(DateTime? leasedUntilUtc) + { + var task = new ScheduleTask { + ScheduleTaskName = "TestTask", + Enabled = true, + TimeInterval = 1, + LastStartUtc = DateTime.UtcNow.AddMinutes(-10), + LeasedUntilUtc = leasedUntilUtc, + LeasedByInstance = "instance-a" + }; + await _service.InsertTask(task); + return await _service.GetTaskById(task.Id); + } + + [TestMethod] + public async Task TryClaimTaskRun_PreviousRunStillLeased_ReturnsFalse() + { + var task = await InsertTask(DateTime.UtcNow.AddMinutes(3)); + + var claimed = await _service.TryClaimTaskRun(task.Id, task.LastStartUtc, DateTime.UtcNow, + DateTime.UtcNow.AddMinutes(5), "instance-b"); + + Assert.IsFalse(claimed); + } + + [TestMethod] + public async Task TryClaimTaskRun_NoLeaseOrLeaseRanOut_ReturnsTrue() + { + var released = await InsertTask(null); + var expired = await InsertTask(DateTime.UtcNow.AddMinutes(-4)); + + Assert.IsTrue(await _service.TryClaimTaskRun(released.Id, released.LastStartUtc, DateTime.UtcNow, + DateTime.UtcNow.AddMinutes(5), "instance-b")); + Assert.IsTrue(await _service.TryClaimTaskRun(expired.Id, expired.LastStartUtc, DateTime.UtcNow, + DateTime.UtcNow.AddMinutes(5), "instance-b")); + } } diff --git a/src/Tests/Grand.Web.Common.Tests/Infrastructure/BackgroundServiceTaskTests.cs b/src/Tests/Grand.Web.Common.Tests/Infrastructure/BackgroundServiceTaskTests.cs index 824c8ced38..d7180993a6 100644 --- a/src/Tests/Grand.Web.Common.Tests/Infrastructure/BackgroundServiceTaskTests.cs +++ b/src/Tests/Grand.Web.Common.Tests/Infrastructure/BackgroundServiceTaskTests.cs @@ -59,7 +59,7 @@ public void Init() public async Task ExecuteAsync_ClaimWon_ExecutesTaskAndPersistsResult() { _scheduleTaskServiceMock.Setup(s => s.TryClaimTaskRun(_task.Id, It.IsAny(), - It.IsAny(), It.IsAny())) + It.IsAny(), It.IsAny(), It.IsAny())) .ReturnsAsync(true); var service = new BackgroundServiceTask(TaskName, _serviceProvider); @@ -75,12 +75,37 @@ await WaitFor(() => _scheduleTaskServiceMock.Invocations.Any(i => x.LastSuccessUtc != null && x.LastStartUtc != null)), Times.Once); } + [TestMethod] + public async Task ExecuteAsync_ClaimWon_LeasesTheRunAndReleasesItWhenDone() + { + DateTime? claimedRunStartUtc = null, claimedLeasedUntilUtc = null; + _scheduleTaskServiceMock.Setup(s => s.TryClaimTaskRun(_task.Id, It.IsAny(), + It.IsAny(), It.IsAny(), It.IsAny())) + .Callback((_, _, runStartUtc, leasedUntilUtc, _) => { + claimedRunStartUtc = runStartUtc; + claimedLeasedUntilUtc = leasedUntilUtc; + }) + .ReturnsAsync(true); + + var service = new BackgroundServiceTask(TaskName, _serviceProvider); + using var cts = new CancellationTokenSource(); + await service.StartAsync(cts.Token); + + await WaitFor(() => _scheduleTaskServiceMock.Invocations.Any(i => + i.Method.Name == nameof(IScheduleTaskService.UpdateTask))); + cts.Cancel(); + + Assert.AreEqual(LeaseDuration, claimedLeasedUntilUtc - claimedRunStartUtc); + _scheduleTaskServiceMock.Verify(s => s.UpdateTask(It.Is(x => + x.LeasedUntilUtc == null)), Times.Once); + } + [TestMethod] public async Task ExecuteAsync_ClaimWon_TaskSeesBackgroundWorkAndStoreContext() { _task.StoreId = "store-1"; _scheduleTaskServiceMock.Setup(s => s.TryClaimTaskRun(_task.Id, It.IsAny(), - It.IsAny(), It.IsAny())) + It.IsAny(), It.IsAny(), It.IsAny())) .ReturnsAsync(true); var storeContext = Mock.Of(); @@ -124,7 +149,7 @@ await WaitFor(() => _scheduleTaskServiceMock.Invocations.Any(i => public async Task ExecuteAsync_ClaimLost_DoesNotExecuteAndDoesNotPersist() { _scheduleTaskServiceMock.Setup(s => s.TryClaimTaskRun(_task.Id, It.IsAny(), - It.IsAny(), It.IsAny())) + It.IsAny(), It.IsAny(), It.IsAny())) .ReturnsAsync(false); var service = new BackgroundServiceTask(TaskName, _serviceProvider); @@ -157,7 +182,7 @@ await WaitFor(() => _scheduleTaskServiceMock.Invocations.Any(i => cts.Cancel(); _scheduleTaskServiceMock.Verify(s => s.TryClaimTaskRun(It.IsAny(), It.IsAny(), - It.IsAny(), It.IsAny()), Times.Never); + It.IsAny(), It.IsAny(), It.IsAny()), Times.Never); _scheduleTaskMock.Verify(t => t.Execute(), Times.Never); _scheduleTaskServiceMock.Verify(s => s.UpdateTask(It.IsAny()), Times.Never); } @@ -250,10 +275,10 @@ private static ScheduleTask NewTask(bool enabled = true, string leasedBy = null, [TestMethod] public void Decide_TaskNotSeededYet_ReturnsRetryInOneMinute() { - var decision = Decide(null, MachineName, UtcNow); + var decision = Decide(null, MachineName, null, UtcNow); Assert.AreEqual(ScheduleAction.Retry, decision.Action); - Assert.AreEqual(1, decision.DelayMinutes); + Assert.AreEqual(TimeSpan.FromMinutes(1), decision.Delay); } [TestMethod] @@ -261,10 +286,10 @@ public void Decide_TaskDisabled_ReturnsRetryAfterOwnInterval() { var task = NewTask(enabled: false, timeInterval: 15); - var decision = Decide(task, MachineName, UtcNow); + var decision = Decide(task, MachineName, null, UtcNow); Assert.AreEqual(ScheduleAction.Retry, decision.Action); - Assert.AreEqual(15, decision.DelayMinutes); + Assert.AreEqual(TimeSpan.FromMinutes(15), decision.Delay); } [TestMethod] @@ -272,10 +297,10 @@ public void Decide_TaskLeasedByAnotherMachine_ReturnsRetryAfterOwnInterval() { var task = NewTask(leasedBy: "other-machine", timeInterval: 15); - var decision = Decide(task, MachineName, UtcNow); + var decision = Decide(task, MachineName, null, UtcNow); Assert.AreEqual(ScheduleAction.Retry, decision.Action); - Assert.AreEqual(15, decision.DelayMinutes); + Assert.AreEqual(TimeSpan.FromMinutes(15), decision.Delay); } [TestMethod] @@ -283,7 +308,7 @@ public void Decide_TaskLeasedByOwnMachine_IsEligible() { var task = NewTask(leasedBy: MachineName); - var decision = Decide(task, MachineName, UtcNow); + var decision = Decide(task, MachineName, null, UtcNow); Assert.AreEqual(ScheduleAction.RunNow, decision.Action); } @@ -293,10 +318,10 @@ public void Decide_EnabledAndNeverRunBefore_ReturnsRunNow() { var task = NewTask(timeInterval: 30); - var decision = Decide(task, MachineName, UtcNow); + var decision = Decide(task, MachineName, null, UtcNow); Assert.AreEqual(ScheduleAction.RunNow, decision.Action); - Assert.AreEqual(30, decision.DelayMinutes); + Assert.AreEqual(TimeSpan.FromMinutes(30), decision.Delay); } [TestMethod] @@ -304,21 +329,100 @@ public void Decide_EnabledAndIntervalElapsed_ReturnsRunNow() { var task = NewTask(timeInterval: 30, lastStartUtc: UtcNow.AddMinutes(-31)); - var decision = Decide(task, MachineName, UtcNow); + var decision = Decide(task, MachineName, null, UtcNow); Assert.AreEqual(ScheduleAction.RunNow, decision.Action); - Assert.AreEqual(30, decision.DelayMinutes); + Assert.AreEqual(TimeSpan.FromMinutes(30), decision.Delay); } [TestMethod] - public void Decide_EnabledButNotDueYet_ReturnsWaitForRemainingTimeTruncatedToWholeMinutes() + public void Decide_EnabledButNotDueYet_ReturnsWaitForExactRemainingTime() { //10.5 minutes elapsed of a 30-minute interval -> 19.5 minutes remain var task = NewTask(timeInterval: 30, lastStartUtc: UtcNow.AddSeconds(-(10 * 60 + 30))); - var decision = Decide(task, MachineName, UtcNow); + var decision = Decide(task, MachineName, null, UtcNow); + + Assert.AreEqual(ScheduleAction.WaitForNextRun, decision.Action); + Assert.AreEqual(TimeSpan.FromSeconds(19 * 60 + 30), decision.Delay); + } + + [TestMethod] + public void Decide_LessThanAMinuteBeforeNextRun_WaitsTheRemainingSecondsNotZero() + { + //truncating to whole minutes gave 0 here - the loop then re-read the task without any delay + var task = NewTask(timeInterval: 30, lastStartUtc: UtcNow.AddSeconds(-(29 * 60 + 20))); + + var decision = Decide(task, MachineName, null, UtcNow); + + Assert.AreEqual(ScheduleAction.WaitForNextRun, decision.Action); + Assert.AreEqual(TimeSpan.FromSeconds(40), decision.Delay); + } + + [TestMethod] + public void Decide_PinnedToConfiguredInstanceName_IsEligible() + { + //container hostnames change on every deployment - the configured name does not + var task = NewTask(leasedBy: "worker"); + + var decision = Decide(task, "grandnode-7f9c4b-x2kq", "worker", UtcNow); + + Assert.AreEqual(ScheduleAction.RunNow, decision.Action); + } + + [TestMethod] + public void Decide_PinnedToMachineName_IgnoresCase() + { + var task = NewTask(leasedBy: "THIS-MACHINE"); + + var decision = Decide(task, MachineName, null, UtcNow); + + Assert.AreEqual(ScheduleAction.RunNow, decision.Action); + } + + [TestMethod] + public void Decide_PinnedToAnotherInstanceName_ReturnsRetry() + { + var task = NewTask(leasedBy: "worker", timeInterval: 15); + + var decision = Decide(task, MachineName, "web", UtcNow); + + Assert.AreEqual(ScheduleAction.Retry, decision.Action); + Assert.AreEqual(TimeSpan.FromMinutes(15), decision.Delay); + } + + [TestMethod] + public void Decide_DueButPreviousRunStillLeased_RetriesWhenLeaseRunsOut() + { + var task = NewTask(timeInterval: 1, lastStartUtc: UtcNow.AddMinutes(-3)); + task.LeasedUntilUtc = UtcNow.AddMinutes(2); + + var decision = Decide(task, MachineName, null, UtcNow); + + Assert.AreEqual(ScheduleAction.Retry, decision.Action); + Assert.AreEqual(TimeSpan.FromMinutes(2), decision.Delay); + } + + [TestMethod] + public void Decide_DueAndPreviousLeaseRanOut_ReturnsRunNow() + { + var task = NewTask(timeInterval: 1, lastStartUtc: UtcNow.AddMinutes(-10)); + task.LeasedUntilUtc = UtcNow.AddMinutes(-4); + + var decision = Decide(task, MachineName, null, UtcNow); + + Assert.AreEqual(ScheduleAction.RunNow, decision.Action); + } + + [TestMethod] + public void Decide_DatesReadAsLocalTime_AreComparedInUtc() + { + //LiteDB returns dates as local time - the same instants as the UTC ones + var task = NewTask(timeInterval: 30, lastStartUtc: UtcNow.AddMinutes(-10).ToLocalTime()); + + var decision = Decide(task, MachineName, null, UtcNow); Assert.AreEqual(ScheduleAction.WaitForNextRun, decision.Action); - Assert.AreEqual(19, decision.DelayMinutes); //truncated, same as the original code + Assert.AreEqual(TimeSpan.FromMinutes(20), decision.Delay); } } diff --git a/src/Web/Grand.Web.Admin/App_Data/appsettings.json b/src/Web/Grand.Web.Admin/App_Data/appsettings.json index 1abc580117..fec1c26f5d 100644 --- a/src/Web/Grand.Web.Admin/App_Data/appsettings.json +++ b/src/Web/Grand.Web.Admin/App_Data/appsettings.json @@ -8,6 +8,10 @@ //firebase-messaging-sw.js. Empty keeps them in wwwroot. When running more than one instance (pods), //set it to a shared volume mounted on every instance. A relative path is resolved against the application directory. "MediaPath": "", + //Name this instance answers to when a scheduled task is pinned to a machine ("Leased by machine name"). + //Empty uses only the machine name - in a container the hostname, which changes on every deployment. + //Several instances may share the name (e.g. Application__InstanceName=worker); one of them runs each due task. + "InstanceName": "", //Enable the session-based TempData provider "UseSessionStateTempDataProvider": false, //A value indicating whether SEO friendly URLs with multiple languages are enabled diff --git a/src/Web/Grand.Web.Common/Infrastructure/BackgroundServiceTask.cs b/src/Web/Grand.Web.Common/Infrastructure/BackgroundServiceTask.cs index 9dd5dc81ca..538356b7b3 100644 --- a/src/Web/Grand.Web.Common/Infrastructure/BackgroundServiceTask.cs +++ b/src/Web/Grand.Web.Common/Infrastructure/BackgroundServiceTask.cs @@ -2,6 +2,7 @@ using Grand.Business.Core.Interfaces.System.ScheduleTasks; using Grand.Domain.Tasks; using Grand.Infrastructure; +using Grand.Infrastructure.Configuration; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; @@ -25,17 +26,23 @@ public BackgroundServiceTask(string name, IServiceProvider serviceProvider) _serviceProvider = serviceProvider; } + //how long a claimed run blocks other instances; the running instance renews it while the + //task executes, so it only runs out when that instance dies without finishing the run + internal static readonly TimeSpan LeaseDuration = TimeSpan.FromMinutes(5); + internal static readonly TimeSpan LeaseRenewInterval = TimeSpan.FromMinutes(1); + internal enum ScheduleAction { - //task missing, disabled, or leased to another instance - all reversible, poll again later + //task missing, disabled, pinned to another instance, or still running elsewhere - + //all reversible, poll again later Retry, - //own lease (or none), enabled, and due - claim and execute + //enabled, due, and not running elsewhere - claim and execute RunNow, - //own lease, enabled, but not due yet - sleep until it is + //enabled, but not due yet - sleep until it is WaitForNextRun } - internal readonly record struct ScheduleDecision(ScheduleAction Action, int DelayMinutes); + internal readonly record struct ScheduleDecision(ScheduleAction Action, TimeSpan Delay); /// /// Pure decision of what the loop should do this iteration and how long to sleep @@ -43,33 +50,60 @@ internal enum ScheduleAction /// resolves to a delay-and-retry, never to "stop the loop" - so the reversible states /// (task not seeded yet, disabled, leased elsewhere) can never end the loop for good. /// - internal static ScheduleDecision Decide(ScheduleTask task, string machineName, DateTime utcNow) + /// Task as read from the database + /// Name of the machine (container hostname, pod name) + /// Configured Application:InstanceName, empty when not set + /// Current time + internal static ScheduleDecision Decide(ScheduleTask task, string machineName, string instanceName, + DateTime utcNow) { if (task == null) - return new ScheduleDecision(ScheduleAction.Retry, 1); + return new ScheduleDecision(ScheduleAction.Retry, TimeSpan.FromMinutes(1)); - var timeInterval = task.TimeInterval > 0 ? task.TimeInterval : 1; + var timeInterval = TimeSpan.FromMinutes(task.TimeInterval > 0 ? task.TimeInterval : 1); - var eligible = task.Enabled && - (string.IsNullOrEmpty(task.LeasedByMachineName) || machineName == task.LeasedByMachineName); - if (!eligible) + if (!task.Enabled || !IsPinnedToThisInstance(task.LeasedByMachineName, machineName, instanceName)) return new ScheduleDecision(ScheduleAction.Retry, timeInterval); + //LiteDB returns dates as local time - compared with utcNow, they must be UTC if (task.LastStartUtc.HasValue) { - var nextRunUtc = task.LastStartUtc.Value.AddMinutes(task.TimeInterval); + var nextRunUtc = task.LastStartUtc.Value.ToUniversalTime().AddMinutes(task.TimeInterval); + //the exact remaining time, not whole minutes - truncating it turned the last + //minute before every run into a delay-free loop of database reads if (utcNow < nextRunUtc) - return new ScheduleDecision(ScheduleAction.WaitForNextRun, (int)(nextRunUtc - utcNow).TotalMinutes); + return new ScheduleDecision(ScheduleAction.WaitForNextRun, nextRunUtc - utcNow); } + //the previous run is still executing on another instance - come back when its lease runs out + var leasedUntilUtc = task.LeasedUntilUtc?.ToUniversalTime(); + if (leasedUntilUtc > utcNow) + return new ScheduleDecision(ScheduleAction.Retry, leasedUntilUtc.Value - utcNow); + return new ScheduleDecision(ScheduleAction.RunNow, timeInterval); } + /// + /// A task pinned to a machine runs only on the instance with that name. Container hostnames + /// change on every deployment, so the configured instance name is accepted too - and it can be + /// shared by several instances, which then compete for the run like unpinned ones. + /// + private static bool IsPinnedToThisInstance(string leasedByMachineName, string machineName, string instanceName) + { + if (string.IsNullOrEmpty(leasedByMachineName)) + return true; + + return string.Equals(leasedByMachineName, machineName, StringComparison.OrdinalIgnoreCase) || + (!string.IsNullOrEmpty(instanceName) && + string.Equals(leasedByMachineName, instanceName, StringComparison.OrdinalIgnoreCase)); + } + protected override async Task ExecuteAsync(CancellationToken stoppingToken) { //resolved once from the root provider (not the per-iteration scope) so it also //survives to log failures raised before/around a scope, e.g. in the outer catch var logger = _serviceProvider.GetService>(); + var instanceName = _serviceProvider.GetService()?.InstanceName; while (!stoppingToken.IsCancellationRequested) try @@ -78,7 +112,7 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken) var serviceProvider = scope.ServiceProvider; var scheduleTaskService = serviceProvider.GetService(); var task = await scheduleTaskService.GetTaskByName(Name); - var decision = Decide(task, Environment.MachineName, DateTime.UtcNow); + var decision = Decide(task, Environment.MachineName, instanceName, DateTime.UtcNow); if (task == null) logger.LogInformation("Task {TaskName} is not exists in the database", Name); @@ -88,7 +122,7 @@ await RunTask(serviceProvider, scheduleTaskService, task, //every branch above falls through to here - the loop always sleeps and //retries, it never exits on its own (only cancellation ends it) - await Task.Delay(TimeSpan.FromMinutes(decision.DelayMinutes), stoppingToken); + await Task.Delay(decision.Delay, stoppingToken); } catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) { @@ -115,12 +149,16 @@ private async Task RunTask(IServiceProvider serviceProvider, IScheduleTaskServic //only one wins and executes the task (no duplicated e-mails etc.) var runStartUtc = DateTime.UtcNow; var claimed = await scheduleTaskService.TryClaimTaskRun(task.Id, task.LastStartUtc, - runStartUtc, InstanceId); + runStartUtc, runStartUtc.Add(LeaseDuration), InstanceId); if (claimed) { updateTask = true; task.LastStartUtc = runStartUtc; task.LeasedByInstance = InstanceId; + + //keeps other instances off the task while it runs, even past its next due time + using var leaseCts = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken); + var leaseRenewal = RenewLease(task.Id, logger, leaseCts.Token); try { //assign current customer (background task) / current store (from task). @@ -149,6 +187,14 @@ private async Task RunTask(IServiceProvider serviceProvider, IScheduleTaskServic "Error while running the \'{TaskScheduleTaskName}\' schedule task", task.ScheduleTaskName); } + finally + { + await leaseCts.CancelAsync(); + await leaseRenewal; + } + + //the run is over - the write below releases the lease for the next one + task.LeasedUntilUtc = null; } else { @@ -171,6 +217,32 @@ private async Task RunTask(IServiceProvider serviceProvider, IScheduleTaskServic await scheduleTaskService.UpdateTask(task); } + /// + /// Extends the lease of the claimed run until cancelled. Uses its own scope: it runs alongside + /// the task, which keeps using the iteration's scope. A failed renewal is only logged - the + /// lease is several renewals long, and the run itself must not fail because of it. + /// + private async Task RenewLease(string taskId, ILogger logger, CancellationToken cancellationToken) + { + while (!cancellationToken.IsCancellationRequested) + try + { + await Task.Delay(LeaseRenewInterval, cancellationToken); + + using var scope = _serviceProvider.CreateScope(); + var scheduleTaskService = scope.ServiceProvider.GetRequiredService(); + await scheduleTaskService.RenewTaskLease(taskId, InstanceId, DateTime.UtcNow.Add(LeaseDuration)); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + //the run has ended + } + catch (Exception exc) + { + logger.LogWarning(exc, "Could not renew the lease of task {TaskName}", Name); + } + } + private static async Task<(IStoreContext storeContext, IWorkContext workContext)> InitializeContext( IServiceProvider serviceProvider, ScheduleTask scheduleTask) { diff --git a/src/Web/Grand.Web.Store/App_Data/appsettings.json b/src/Web/Grand.Web.Store/App_Data/appsettings.json index be8f462ca2..0bfee7373b 100644 --- a/src/Web/Grand.Web.Store/App_Data/appsettings.json +++ b/src/Web/Grand.Web.Store/App_Data/appsettings.json @@ -8,6 +8,10 @@ //firebase-messaging-sw.js. Empty keeps them in wwwroot. When running more than one instance (pods), //set it to a shared volume mounted on every instance. A relative path is resolved against the application directory. "MediaPath": "", + //Name this instance answers to when a scheduled task is pinned to a machine ("Leased by machine name"). + //Empty uses only the machine name - in a container the hostname, which changes on every deployment. + //Several instances may share the name (e.g. Application__InstanceName=worker); one of them runs each due task. + "InstanceName": "", //Enable the session-based TempData provider "UseSessionStateTempDataProvider": false, //A value indicating whether SEO friendly URLs with multiple languages are enabled diff --git a/src/Web/Grand.Web.Vendor/App_Data/appsettings.json b/src/Web/Grand.Web.Vendor/App_Data/appsettings.json index 1abc580117..fec1c26f5d 100644 --- a/src/Web/Grand.Web.Vendor/App_Data/appsettings.json +++ b/src/Web/Grand.Web.Vendor/App_Data/appsettings.json @@ -8,6 +8,10 @@ //firebase-messaging-sw.js. Empty keeps them in wwwroot. When running more than one instance (pods), //set it to a shared volume mounted on every instance. A relative path is resolved against the application directory. "MediaPath": "", + //Name this instance answers to when a scheduled task is pinned to a machine ("Leased by machine name"). + //Empty uses only the machine name - in a container the hostname, which changes on every deployment. + //Several instances may share the name (e.g. Application__InstanceName=worker); one of them runs each due task. + "InstanceName": "", //Enable the session-based TempData provider "UseSessionStateTempDataProvider": false, //A value indicating whether SEO friendly URLs with multiple languages are enabled diff --git a/src/Web/Grand.Web/App_Data/appsettings.json b/src/Web/Grand.Web/App_Data/appsettings.json index caf0e98b4d..6b8000eeff 100644 --- a/src/Web/Grand.Web/App_Data/appsettings.json +++ b/src/Web/Grand.Web/App_Data/appsettings.json @@ -8,6 +8,10 @@ //firebase-messaging-sw.js. Empty keeps them in wwwroot. When running more than one instance (pods), //set it to a shared volume mounted on every instance. A relative path is resolved against the application directory. "MediaPath": "", + //Name this instance answers to when a scheduled task is pinned to a machine ("Leased by machine name"). + //Empty uses only the machine name - in a container the hostname, which changes on every deployment. + //Several instances may share the name (e.g. Application__InstanceName=worker); one of them runs each due task. + "InstanceName": "", //Enable the session-based TempData provider "UseSessionStateTempDataProvider": false, //A value indicating whether SEO friendly URLs with multiple languages are enabled From 5e7c327e770fae71eddffac827dbf471b4775ceb Mon Sep 17 00:00:00 2001 From: KrzysztofPajak Date: Thu, 15 Oct 2026 07:31:56 +0200 Subject: [PATCH 2/2] Release the scheduled task lease with a targeted update Write only the run result and release the lease, and only while this instance still holds it. The whole-entity write reverted admin edits made during the run and, after a lease that ran out, overwrote the claim of the instance that took the task over, letting a third one start. Keep the four-argument TryClaimTaskRun as an obsolete overload so callers built against 2.4 keep working. --- .../ScheduleTasks/IScheduleTaskService.cs | 24 +++++ .../BackgroundServices/ScheduleTaskService.cs | 21 +++++ .../ScheduleTaskServiceTests.cs | 87 +++++++++++++++++++ .../BackgroundServiceTaskTests.cs | 44 ++++++++-- .../Infrastructure/BackgroundServiceTask.cs | 25 ++---- 5 files changed, 177 insertions(+), 24 deletions(-) diff --git a/src/Business/Grand.Business.Core/Interfaces/System/ScheduleTasks/IScheduleTaskService.cs b/src/Business/Grand.Business.Core/Interfaces/System/ScheduleTasks/IScheduleTaskService.cs index 8356080650..b4dcfe02b2 100644 --- a/src/Business/Grand.Business.Core/Interfaces/System/ScheduleTasks/IScheduleTaskService.cs +++ b/src/Business/Grand.Business.Core/Interfaces/System/ScheduleTasks/IScheduleTaskService.cs @@ -52,6 +52,17 @@ public interface IScheduleTaskService Task TryClaimTaskRun(string taskId, DateTime? expectedLastStartUtc, DateTime runStartUtc, DateTime leasedUntilUtc, string instanceId); + /// + /// Atomically claims a single run of the task without a lease - the run blocks other + /// instances only until its interval elapses, as before leases were introduced. + /// + [Obsolete("Use the overload with leasedUntilUtc - a run claimed without a lease is started again elsewhere once its interval elapses, even if it is still running")] + Task TryClaimTaskRun(string taskId, DateTime? expectedLastStartUtc, DateTime runStartUtc, + string instanceId) + { + return TryClaimTaskRun(taskId, expectedLastStartUtc, runStartUtc, runStartUtc, instanceId); + } + /// /// Extends the lease of a running task. Does nothing when the instance no longer holds /// the lease (released, or taken over by another instance after it ran out). @@ -61,6 +72,19 @@ Task TryClaimTaskRun(string taskId, DateTime? expectedLastStartUtc, DateTi /// New end of the lease Task RenewTaskLease(string taskId, string instanceId, DateTime leasedUntilUtc); + /// + /// Records the result of a claimed run and releases its lease. Writes only the result fields, + /// and only while the instance still holds the lease - so it neither reverts changes saved in + /// the admin during the run, nor overwrites the claim of an instance that took the task over + /// after the lease ran out. + /// + /// Task identifier + /// Unique identifier of the application instance that ran the task + /// Whether the run completed without an error + /// End of the run + /// Disables the task (a failed run of a task that stops on error) + Task CompleteTaskRun(string taskId, string instanceId, bool succeeded, DateTime endUtc, bool disableTask); + /// /// Delete the task /// diff --git a/src/Modules/Grand.Module.ScheduledTasks/BackgroundServices/ScheduleTaskService.cs b/src/Modules/Grand.Module.ScheduledTasks/BackgroundServices/ScheduleTaskService.cs index 7ce0d320b0..a06adf1eaa 100644 --- a/src/Modules/Grand.Module.ScheduledTasks/BackgroundServices/ScheduleTaskService.cs +++ b/src/Modules/Grand.Module.ScheduledTasks/BackgroundServices/ScheduleTaskService.cs @@ -117,6 +117,27 @@ await _taskRepository.UpdateOneAsync( .Set(x => x.LeasedUntilUtc, TruncateToMilliseconds(leasedUntilUtc))); } + /// + /// Records the result of a claimed run and releases its lease, only while the given instance still holds it + /// + public virtual async Task CompleteTaskRun(string taskId, string instanceId, bool succeeded, DateTime endUtc, + bool disableTask) + { + var update = UpdateBuilder.Create() + .Set(x => x.LeasedUntilUtc, null); + if (succeeded) + update.Set(x => x.LastSuccessUtc, TruncateToMilliseconds(endUtc)) + .Set(x => x.LastNonSuccessEndUtc, null); + else + update.Set(x => x.LastNonSuccessEndUtc, TruncateToMilliseconds(endUtc)); + if (disableTask) + update.Set(x => x.Enabled, false); + + //LeasedByInstance changes only with a new claim, so a match means no other instance has + //taken the task over - runs of one task on one instance never overlap + await _taskRepository.UpdateOneAsync(x => x.Id == taskId && x.LeasedByInstance == instanceId, update); + } + private static DateTime TruncateToMilliseconds(DateTime value) { return new DateTime(value.Ticks - value.Ticks % TimeSpan.TicksPerMillisecond, DateTimeKind.Utc); diff --git a/src/Tests/Grand.Modules.Tests/Services/BackgroundService/ScheduleTaskServiceTests.cs b/src/Tests/Grand.Modules.Tests/Services/BackgroundService/ScheduleTaskServiceTests.cs index cf4ffe9aef..45cc2f1608 100644 --- a/src/Tests/Grand.Modules.Tests/Services/BackgroundService/ScheduleTaskServiceTests.cs +++ b/src/Tests/Grand.Modules.Tests/Services/BackgroundService/ScheduleTaskServiceTests.cs @@ -1,3 +1,4 @@ +using Grand.Business.Core.Interfaces.System.ScheduleTasks; using Grand.Data.Tests.LiteDb; using Grand.Data.Tests.MongoDb; using Grand.Domain.Tasks; @@ -150,6 +151,79 @@ public async Task RenewTaskLease_LeaseReleased_DoesNotRecreateIt() var updated = await _service.GetTaskById(task.Id); Assert.IsNull(updated.LeasedUntilUtc); } + + [TestMethod] + public async Task CompleteTaskRun_Succeeded_RecordsSuccessAndReleasesLease() + { + var task = await InsertTask(DateTime.UtcNow, DateTime.UtcNow.AddMinutes(5), "instance-a"); + + await _service.CompleteTaskRun(task.Id, "instance-a", true, DateTime.UtcNow, false); + + var updated = await _service.GetTaskById(task.Id); + Assert.IsNull(updated.LeasedUntilUtc); + Assert.IsNotNull(updated.LastSuccessUtc); + Assert.IsNull(updated.LastNonSuccessEndUtc); + Assert.IsTrue(updated.Enabled); + } + + [TestMethod] + public async Task CompleteTaskRun_FailedAndDisable_RecordsFailureAndDisablesTask() + { + var task = await InsertTask(DateTime.UtcNow, DateTime.UtcNow.AddMinutes(5), "instance-a"); + + await _service.CompleteTaskRun(task.Id, "instance-a", false, DateTime.UtcNow, true); + + var updated = await _service.GetTaskById(task.Id); + Assert.IsNull(updated.LeasedUntilUtc); + Assert.IsNull(updated.LastSuccessUtc); + Assert.IsNotNull(updated.LastNonSuccessEndUtc); + Assert.IsFalse(updated.Enabled); + } + + [TestMethod] + public async Task CompleteTaskRun_LeaseTakenOverByAnotherInstance_LeavesItsClaimAlone() + { + //instance A's lease ran out during a long run and instance B claimed the task + var task = await InsertTask(DateTime.UtcNow, DateTime.UtcNow.AddMinutes(5), "instance-b"); + + await _service.CompleteTaskRun(task.Id, "instance-a", true, DateTime.UtcNow, false); + + var updated = await _service.GetTaskById(task.Id); + Assert.AreEqual("instance-b", updated.LeasedByInstance); + Assert.IsNotNull(updated.LeasedUntilUtc); + Assert.IsNull(updated.LastSuccessUtc); + } + + [TestMethod] + public async Task CompleteTaskRun_KeepsChangesSavedDuringTheRun() + { + var task = await InsertTask(DateTime.UtcNow, DateTime.UtcNow.AddMinutes(5), "instance-a"); + //the admin disables the task and changes its interval while it runs + task.Enabled = false; + task.TimeInterval = 30; + await _service.UpdateTask(task); + + await _service.CompleteTaskRun(task.Id, "instance-a", true, DateTime.UtcNow, false); + + var updated = await _service.GetTaskById(task.Id); + Assert.IsFalse(updated.Enabled); + Assert.AreEqual(30, updated.TimeInterval); + } + + [TestMethod] + public async Task TryClaimTaskRun_ObsoleteOverloadWithoutLease_DoesNotBlockTheNextClaim() + { + var task = await InsertTask(DateTime.UtcNow.AddMinutes(-10)); + IScheduleTaskService service = _service; + +#pragma warning disable CS0618 + var claimed = await service.TryClaimTaskRun(task.Id, task.LastStartUtc, DateTime.UtcNow, "instance-a"); +#pragma warning restore CS0618 + + Assert.IsTrue(claimed); + var updated = await _service.GetTaskById(task.Id); + Assert.IsFalse(updated.LeasedUntilUtc > DateTime.UtcNow); + } } /// @@ -203,4 +277,17 @@ public async Task TryClaimTaskRun_NoLeaseOrLeaseRanOut_ReturnsTrue() Assert.IsTrue(await _service.TryClaimTaskRun(expired.Id, expired.LastStartUtc, DateTime.UtcNow, DateTime.UtcNow.AddMinutes(5), "instance-b")); } + + [TestMethod] + public async Task CompleteTaskRun_OwnLease_ReleasesItAndLeavesTakenOverOneAlone() + { + var own = await InsertTask(DateTime.UtcNow.AddMinutes(5)); + var takenOver = await InsertTask(DateTime.UtcNow.AddMinutes(5)); + + await _service.CompleteTaskRun(own.Id, "instance-a", true, DateTime.UtcNow, false); + await _service.CompleteTaskRun(takenOver.Id, "instance-b", true, DateTime.UtcNow, false); + + Assert.IsNull((await _service.GetTaskById(own.Id)).LeasedUntilUtc); + Assert.IsNotNull((await _service.GetTaskById(takenOver.Id)).LeasedUntilUtc); + } } diff --git a/src/Tests/Grand.Web.Common.Tests/Infrastructure/BackgroundServiceTaskTests.cs b/src/Tests/Grand.Web.Common.Tests/Infrastructure/BackgroundServiceTaskTests.cs index d7180993a6..65e332c251 100644 --- a/src/Tests/Grand.Web.Common.Tests/Infrastructure/BackgroundServiceTaskTests.cs +++ b/src/Tests/Grand.Web.Common.Tests/Infrastructure/BackgroundServiceTaskTests.cs @@ -67,23 +67,27 @@ public async Task ExecuteAsync_ClaimWon_ExecutesTaskAndPersistsResult() await service.StartAsync(cts.Token); await WaitFor(() => _scheduleTaskServiceMock.Invocations.Any(i => - i.Method.Name == nameof(IScheduleTaskService.UpdateTask))); + i.Method.Name == nameof(IScheduleTaskService.CompleteTaskRun))); cts.Cancel(); _scheduleTaskMock.Verify(t => t.Execute(), Times.Once); - _scheduleTaskServiceMock.Verify(s => s.UpdateTask(It.Is(x => - x.LastSuccessUtc != null && x.LastStartUtc != null)), Times.Once); + _scheduleTaskServiceMock.Verify(s => s.CompleteTaskRun(_task.Id, It.IsAny(), true, + It.IsAny(), false), Times.Once); + //a write of the whole entity read before the run would revert admin edits made during it + _scheduleTaskServiceMock.Verify(s => s.UpdateTask(It.IsAny()), Times.Never); } [TestMethod] public async Task ExecuteAsync_ClaimWon_LeasesTheRunAndReleasesItWhenDone() { DateTime? claimedRunStartUtc = null, claimedLeasedUntilUtc = null; + string claimingInstanceId = null; _scheduleTaskServiceMock.Setup(s => s.TryClaimTaskRun(_task.Id, It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny())) - .Callback((_, _, runStartUtc, leasedUntilUtc, _) => { + .Callback((_, _, runStartUtc, leasedUntilUtc, instanceId) => { claimedRunStartUtc = runStartUtc; claimedLeasedUntilUtc = leasedUntilUtc; + claimingInstanceId = instanceId; }) .ReturnsAsync(true); @@ -92,12 +96,34 @@ public async Task ExecuteAsync_ClaimWon_LeasesTheRunAndReleasesItWhenDone() await service.StartAsync(cts.Token); await WaitFor(() => _scheduleTaskServiceMock.Invocations.Any(i => - i.Method.Name == nameof(IScheduleTaskService.UpdateTask))); + i.Method.Name == nameof(IScheduleTaskService.CompleteTaskRun))); cts.Cancel(); Assert.AreEqual(LeaseDuration, claimedLeasedUntilUtc - claimedRunStartUtc); - _scheduleTaskServiceMock.Verify(s => s.UpdateTask(It.Is(x => - x.LeasedUntilUtc == null)), Times.Once); + //released as the instance that claimed the run - a takeover by another instance is left alone + _scheduleTaskServiceMock.Verify(s => s.CompleteTaskRun(_task.Id, claimingInstanceId, It.IsAny(), + It.IsAny(), It.IsAny()), Times.Once); + } + + [TestMethod] + public async Task ExecuteAsync_TaskFailsAndStopsOnError_RecordsFailureAndDisablesTask() + { + _task.StopOnError = true; + _scheduleTaskServiceMock.Setup(s => s.TryClaimTaskRun(_task.Id, It.IsAny(), + It.IsAny(), It.IsAny(), It.IsAny())) + .ReturnsAsync(true); + _scheduleTaskMock.Setup(t => t.Execute()).ThrowsAsync(new InvalidOperationException("task failure")); + + var service = new BackgroundServiceTask(TaskName, _serviceProvider); + using var cts = new CancellationTokenSource(); + await service.StartAsync(cts.Token); + + await WaitFor(() => _scheduleTaskServiceMock.Invocations.Any(i => + i.Method.Name == nameof(IScheduleTaskService.CompleteTaskRun))); + cts.Cancel(); + + _scheduleTaskServiceMock.Verify(s => s.CompleteTaskRun(_task.Id, It.IsAny(), false, + It.IsAny(), true), Times.Once); } [TestMethod] @@ -138,7 +164,7 @@ public async Task ExecuteAsync_ClaimWon_TaskSeesBackgroundWorkAndStoreContext() await service.StartAsync(cts.Token); await WaitFor(() => _scheduleTaskServiceMock.Invocations.Any(i => - i.Method.Name == nameof(IScheduleTaskService.UpdateTask))); + i.Method.Name == nameof(IScheduleTaskService.CompleteTaskRun))); cts.Cancel(); Assert.AreSame(workContext, seenWorkContext); @@ -165,6 +191,8 @@ await WaitFor(() => _scheduleTaskServiceMock.Invocations.Any(i => _scheduleTaskMock.Verify(t => t.Execute(), Times.Never); //losing instance must not overwrite the winner's data with a stale entity _scheduleTaskServiceMock.Verify(s => s.UpdateTask(It.IsAny()), Times.Never); + _scheduleTaskServiceMock.Verify(s => s.CompleteTaskRun(It.IsAny(), It.IsAny(), + It.IsAny(), It.IsAny(), It.IsAny()), Times.Never); } [TestMethod] diff --git a/src/Web/Grand.Web.Common/Infrastructure/BackgroundServiceTask.cs b/src/Web/Grand.Web.Common/Infrastructure/BackgroundServiceTask.cs index 538356b7b3..eeb99c3120 100644 --- a/src/Web/Grand.Web.Common/Infrastructure/BackgroundServiceTask.cs +++ b/src/Web/Grand.Web.Common/Infrastructure/BackgroundServiceTask.cs @@ -138,7 +138,6 @@ await RunTask(serviceProvider, scheduleTaskService, task, private async Task RunTask(IServiceProvider serviceProvider, IScheduleTaskService scheduleTaskService, ScheduleTask task, bool due, ILogger logger, CancellationToken stoppingToken) { - var updateTask = false; var scheduleTask = serviceProvider.GetRequiredKeyedService(task.ScheduleTaskName); if (scheduleTask != null) { @@ -152,9 +151,7 @@ private async Task RunTask(IServiceProvider serviceProvider, IScheduleTaskServic runStartUtc, runStartUtc.Add(LeaseDuration), InstanceId); if (claimed) { - updateTask = true; - task.LastStartUtc = runStartUtc; - task.LeasedByInstance = InstanceId; + var succeeded = false; //keeps other instances off the task while it runs, even past its next due time using var leaseCts = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken); @@ -171,8 +168,7 @@ private async Task RunTask(IServiceProvider serviceProvider, IScheduleTaskServic logger.LogInformation("Task {TaskName} execute", Name); await scheduleTask.Execute(); - task.LastSuccessUtc = DateTime.UtcNow; - task.LastNonSuccessEndUtc = null; + succeeded = true; } catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) { @@ -181,8 +177,6 @@ private async Task RunTask(IServiceProvider serviceProvider, IScheduleTaskServic } catch (Exception exc) { - task.LastNonSuccessEndUtc = DateTime.UtcNow; - task.Enabled = !task.StopOnError; logger.LogError(exc, "Error while running the \'{TaskScheduleTaskName}\' schedule task", task.ScheduleTaskName); @@ -193,8 +187,12 @@ private async Task RunTask(IServiceProvider serviceProvider, IScheduleTaskServic await leaseRenewal; } - //the run is over - the write below releases the lease for the next one - task.LeasedUntilUtc = null; + //the run is over - record its result and release the lease for the next one. + //Not a write of the whole entity: that would revert changes saved in the admin + //during the run, and - if the lease ran out and another instance took the task + //over - overwrite that instance's claim, letting a third one start + await scheduleTaskService.CompleteTaskRun(task.Id, InstanceId, succeeded, DateTime.UtcNow, + !succeeded && task.StopOnError); } else { @@ -205,16 +203,11 @@ private async Task RunTask(IServiceProvider serviceProvider, IScheduleTaskServic } else { - updateTask = true; task.Enabled = !task.StopOnError; task.LastNonSuccessEndUtc = DateTime.UtcNow; logger.LogError("Type {TaskName} is not registered", Name); - } - - //persist only when this instance actually ran the task - an unconditional - //write would overwrite the claim/results of the winning instance with stale data - if (updateTask) await scheduleTaskService.UpdateTask(task); + } } ///