feat(durable-jobs): support successful rescheduling

src/Orleans.DurableJobs/DurableJob.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.Text.Json.Serialization;
 using Orleans.Runtime;
 
 namespace Orleans.DurableJobs;
@@ -60,4 +61,22 @@ public sealed class DurableJob
     /// </summary>
     [Id(7)]
     public string? TraceState { get; init; }
+
+    [Id(8)]
+    [JsonInclude]
+    internal long ExecutionGeneration { get; init; }
+
+    internal DurableJob WithExecutionGeneration(long executionGeneration) =>
+        new()
+        {
+            Id = Id,
+            Name = Name,
+            DueTime = DueTime,
+            TargetGrainId = TargetGrainId,
+            ShardId = ShardId,
+            Metadata = Metadata,
+            TraceParent = TraceParent,
+            TraceState = TraceState,
+            ExecutionGeneration = executionGeneration
+        };
 }

src/Orleans.DurableJobs/DurableJobRunResult.cs

@@ -16,16 +16,27 @@ public sealed class DurableJobRunResult
     public bool IsFailed => Status == DurableJobRunStatus.Failed;
 
     /// <summary>
-    /// Gets a value indicating whether the job should be polled again after a delay.
+    /// Gets a value indicating whether the job execution remains in progress and should be polled after a delay.
     /// </summary>
     [MemberNotNullWhen(true, nameof(PollAfterDelay))]
-    public bool IsPending => Status == DurableJobRunStatus.PollAfter;
+    public bool IsInProgress => Status == DurableJobRunStatus.InProgress;
 
-    private DurableJobRunResult(DurableJobRunStatus status, TimeSpan? pollAfter, Exception? exception)
+    /// <summary>
+    /// Gets a value indicating whether the successfully handled job requested durable rescheduling.
+    /// </summary>
+    [MemberNotNullWhen(true, nameof(RescheduleTime))]
+    public bool IsRescheduleRequested => Status == DurableJobRunStatus.RescheduleRequested && RescheduleTime is not null;
+
+    private DurableJobRunResult(
+        DurableJobRunStatus status,
+        TimeSpan? pollAfterDelay,
+        Exception? exception,
+        DateTimeOffset? rescheduleTime)
     {
         Status = status;
-        PollAfterDelay = pollAfter;
+        PollAfterDelay = pollAfterDelay;
         Exception = exception;
+        RescheduleTime = rescheduleTime;
     }
 
     /// <summary>
@@ -35,7 +46,7 @@ private DurableJobRunResult(DurableJobRunStatus status, TimeSpan? pollAfter, Exc
     public DurableJobRunStatus Status { get; }
 
     /// <summary>
-    /// Gets the delay before the next status check when <see cref="Status"/> is <see cref="DurableJobRunStatus.PollAfter"/>.
+    /// Gets the delay before the executor polls the ongoing run when <see cref="Status"/> is <see cref="DurableJobRunStatus.InProgress"/>.
     /// </summary>
     [Id(1)]
     public TimeSpan? PollAfterDelay { get; }
@@ -46,28 +57,49 @@ private DurableJobRunResult(DurableJobRunStatus status, TimeSpan? pollAfter, Exc
     [Id(2)]
     public Exception? Exception { get; }
 
-    private static readonly DurableJobRunResult CompletedInstance = new(DurableJobRunStatus.Completed, null, null);
+    /// <summary>
+    /// Gets the time at which the job should be durably rescheduled after the current execution completes successfully.
+    /// </summary>
+    [Id(3)]
+    public DateTimeOffset? RescheduleTime { get; }
+
+    private static readonly DurableJobRunResult CompletedInstance = new(DurableJobRunStatus.Completed, null, null, null);
 
     /// <summary>
     /// Gets a result indicating the job completed successfully.
     /// </summary>
     public static DurableJobRunResult Completed => CompletedInstance;
 
     /// <summary>
-    /// Creates a result indicating the job should be polled again after the specified delay.
+    /// Creates a result indicating the job execution remains in progress and should be polled after the specified delay.
     /// </summary>
-    /// <param name="delay">The time to wait before checking the job status again.</param>
-    /// <returns>A poll-after job result.</returns>
+    /// <param name="delay">The time to wait before polling the run again.</param>
+    /// <returns>An in-progress job result.</returns>
     /// <remarks>
-    /// The job will remain in an inline polling loop without being re-queued.
-    /// The polling loop will hold a concurrency slot until the job completes or fails.
+    /// The executor keeps the current run and its concurrency slot active, then polls the receiver again after the delay.
     /// TODO: Add validation for minimum/maximum poll delays to prevent abuse.
     /// TODO: Consider concurrency slot management for long-running polls.
     /// </remarks>
-    public static DurableJobRunResult PollAfter(TimeSpan delay)
+    public static DurableJobRunResult InProgress(TimeSpan delay)
     {
         ArgumentOutOfRangeException.ThrowIfLessThanOrEqual(delay, TimeSpan.Zero, nameof(delay));
-        return new(DurableJobRunStatus.PollAfter, delay, null);
+        return new(DurableJobRunStatus.InProgress, delay, null, null);
+    }
+
+    /// <summary>
+    /// Creates a result indicating that the current execution completed successfully and requested durable rescheduling.
+    /// </summary>
+    /// <param name="dueTime">The time at which the next execution should become due.</param>
+    /// <returns>A durable rescheduling result.</returns>
+    /// <remarks>
+    /// Rescheduling resets the failure-attempt count, so the next dequeue count is one.
+    /// During a mixed-version rolling upgrade, callers emit this result after every durable job executor
+    /// understands <see cref="DurableJobRunStatus.RescheduleRequested"/> (serialized value 3).
+    /// This rollout order preserves compatibility with executors that understand serialized values 0 through 2.
+    /// </remarks>
+    public static DurableJobRunResult RescheduleAt(DateTimeOffset dueTime)
+    {
+        return new(DurableJobRunStatus.RescheduleRequested, null, null, dueTime);
     }
 
     /// <summary>
@@ -81,7 +113,7 @@ public static DurableJobRunResult PollAfter(TimeSpan delay)
     public static DurableJobRunResult Failed(Exception exception)
     {
         ArgumentNullException.ThrowIfNull(exception);
-        return new(DurableJobRunStatus.Failed, null, exception);
+        return new(DurableJobRunStatus.Failed, null, exception, null);
     }
 }
 
@@ -93,15 +125,24 @@ public enum DurableJobRunStatus
     /// <summary>
     /// The job completed successfully and should be removed from the queue.
     /// </summary>
-    Completed,
+    Completed = 0,
 
     /// <summary>
-    /// The job is still running and should be polled again after the specified delay.
+    /// The job execution remains in progress and should be polled again after the specified delay.
     /// </summary>
-    PollAfter,
+    InProgress = 1,
 
     /// <summary>
     /// The job failed and should be processed through the retry policy.
     /// </summary>
-    Failed
+    Failed = 2,
+
+    /// <summary>
+    /// The current execution completed successfully and requested durable rescheduling with its failure-attempt count reset.
+    /// </summary>
+    /// <remarks>
+    /// This value is serialized as 3. During rolling upgrades, emit it after all durable job executors support it.
+    /// Newer executors treat unknown disposition values as failures and route them through the configured retry policy.
+    /// </remarks>
+    RescheduleRequested = 3
 }

src/Orleans.DurableJobs/DurableJobsInstruments.cs

@@ -15,6 +15,7 @@ internal sealed class DurableJobsInstruments(OrleansInstruments instruments)
     private const string StatusCompleted = "completed";
     private const string StatusFailed = "failed";
     private const string StatusRetried = "retried";
+    private const string StatusRescheduled = "rescheduled";
     private const string StatusCanceled = "canceled";
     private const string StatusError = "error";
     private const string StatusOk = "ok";
@@ -34,6 +35,7 @@ internal static DurableJobsInstruments CreateForDirectConstruction()
     private readonly Counter<long> JobsCompleted = instruments.Meter.CreateCounter<long>("orleans-durablejobs-jobs-completed");
     private readonly Counter<long> JobsFailed = instruments.Meter.CreateCounter<long>("orleans-durablejobs-jobs-failed");
     private readonly Counter<long> JobsRetried = instruments.Meter.CreateCounter<long>("orleans-durablejobs-jobs-retried");
+    private readonly Counter<long> JobsRescheduled = instruments.Meter.CreateCounter<long>("orleans-durablejobs-jobs-rescheduled");
     private readonly Counter<long> JobsCanceled = instruments.Meter.CreateCounter<long>("orleans-durablejobs-jobs-canceled");
     private readonly Counter<long> ShardsProcessed = instruments.Meter.CreateCounter<long>("orleans-durablejobs-shards-processed");
     private readonly Counter<long> ScheduleJobCalls = instruments.Meter.CreateCounter<long>("orleans-durablejobs-schedule-job-calls");
@@ -86,6 +88,12 @@ internal void OnJobRetried(TimeSpan latency)
         Record(JobAttemptDuration, latency, StatusRetried);
     }
 
+    internal void OnJobRescheduled(TimeSpan latency)
+    {
+        JobsRescheduled.Add(1);
+        Record(JobAttemptDuration, latency, StatusRescheduled);
+    }
+
     internal void OnJobCanceled()
     {
         JobsCanceled.Add(1);

src/Orleans.DurableJobs/IDurableJobReceiverExtension.cs

@@ -11,9 +11,9 @@ namespace Orleans.DurableJobs;
 internal interface IDurableJobReceiverExtension : IGrainExtension
 {
     /// <summary>
-    /// Handles a durable job by either starting execution or checking the status of an already running job.
+    /// Handles a durable job by either starting execution or checking the status of an execution which remains in progress.
     /// If the job attempt identified by <see cref="IJobRunContext.Job"/> and <see cref="IJobRunContext.DequeueCount"/> has not been started, it will be executed.
-    /// If it is already running, the current status is returned.
+    /// If it remains in progress, the current status is returned.
     /// </summary>
     /// <param name="context">The context containing information about the durable job.</param>
     /// <param name="cancellationToken">A token to monitor for cancellation requests.</param>
@@ -30,7 +30,7 @@ internal sealed partial class DurableJobReceiverExtension : IDurableJobReceiverE
 
     private readonly IGrainContext _grain;
     private readonly DurableJobReceiverExtensionShared _shared;
-    private readonly Dictionary<(string JobId, int DequeueCount), JobAttemptState> _jobAttempts = [];
+    private readonly Dictionary<(string JobId, long ExecutionGeneration, int DequeueCount), JobAttemptState> _jobAttempts = [];
     private readonly Queue<CompletedJobAttempt> _completedJobAttempts = new();
     private int _completedJobAttemptCount;
 
@@ -96,7 +96,11 @@ private async Task<DurableJobRunResult> ExecuteHandlerAsync(IDurableJobHandler h
         }
     }
 
-    private ValueTask<DurableJobRunResult> GetJobStatusAsync((string JobId, int DequeueCount) key, IJobRunContext context, JobAttemptState state, bool newJob)
+    private ValueTask<DurableJobRunResult> GetJobStatusAsync(
+        (string JobId, long ExecutionGeneration, int DequeueCount) key,
+        IJobRunContext context,
+        JobAttemptState state,
+        bool newJob)
     {
         // Cancellation is cooperative: only terminal task state is authoritative for job outcome.
         if (!state.Task.IsCompleted)
@@ -108,7 +112,7 @@ private ValueTask<DurableJobRunResult> GetJobStatusAsync((string JobId, int Dequ
                 return LongPollGetJobStatusAsync(key, context, state);
             }
 
-            return new(DurableJobRunResult.PollAfter(_shared.Options.JobStatusPollInterval));
+            return new(DurableJobRunResult.InProgress(_shared.Options.JobStatusPollInterval));
         }
 
         RecordCompletedJobAttempt(key, state);
@@ -127,7 +131,10 @@ private ValueTask<DurableJobRunResult> GetJobStatusAsync((string JobId, int Dequ
 
         return ValueTask.FromCanceled<DurableJobRunResult>(new CancellationToken(canceled: true));
 
-        async ValueTask<DurableJobRunResult> LongPollGetJobStatusAsync((string JobId, int DequeueCount) key, IJobRunContext context, JobAttemptState state)
+        async ValueTask<DurableJobRunResult> LongPollGetJobStatusAsync(
+            (string JobId, long ExecutionGeneration, int DequeueCount) key,
+            IJobRunContext context,
+            JobAttemptState state)
         {
             if (!state.Task.IsCompleted)
             {
@@ -137,7 +144,7 @@ async ValueTask<DurableJobRunResult> LongPollGetJobStatusAsync((string JobId, in
 
                 if (!state.Task.IsCompleted)
                 {
-                    return DurableJobRunResult.PollAfter(_shared.Options.JobStatusPollInterval);
+                    return DurableJobRunResult.InProgress(_shared.Options.JobStatusPollInterval);
                 }
             }
 
@@ -155,7 +162,7 @@ async ValueTask<DurableJobRunResult> LongPollGetJobStatusAsync((string JobId, in
         }
     }
 
-    private void RecordCompletedJobAttempt((string JobId, int DequeueCount) key, JobAttemptState state)
+    private void RecordCompletedJobAttempt((string JobId, long ExecutionGeneration, int DequeueCount) key, JobAttemptState state)
     {
         if (!state.CompletionRecorded)
         {
@@ -190,16 +197,16 @@ private void PruneCompletedJobAttempts()
                 && state.Task.IsCompleted
                 && state.CompletedTimestamp == completedAttempt.CompletedTimestamp)
             {
-                ((ICollection<KeyValuePair<(string JobId, int DequeueCount), JobAttemptState>>)_jobAttempts).Remove(
-                    new KeyValuePair<(string JobId, int DequeueCount), JobAttemptState>(completedAttempt.Key, state));
+                ((ICollection<KeyValuePair<(string JobId, long ExecutionGeneration, int DequeueCount), JobAttemptState>>)_jobAttempts).Remove(
+                    new KeyValuePair<(string JobId, long ExecutionGeneration, int DequeueCount), JobAttemptState>(completedAttempt.Key, state));
             }
 
             _completedJobAttemptCount--;
         }
     }
 
-    private static (string JobId, int DequeueCount) GetExecutionKey(IJobRunContext context)
-        => (context.Job.Id, context.DequeueCount);
+    private static (string JobId, long ExecutionGeneration, int DequeueCount) GetExecutionKey(IJobRunContext context)
+        => (context.Job.Id, context.Job.ExecutionGeneration, context.DequeueCount);
 
     private sealed class JobAttemptState(Task<DurableJobRunResult> task)
     {
@@ -210,7 +217,9 @@ private sealed class JobAttemptState(Task<DurableJobRunResult> task)
         public long CompletedTimestamp;
     }
 
-    private readonly record struct CompletedJobAttempt((string JobId, int DequeueCount) Key, long CompletedTimestamp);
+    private readonly record struct CompletedJobAttempt(
+        (string JobId, long ExecutionGeneration, int DequeueCount) Key,
+        long CompletedTimestamp);
 
     [LoggerMessage(Level = LogLevel.Error, Message = "Error executing durable job {JobId} on grain {GrainId}")]
     private static partial void LogErrorExecutingDurableJob(ILogger logger, Exception exception, string jobId, GrainId grainId);

src/Orleans.DurableJobs/InMemoryJobQueue.cs

@@ -112,7 +112,7 @@ public bool CancelJob(string jobId)
     public void RetryJobLater(IJobRunContext jobContext, DateTimeOffset newDueTime)
     {
         ArgumentNullException.ThrowIfNull(jobContext);
-        _ = RetryJobLater(jobContext.Job.Id, newDueTime, jobContext.DequeueCount);
+        _ = RetryJobLater(jobContext.Job.Id, newDueTime, jobContext.DequeueCount, executionGeneration: null);
     }
 
     /// <summary>
@@ -123,6 +123,9 @@ public void RetryJobLater(IJobRunContext jobContext, DateTimeOffset newDueTime)
     /// <param name="dequeueCount">The persisted dequeue count to associate with the retried job.</param>
     /// <returns>True if the job was found and rescheduled; false if the job was not found.</returns>
     public bool RetryJobLater(string jobId, DateTimeOffset newDueTime, int dequeueCount)
+        => RetryJobLater(jobId, newDueTime, dequeueCount, executionGeneration: null);
+
+    internal bool RetryJobLater(string jobId, DateTimeOffset newDueTime, int dequeueCount, long? executionGeneration)
     {
         ArgumentException.ThrowIfNullOrWhiteSpace(jobId);
         if (dequeueCount < 0)
@@ -147,6 +150,7 @@ public bool RetryJobLater(string jobId, DateTimeOffset newDueTime, int dequeueCo
                 Metadata = existing.Job.Metadata,
                 TraceParent = existing.Job.TraceParent,
                 TraceState = existing.Job.TraceState,
+                ExecutionGeneration = executionGeneration ?? existing.Job.ExecutionGeneration,
             };
 
             oldBucket.RemoveJob(jobId);

src/Orleans.DurableJobs/JobShard.cs

@@ -86,6 +86,18 @@ public interface IJobShard : IAsyncDisposable
     /// <returns>A task that represents the asynchronous operation.</returns>
     Task RetryJobLaterAsync(IJobRunContext jobContext, DateTimeOffset newDueTime, CancellationToken cancellationToken);
 
+    /// <summary>
+    /// Reschedules a successfully completed execution with its dequeue count reset.
+    /// </summary>
+    /// <param name="jobContext">The context of the completed execution.</param>
+    /// <param name="newDueTime">
+    /// The new due time for the job. This value supersedes <see cref="DurableJob.DueTime"/> on
+    /// <see cref="IJobRunContext.Job"/>.
+    /// </param>
+    /// <param name="cancellationToken">A token to cancel the operation.</param>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    Task RescheduleJobAsync(IJobRunContext jobContext, DateTimeOffset newDueTime, CancellationToken cancellationToken);
+
     /// <summary>
     /// Attempts to schedule a new job on this shard.
     /// </summary>
@@ -194,6 +206,22 @@ public async Task RetryJobLaterAsync(IJobRunContext jobContext, DateTimeOffset n
         _jobQueue.RetryJobLater(jobContext, newDueTime);
     }
 
+    /// <inheritdoc/>
+    public async Task RescheduleJobAsync(
+        IJobRunContext jobContext,
+        DateTimeOffset newDueTime,
+        CancellationToken cancellationToken)
+    {
+        ArgumentNullException.ThrowIfNull(jobContext);
+        var executionGeneration = checked(jobContext.Job.ExecutionGeneration + 1);
+        var resetContext = new JobRunContext(
+            jobContext.Job.WithExecutionGeneration(executionGeneration),
+            jobContext.RunId,
+            retryCount: 0);
+        await PersistRetryJobAsync(resetContext, newDueTime, cancellationToken);
+        _jobQueue.RetryJobLater(jobContext.Job.Id, newDueTime, dequeueCount: 0, executionGeneration);
+    }
+
     /// <summary>
     /// Enqueues a job into the in-memory queue with the specified dequeue count.
     /// </summary>