From d88a59149cfd2605d6d0924236eb4cd4f7cfe86a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jacek=20=C5=81api=C5=84ski?= Date: Wed, 11 Nov 2026 14:10:07 +0100 Subject: [PATCH] fix: save recurrence changes that keep the same expression SetRecurrenceAsync promised to update a recurrence when its data changed, but UpdateAsync only acted on a new cron expression: new data or options with the same expression were silently dropped. JobOptions also compared the retry delays by reference, so a recurrence read from the store almost never equalled the new one. - JobOptions compares and hashes the retry delays by value. - Any change of the expression, data or options updates the recurrence. The job waiting for the next occurrence is reconfigured in place (Job.Reconfigure) and, for a new expression, moved to the new time, instead of a second job being added next to it (which used to run once more on the old schedule). A running job keeps its settings; the next one is created from the updated recurrence. If a runner takes the waiting job meanwhile, the recurrence is still updated. - The EF store saves a job's data and options on update. RecurringJobBuilderTests.SetRecurrenceAsync_ExistingRecurrenceWithSameRule_DoesNotUpdate passed only because of the bug (its stored data and options differed from what the builder produces); it now compares an unchanged recurrence. Co-Authored-By: Claude Opus 5.5 --- README.md | 2 +- src/RunWay.Core/Jobs/Job.cs | 23 ++- src/RunWay.Core/Jobs/JobOptions.cs | 9 +- src/RunWay.Core/Recurrences/Recurrence.cs | 7 + src/RunWay.EntityFramework/Store.cs | 4 + src/RunWay.Storage.Tests/StoreJobTestsBase.cs | 18 +- .../Unit/RecurrenceUpdateTests.cs | 171 ++++++++++++++++++ .../Unit/RecurringJobBuilderTests.cs | 32 +++- src/RunWay/Schedule/RecurringJobBuilder.cs | 32 +++- 9 files changed, 283 insertions(+), 15 deletions(-) create mode 100644 src/RunWay.Tests/Unit/RecurrenceUpdateTests.cs diff --git a/README.md b/README.md index b486a37..4bd18f0 100644 --- a/README.md +++ b/README.md @@ -114,7 +114,7 @@ await app.SetRecurrenceAsync("nightly-cleanup", "0 2 * * *", new CleanupJob(), o }); ``` -- Safe to call on every startup — only updates if the expression or data changed +- Safe to call on every startup — only updates if the expression, data or options changed; a job already waiting for the next occurrence takes the change, a running one keeps its settings and the next one uses the new ones - Each recurrence is identified by a unique string key --- diff --git a/src/RunWay.Core/Jobs/Job.cs b/src/RunWay.Core/Jobs/Job.cs index 27d8dda..f4090b7 100644 --- a/src/RunWay.Core/Jobs/Job.cs +++ b/src/RunWay.Core/Jobs/Job.cs @@ -25,9 +25,9 @@ public static Job CreateNew(Recurrence recurrence, DateTimeOffset scheduledAt) public int Version { get; } = version; - public JobData Data { get; } = data; + public JobData Data { get; private set; } = data; - public JobOptions Options { get; } = options; + public JobOptions Options { get; private set; } = options; public JobStatus Status { get; private set; } = status; @@ -43,6 +43,25 @@ public static Job CreateNew(Recurrence recurrence, DateTimeOffset scheduledAt) public IEnumerable PendingEvents => _pendingEvents; + /// + /// Changes the data, options and time of a job that is still waiting, e.g. when its recurrence changes. + /// + public void Reconfigure(JobData data, JobOptions options, DateTimeOffset scheduledAt) + { + if (this.RunnerId != null || this.Status != JobStatus.Scheduled) + throw new InvalidOperationException("Only scheduled jobs that are not being processed by any runner can be reconfigured."); + + this.Data = data; + this.Options = options; + if (scheduledAt != this.ScheduledAt) + { + this.ScheduledAt = scheduledAt; + this._pendingEvents.Add(new JobAuditRecord(DateTimeOffset.UtcNow, JobAuditRecordType.Scheduled, scheduledAt.UtcDateTime.ToString("O"))); + } + } + + public bool IsWaiting => this.Status == JobStatus.Scheduled && this.RunnerId == null; + public void Started(RunnerInfo runnerInfo) { if(this.RunnerId != null || this.Status != JobStatus.Scheduled) diff --git a/src/RunWay.Core/Jobs/JobOptions.cs b/src/RunWay.Core/Jobs/JobOptions.cs index 699e234..97394e6 100644 --- a/src/RunWay.Core/Jobs/JobOptions.cs +++ b/src/RunWay.Core/Jobs/JobOptions.cs @@ -14,11 +14,16 @@ public bool Equals(JobOptions? other) { if (other is null) return false; if (ReferenceEquals(this, other)) return true; - return Priority == other.Priority && RetryDelaysInSeconds.Equals(other.RetryDelaysInSeconds) && Timeout.Equals(other.Timeout); + // The delays are compared by value: options read from the store are a different list instance. + return Priority == other.Priority && RetryDelaysInSeconds.SequenceEqual(other.RetryDelaysInSeconds) && Timeout.Equals(other.Timeout); } public override int GetHashCode() { - return HashCode.Combine(Priority, RetryDelaysInSeconds, Timeout); + var hash = new HashCode(); + hash.Add(Priority); + hash.Add(Timeout); + foreach (var delay in RetryDelaysInSeconds) hash.Add(delay); + return hash.ToHashCode(); } } \ No newline at end of file diff --git a/src/RunWay.Core/Recurrences/Recurrence.cs b/src/RunWay.Core/Recurrences/Recurrence.cs index 8a987d8..dfcd6c6 100644 --- a/src/RunWay.Core/Recurrences/Recurrence.cs +++ b/src/RunWay.Core/Recurrences/Recurrence.cs @@ -29,6 +29,13 @@ public void Update(string rule, JobData data, JobOptions options, Job currentJob this.CurrentJob = currentJob; } + public void Update(string rule, JobData data, JobOptions options) + { + this.Rule = rule; + this.Data = data; + this.Options = options; + } + public void AssignJob(Job job) { if (this.CurrentJob?.Status.IsCompleted() == false) diff --git a/src/RunWay.EntityFramework/Store.cs b/src/RunWay.EntityFramework/Store.cs index 6e42a0b..5f71ae5 100644 --- a/src/RunWay.EntityFramework/Store.cs +++ b/src/RunWay.EntityFramework/Store.cs @@ -165,6 +165,10 @@ public async Task UpdateWithoutTransactionAsync(Job job, CancellationToken dbJob.ScheduledAt = job.ScheduledAt.UtcDateTime; dbJob.RetriesCount = job.RetriesCount; dbJob.RunnerId = job.RunnerId != null ? ParseId(job.RunnerId.Value) : null; + dbJob.Type = job.Data.Type; + dbJob.Data = job.Data.Data; + dbJob.Priority = job.Options.Priority; + dbJob.Options = new DbJobOptions(job.Options.RetryDelaysInSeconds.ToArray(), job.Options.Timeout); foreach (var auditRecord in job.PendingEvents) { diff --git a/src/RunWay.Storage.Tests/StoreJobTestsBase.cs b/src/RunWay.Storage.Tests/StoreJobTestsBase.cs index c182334..cb11402 100644 --- a/src/RunWay.Storage.Tests/StoreJobTestsBase.cs +++ b/src/RunWay.Storage.Tests/StoreJobTestsBase.cs @@ -1,4 +1,4 @@ -using Kododo.RunWay.Core.Jobs; +using Kododo.RunWay.Core.Jobs; using Kododo.RunWay.Core.Jobs.Audit; using Kododo.RunWay.Core.Runners; using Kododo.RunWay.Core.Store; @@ -92,6 +92,22 @@ public async Task UpdateAsync_SavesAuditRecords() Assert.Contains(details.Audit, a => a.Type == JobAuditRecordType.Started); } + [Fact] + public async Task UpdateAsync_Reconfigured_PersistsDataOptionsAndTime() + { + var job = await Store.CreateAsync(NewJob(), CancellationToken.None); + var at = new DateTimeOffset(2030, 1, 1, 5, 0, 0, TimeSpan.Zero); + + job.Reconfigure(new JobData("NewType", "{\"kind\":\"weekly\"}"), new JobOptions(5, [30, 90], TimeSpan.FromMinutes(2)), at); + await Store.UpdateAsync(job, CancellationToken.None); + + var updated = (await Store.FindAsync(job.Id, CancellationToken.None))!; + Assert.Equal(new JobData("NewType", "{\"kind\":\"weekly\"}"), updated.Data); + Assert.Equal(new JobOptions(5, [30, 90], TimeSpan.FromMinutes(2)), updated.Options); + Assert.Equal(at, updated.ScheduledAt); + Assert.Equal(JobStatus.Scheduled, updated.Status); + } + [Fact] public async Task UpdateAsync_Started_Succeeded_StatusIsSucceeded() { diff --git a/src/RunWay.Tests/Unit/RecurrenceUpdateTests.cs b/src/RunWay.Tests/Unit/RecurrenceUpdateTests.cs new file mode 100644 index 0000000..1576802 --- /dev/null +++ b/src/RunWay.Tests/Unit/RecurrenceUpdateTests.cs @@ -0,0 +1,171 @@ +using Kododo.RunWay.Core.Jobs; +using Kododo.RunWay.Core.Recurrences; +using Kododo.RunWay.Core.Runners; +using Kododo.RunWay.Core.Store; +using Kododo.RunWay.Schedule; +using NSubstitute; +using NSubstitute.ExceptionExtensions; +using Xunit; + +namespace Kododo.RunWay.Tests.Unit; + +/// +/// SetRecurrenceAsync on an existing recurrence: a change of the data, the options or the expression is saved, and +/// the job waiting for the next occurrence gets it, without a second job. +/// +public class RecurrenceUpdateTests +{ + private const string Key = "report"; + private const string Cron = "0 2 * * *"; + private static readonly DateTimeOffset WaitingAt = new(2030, 1, 1, 2, 0, 0, TimeSpan.Zero); + private static readonly DateTimeOffset NextOccurrence = new(2030, 1, 1, 5, 0, 0, TimeSpan.Zero); + + private sealed class ReportJob + { + public string Kind { get; set; } = ""; + } + + private readonly IStore _store = Substitute.For(); + private readonly IRecurrenceCalculator _calculator = Substitute.For(); + private readonly RunWayOptions _options = new(); + + public RecurrenceUpdateTests() + { + _calculator.CalculateNextOccurrence(Arg.Any(), Arg.Any()).Returns(NextOccurrence); + _store.UpdateWithoutTransactionAsync(Arg.Any(), Arg.Any()).Returns(c => c.Arg()!); + _store.CreateWithoutTransactionAsync(Arg.Any(), Arg.Any()).Returns(c => c.Arg()!); + _store.UpdateRecurrenceWithoutTransactionAsync(Arg.Any(), Arg.Any()).Returns(c => c.Arg()!); + } + + private JobData Data(string kind) => _options.JobDataSerializer.Serialize(new ReportJob { Kind = kind }); + + private static JobOptions Options(int priority = 0, int[]? delays = null) => + new(priority, delays ?? [0, 10, 60, 600], TimeSpan.FromMinutes(10)); + + private static Job ExistingJob(JobData data, JobOptions options, JobStatus status = JobStatus.Scheduled, RunnerId? runner = null) => + new(new JobId("1"), 3, data, options, new RecurrenceId(Key), status, runner, DateTimeOffset.UtcNow, WaitingAt, 0); + + /// The recurrence as read back from the store: new instances, equal values. + private void Stored(string rule, JobData data, JobOptions options, Job? currentJob) => + _store.FindRecurrenceAsync(Arg.Any(), Arg.Any()) + .Returns(new Recurrence(new RecurrenceId(Key), 2, rule, data with { }, options with { RetryDelaysInSeconds = options.RetryDelaysInSeconds.ToArray() }, currentJob)); + + private Task SetAsync(string cron, string kind, Action>? configure = null) + { + var builder = new Scheduler(_store, _calculator, _options).Recurrence(new ReportJob { Kind = kind }); + configure?.Invoke(builder); + return builder.SetRecurrenceAsync(Key, cron, CancellationToken.None); + } + + [Fact] + public void JobOptions_WithEqualDelaysInDifferentLists_AreEqual() + { + var a = Options(1, [5, 10]); + var b = Options(1, [5, 10]); + + Assert.Equal(a, b); + Assert.Equal(a.GetHashCode(), b.GetHashCode()); + Assert.NotEqual(a, Options(1, [5, 11])); + } + + [Fact] + public async Task Unchanged_ReadBackFromStore_WritesNothing() + { + Stored(Cron, Data("daily"), Options(), ExistingJob(Data("daily"), Options())); + + await SetAsync(Cron, "daily"); + + await _store.DidNotReceiveWithAnyArgs().UpdateRecurrenceWithoutTransactionAsync(default!, default); + await _store.DidNotReceiveWithAnyArgs().UpdateWithoutTransactionAsync(default!, default); + await _store.DidNotReceiveWithAnyArgs().CreateWithoutTransactionAsync(default!, default); + } + + [Fact] + public async Task DataChanged_SameExpression_UpdatesRecurrenceAndWaitingJob() + { + Stored(Cron, Data("daily"), Options(), ExistingJob(Data("daily"), Options())); + + await SetAsync(Cron, "weekly"); + + await _store.Received(1).UpdateRecurrenceWithoutTransactionAsync( + Arg.Is(r => r != null && r.Data == Data("weekly") && r.CurrentJob!.Id == new JobId("1")), Arg.Any()); + await _store.Received(1).UpdateWithoutTransactionAsync( + Arg.Is(j => j != null && j.Data == Data("weekly") && j.ScheduledAt == WaitingAt), Arg.Any()); + await _store.DidNotReceiveWithAnyArgs().CreateWithoutTransactionAsync(default!, default); + } + + [Fact] + public async Task OptionsChanged_SameExpression_UpdatesRecurrenceAndWaitingJob() + { + Stored(Cron, Data("daily"), Options(), ExistingJob(Data("daily"), Options())); + + await SetAsync(Cron, "daily", b => b.WithPriority(5).WithRetryDelaysInSeconds(30)); + + await _store.Received(1).UpdateRecurrenceWithoutTransactionAsync( + Arg.Is(r => r != null && r.Options.Priority == 5), Arg.Any()); + await _store.Received(1).UpdateWithoutTransactionAsync( + Arg.Is(j => j != null && j.Options.Priority == 5 && j.Options.RetryDelaysInSeconds.SequenceEqual(new[] { 30 })), + Arg.Any()); + } + + [Fact] + public async Task ExpressionChanged_MovesWaitingJob_InsteadOfAddingASecondOne() + { + Stored(Cron, Data("daily"), Options(), ExistingJob(Data("daily"), Options())); + + await SetAsync("0 5 * * *", "daily"); + + await _store.Received(1).UpdateWithoutTransactionAsync( + Arg.Is(j => j != null && j.ScheduledAt == NextOccurrence), Arg.Any()); + await _store.DidNotReceiveWithAnyArgs().CreateWithoutTransactionAsync(default!, default); + await _store.Received(1).UpdateRecurrenceWithoutTransactionAsync( + Arg.Is(r => r != null && r.Rule == "0 5 * * *"), Arg.Any()); + } + + [Fact] + public async Task JobRunning_UpdatesOnlyRecurrence_NextJobUsesNewSettings() + { + var running = ExistingJob(Data("daily"), Options(), JobStatus.Running, new RunnerId("7")); + Stored(Cron, Data("daily"), Options(), running); + + await SetAsync(Cron, "weekly"); + + await _store.Received(1).UpdateRecurrenceWithoutTransactionAsync( + Arg.Is(r => r != null && r.Data == Data("weekly") && r.CurrentJob!.Id == running.Id), Arg.Any()); + await _store.DidNotReceiveWithAnyArgs().UpdateWithoutTransactionAsync(default!, default); + await _store.DidNotReceiveWithAnyArgs().CreateWithoutTransactionAsync(default!, default); + } + + [Fact] + public async Task JobCompleted_CreatesNextJobWithNewSettings() + { + Stored(Cron, Data("daily"), Options(), ExistingJob(Data("daily"), Options(), JobStatus.Succeeded)); + + await SetAsync(Cron, "weekly"); + + await _store.Received(1).CreateWithoutTransactionAsync( + Arg.Is(j => j != null && j.Data == Data("weekly") && j.ScheduledAt == NextOccurrence), Arg.Any()); + } + + [Fact] + public async Task RunnerTookTheJobMeanwhile_RecurrenceIsStillUpdated() + { + Stored(Cron, Data("daily"), Options(), ExistingJob(Data("daily"), Options())); + var conflict = new InvalidOperationException("The job has a newer version"); + _store.UpdateWithoutTransactionAsync(Arg.Any(), Arg.Any()).ThrowsAsync(conflict); + _store.IsConcurrencyException(conflict).Returns(true); + + await SetAsync(Cron, "weekly"); + + await _store.Received(1).UpdateRecurrenceWithoutTransactionAsync( + Arg.Is(r => r != null && r.Data == Data("weekly")), Arg.Any()); + } + + [Fact] + public void Reconfigure_RunningJob_Throws() + { + var running = ExistingJob(Data("daily"), Options(), JobStatus.Running, new RunnerId("7")); + + Assert.Throws(() => running.Reconfigure(Data("weekly"), Options(), WaitingAt)); + } +} diff --git a/src/RunWay.Tests/Unit/RecurringJobBuilderTests.cs b/src/RunWay.Tests/Unit/RecurringJobBuilderTests.cs index 5c0de77..2a53a8c 100644 --- a/src/RunWay.Tests/Unit/RecurringJobBuilderTests.cs +++ b/src/RunWay.Tests/Unit/RecurringJobBuilderTests.cs @@ -1,4 +1,4 @@ -using Kododo.RunWay.Core.Jobs; +using Kododo.RunWay.Core.Jobs; using Kododo.RunWay.Core.Recurrences; using Kododo.RunWay.Core.Store; using Kododo.RunWay.Schedule; @@ -91,11 +91,14 @@ public async Task SetRecurrenceAsync_NewRecurrence_CalculatesNextOccurrence() } [Fact] - public async Task SetRecurrenceAsync_ExistingRecurrenceWithSameRule_DoesNotUpdate() + public async Task SetRecurrenceAsync_ExistingRecurrenceUnchanged_DoesNotUpdate() { const string expression = "0 * * * *"; - var existingData = _options.JobDataSerializer.Serialize(new { }); - var existingOptions = new JobOptions(0, [], TimeSpan.FromMinutes(1)); + // Serialized as the builder does it: RecurringJobBuilder stores the type as System.Object. + var existingData = _options.JobDataSerializer.Serialize(new { }); + // The options the builder produces by default, as a separate list instance, like when read back from the store. + var defaults = new JobOptionsBuilder(_options.SchedulerOptions); + var existingOptions = new JobOptions(defaults.Priority, defaults.RetryDelaysInSeconds.ToArray(), defaults.Timeout); var existing = Recurrence.CreateNew(new RecurrenceId("test-key"), expression, existingData, existingOptions); _store.FindRecurrenceAsync(Arg.Any(), Arg.Any()) @@ -110,6 +113,27 @@ public async Task SetRecurrenceAsync_ExistingRecurrenceWithSameRule_DoesNotUpdat await _store.DidNotReceive().UpdateRecurrenceWithoutTransactionAsync(Arg.Any(), Arg.Any()); } + [Fact] + public async Task SetRecurrenceAsync_ExistingRecurrenceWithSameRuleButOtherOptions_Updates() + { + const string expression = "0 * * * *"; + var existing = Recurrence.CreateNew(new RecurrenceId("test-key"), expression, + _options.JobDataSerializer.Serialize(new { }), new JobOptions(0, [], TimeSpan.FromMinutes(1))); + _store.FindRecurrenceAsync(Arg.Any(), Arg.Any()).Returns(existing); + _store.UpdateRecurrenceWithoutTransactionAsync(Arg.Any(), Arg.Any()) + .Returns(callInfo => callInfo.Arg()!); + _store.CreateWithoutTransactionAsync(Arg.Any(), Arg.Any()) + .Returns(callInfo => callInfo.Arg()!); + _calculator.CalculateNextOccurrence(Arg.Any(), Arg.Any()).Returns(_nextOccurrence); + + var builder = new RecurringJobBuilder(new { }, _store, _calculator, _options); + + await builder.SetRecurrenceAsync("test-key", expression, CancellationToken.None); + + await _store.Received(1).UpdateRecurrenceWithoutTransactionAsync( + Arg.Is(r => r != null && r.Options.Timeout == _options.SchedulerOptions.Timeout), Arg.Any()); + } + [Fact] public async Task SetRecurrenceAsync_ExistingRecurrenceWithDifferentRule_Updates() { diff --git a/src/RunWay/Schedule/RecurringJobBuilder.cs b/src/RunWay/Schedule/RecurringJobBuilder.cs index ca2e74a..1ab24ce 100644 --- a/src/RunWay/Schedule/RecurringJobBuilder.cs +++ b/src/RunWay/Schedule/RecurringJobBuilder.cs @@ -67,14 +67,36 @@ public async Task SetRecurrenceAsync(string key, string cronExpression, Cancella private async Task UpdateAsync(Recurrence recurrence, Recurrence existingRecurrence, CancellationToken stoppingToken) { var cronChanged = !existingRecurrence.Rule.Equals(recurrence.Rule); - if (cronChanged) + existingRecurrence.Update(recurrence.Rule, recurrence.Data, recurrence.Options); + var currentJob = existingRecurrence.CurrentJob; + + if (currentJob is { IsWaiting: true }) + { + // The waiting job takes the new data and options; with a new expression it also moves to the new time. + // No second job is created, so nothing runs twice or with the old settings. + var scheduledAt = cronChanged + ? _recurrenceCalculator.CalculateNextOccurrence(recurrence.Rule, DateTimeOffset.UtcNow) + : currentJob.ScheduledAt; + currentJob.Reconfigure(recurrence.Data, recurrence.Options, scheduledAt); + try + { + await _store.UpdateWithoutTransactionAsync(currentJob, stoppingToken); + } + catch (Exception ex) when (_store.IsConcurrencyException(ex)) + { + // A runner started the job in the meantime. It runs with the old settings; the next one is created + // from the updated recurrence when it completes. + } + } + else if (currentJob is null || currentJob.Status.IsCompleted()) { var nextOccurrence = _recurrenceCalculator.CalculateNextOccurrence(recurrence.Rule, DateTimeOffset.UtcNow); - var job = Job.CreateNew(recurrence, nextOccurrence); - job = await _store.CreateWithoutTransactionAsync(job, stoppingToken); - existingRecurrence.Update(recurrence.Rule, recurrence.Data, recurrence.Options, job); - await _store.UpdateRecurrenceWithoutTransactionAsync(existingRecurrence, stoppingToken); + var job = await _store.CreateWithoutTransactionAsync(Job.CreateNew(existingRecurrence, nextOccurrence), stoppingToken); + existingRecurrence.AssignJob(job); } + // A running job keeps its settings; the next one is created from the updated recurrence when it completes. + + await _store.UpdateRecurrenceWithoutTransactionAsync(existingRecurrence, stoppingToken); } private async Task CreateAsync(Recurrence recurrence, CancellationToken stoppingToken)