fix: save recurrence changes that keep the same expression

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
 
 ---

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<JobAuditRecord> PendingEvents => _pendingEvents;
 
+    /// <summary>
+    /// Changes the data, options and time of a job that is still waiting, e.g. when its recurrence changes.
+    /// </summary>
+    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)

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();
     }
 }

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)

src/RunWay.EntityFramework/Store.cs

@@ -165,6 +165,10 @@ public async Task<Job> 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)
         {

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()
     {

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;
+
+/// <summary>
+/// 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.
+/// </summary>
+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<IStore>();
+    private readonly IRecurrenceCalculator _calculator = Substitute.For<IRecurrenceCalculator>();
+    private readonly RunWayOptions _options = new();
+
+    public RecurrenceUpdateTests()
+    {
+        _calculator.CalculateNextOccurrence(Arg.Any<string>(), Arg.Any<DateTimeOffset>()).Returns(NextOccurrence);
+        _store.UpdateWithoutTransactionAsync(Arg.Any<Job>(), Arg.Any<CancellationToken>()).Returns(c => c.Arg<Job>()!);
+        _store.CreateWithoutTransactionAsync(Arg.Any<Job>(), Arg.Any<CancellationToken>()).Returns(c => c.Arg<Job>()!);
+        _store.UpdateRecurrenceWithoutTransactionAsync(Arg.Any<Recurrence>(), Arg.Any<CancellationToken>()).Returns(c => c.Arg<Recurrence>()!);
+    }
+
+    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);
+
+    /// <summary>The recurrence as read back from the store: new instances, equal values.</summary>
+    private void Stored(string rule, JobData data, JobOptions options, Job? currentJob) =>
+        _store.FindRecurrenceAsync(Arg.Any<RecurrenceId>(), Arg.Any<CancellationToken>())
+            .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<IRecurringJobBuilder<ReportJob>>? 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<Recurrence>(r => r != null && r.Data == Data("weekly") && r.CurrentJob!.Id == new JobId("1")), Arg.Any<CancellationToken>());
+        await _store.Received(1).UpdateWithoutTransactionAsync(
+            Arg.Is<Job>(j => j != null && j.Data == Data("weekly") && j.ScheduledAt == WaitingAt), Arg.Any<CancellationToken>());
+        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<Recurrence>(r => r != null && r.Options.Priority == 5), Arg.Any<CancellationToken>());
+        await _store.Received(1).UpdateWithoutTransactionAsync(
+            Arg.Is<Job>(j => j != null && j.Options.Priority == 5 && j.Options.RetryDelaysInSeconds.SequenceEqual(new[] { 30 })),
+            Arg.Any<CancellationToken>());
+    }
+
+    [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<Job>(j => j != null && j.ScheduledAt == NextOccurrence), Arg.Any<CancellationToken>());
+        await _store.DidNotReceiveWithAnyArgs().CreateWithoutTransactionAsync(default!, default);
+        await _store.Received(1).UpdateRecurrenceWithoutTransactionAsync(
+            Arg.Is<Recurrence>(r => r != null && r.Rule == "0 5 * * *"), Arg.Any<CancellationToken>());
+    }
+
+    [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<Recurrence>(r => r != null && r.Data == Data("weekly") && r.CurrentJob!.Id == running.Id), Arg.Any<CancellationToken>());
+        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<Job>(j => j != null && j.Data == Data("weekly") && j.ScheduledAt == NextOccurrence), Arg.Any<CancellationToken>());
+    }
+
+    [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<Job>(), Arg.Any<CancellationToken>()).ThrowsAsync(conflict);
+        _store.IsConcurrencyException(conflict).Returns(true);
+
+        await SetAsync(Cron, "weekly");
+
+        await _store.Received(1).UpdateRecurrenceWithoutTransactionAsync(
+            Arg.Is<Recurrence>(r => r != null && r.Data == Data("weekly")), Arg.Any<CancellationToken>());
+    }
+
+    [Fact]
+    public void Reconfigure_RunningJob_Throws()
+    {
+        var running = ExistingJob(Data("daily"), Options(), JobStatus.Running, new RunnerId("7"));
+
+        Assert.Throws<InvalidOperationException>(() => running.Reconfigure(Data("weekly"), Options(), WaitingAt));
+    }
+}

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<object> stores the type as System.Object.
+        var existingData = _options.JobDataSerializer.Serialize<object>(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<RecurrenceId>(), Arg.Any<CancellationToken>())
@@ -110,6 +113,27 @@ public async Task SetRecurrenceAsync_ExistingRecurrenceWithSameRule_DoesNotUpdat
         await _store.DidNotReceive().UpdateRecurrenceWithoutTransactionAsync(Arg.Any<Recurrence>(), Arg.Any<CancellationToken>());
     }
 
+    [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<RecurrenceId>(), Arg.Any<CancellationToken>()).Returns(existing);
+        _store.UpdateRecurrenceWithoutTransactionAsync(Arg.Any<Recurrence>(), Arg.Any<CancellationToken>())
+            .Returns(callInfo => callInfo.Arg<Recurrence>()!);
+        _store.CreateWithoutTransactionAsync(Arg.Any<Job>(), Arg.Any<CancellationToken>())
+            .Returns(callInfo => callInfo.Arg<Job>()!);
+        _calculator.CalculateNextOccurrence(Arg.Any<string>(), Arg.Any<DateTimeOffset>()).Returns(_nextOccurrence);
+
+        var builder = new RecurringJobBuilder<object>(new { }, _store, _calculator, _options);
+
+        await builder.SetRecurrenceAsync("test-key", expression, CancellationToken.None);
+
+        await _store.Received(1).UpdateRecurrenceWithoutTransactionAsync(
+            Arg.Is<Recurrence>(r => r != null && r.Options.Timeout == _options.SchedulerOptions.Timeout), Arg.Any<CancellationToken>());
+    }
+
     [Fact]
     public async Task SetRecurrenceAsync_ExistingRecurrenceWithDifferentRule_Updates()
     {

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)