Merge branch 'wangbill/large-payload' of https://github.com/microsoft/durabletask-dotnet into wangbill/large-payload

Author: YunchuWang

src/Client/Core/DurableTaskClient.cs

@@ -399,6 +399,47 @@ public virtual Task<PurgeResult> PurgeAllInstancesAsync(
         throw new NotSupportedException($"{this.GetType()} does not support purging of orchestration instances.");
     }
 
+    /// <summary>
+    /// Restarts an orchestration instance with the same or a new instance ID.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// This method restarts an existing orchestration instance. If <paramref name="restartWithNewInstanceId"/> is <c>true</c>,
+    /// a new instance ID will be generated for the restarted orchestration. If <c>false</c>, the original instance ID will be reused.
+    /// </para><para>
+    /// The restarted orchestration will use the same input data as the original instance. If the original orchestration
+    /// instance is not found, an <see cref="ArgumentException"/> will be thrown.
+    /// </para><para>
+    /// Note that this operation is backend-specific and may not be supported by all durable task backends.
+    /// If the backend does not support restart operations, a <see cref="NotSupportedException"/> will be thrown.
+    /// </para>
+    /// </remarks>
+    /// <param name="instanceId">The ID of the orchestration instance to restart.</param>
+    /// <param name="restartWithNewInstanceId">
+    /// If <c>true</c>, a new instance ID will be generated for the restarted orchestration.
+    /// If <c>false</c>, the original instance ID will be reused.
+    /// </param>
+    /// <param name="cancellation">
+    /// The cancellation token. This only cancels enqueueing the restart request to the backend.
+    /// Does not abort restarting the orchestration once enqueued.
+    /// </param>
+    /// <returns>
+    /// A task that completes when the orchestration instance is successfully restarted.
+    /// The value of this task is the instance ID of the restarted orchestration instance.
+    /// </returns>
+    /// <exception cref="ArgumentException">
+    /// Thrown if an orchestration with the specified <paramref name="instanceId"/> was not found. </exception>
+    /// <exception cref="InvalidOperationException">
+    /// Thrown when attempting to restart an instance using the same instance Id
+    /// while the instance has not yet reached a completed or terminal state. </exception>
+    /// <exception cref="NotSupportedException">
+    /// Thrown if the backend does not support restart operations. </exception>
+    public virtual Task<string> RestartAsync(
+        string instanceId,
+        bool restartWithNewInstanceId = false,
+        CancellationToken cancellation = default)
+        => throw new NotSupportedException($"{this.GetType()} does not support orchestration restart.");
+
     // TODO: Create task hub
 
     // TODO: Delete task hub

src/Client/Grpc/GrpcDurableTaskClient.cs

@@ -414,6 +414,42 @@ public override Task<PurgeResult> PurgeAllInstancesAsync(
         return this.PurgeInstancesCoreAsync(request, cancellation);
     }
 
+    /// <inheritdoc/>
+    public override async Task<string> RestartAsync(
+        string instanceId,
+        bool restartWithNewInstanceId = false,
+        CancellationToken cancellation = default)
+    {
+        Check.NotNullOrEmpty(instanceId);
+        Check.NotEntity(this.options.EnableEntitySupport, instanceId);
+
+        var request = new P.RestartInstanceRequest
+        {
+            InstanceId = instanceId,
+            RestartWithNewInstanceId = restartWithNewInstanceId,
+        };
+
+        try
+        {
+            P.RestartInstanceResponse result = await this.sidecarClient.RestartInstanceAsync(
+                request, cancellationToken: cancellation);
+            return result.InstanceId;
+        }
+        catch (RpcException e) when (e.StatusCode == StatusCode.NotFound)
+        {
+            throw new ArgumentException($"An orchestration with the instanceId {instanceId} was not found.", e);
+        }
+        catch (RpcException e) when (e.StatusCode == StatusCode.FailedPrecondition)
+        {
+            throw new InvalidOperationException($"An orchestration with the instanceId {instanceId} cannot be restarted.", e);
+        }
+        catch (RpcException e) when (e.StatusCode == StatusCode.Cancelled)
+        {
+            throw new OperationCanceledException(
+                $"The {nameof(this.RestartAsync)} operation was canceled.", e, cancellation);
+        }
+    }
+
     static AsyncDisposable GetCallInvoker(GrpcDurableTaskClientOptions options, out CallInvoker callInvoker)
     {
         if (options.Channel is GrpcChannel c)

src/Client/OrchestrationServiceClientShim/ShimDurableTaskClient.cs

@@ -265,6 +265,59 @@ public override async Task<OrchestrationMetadata> WaitForInstanceStartAsync(
         }
     }
 
+    /// <inheritdoc/>
+    public override async Task<string> RestartAsync(
+        string instanceId,
+        bool restartWithNewInstanceId = false,
+        CancellationToken cancellation = default)
+    {
+        Check.NotNullOrEmpty(instanceId);
+        cancellation.ThrowIfCancellationRequested();
+
+        // Get the current orchestration status to retrieve the name and input
+        OrchestrationMetadata? status = await this.GetInstanceAsync(instanceId, getInputsAndOutputs: true, cancellation);
+
+        if (status == null)
+        {
+            throw new ArgumentException($"An orchestration with the instanceId {instanceId} was not found.");
+        }
+
+        bool isInstaceNotCompleted = status.RuntimeStatus == OrchestrationRuntimeStatus.Running ||
+                                    status.RuntimeStatus == OrchestrationRuntimeStatus.Pending ||
+                                    status.RuntimeStatus == OrchestrationRuntimeStatus.Suspended;
+
+        if (isInstaceNotCompleted && !restartWithNewInstanceId)
+        {
+            throw new InvalidOperationException($"Instance '{instanceId}' cannot be restarted while it is in state '{status.RuntimeStatus}'. " +
+                   "Wait until it has completed, or restart with a new instance ID.");
+        }
+
+        // Determine the instance ID for the restarted orchestration
+        string newInstanceId = restartWithNewInstanceId ? Guid.NewGuid().ToString("N") : instanceId;
+
+        OrchestrationInstance instance = new()
+        {
+            InstanceId = newInstanceId,
+            ExecutionId = Guid.NewGuid().ToString("N"),
+        };
+
+        // Use the original serialized input directly to avoid double serialization
+        // TODO: OrchestrationMetada doesn't have version property so we don't support version here.
+        // Issue link: https://github.com/microsoft/durabletask-dotnet/issues/463
+        TaskMessage message = new()
+        {
+            OrchestrationInstance = instance,
+            Event = new ExecutionStartedEvent(-1, status.SerializedInput)
+            {
+                Name = status.Name,
+                OrchestrationInstance = instance,
+            },
+        };
+
+        await this.Client.CreateTaskOrchestrationAsync(message);
+        return newInstanceId;
+    }
+
     [return: NotNullIfNotNull("state")]
     OrchestrationMetadata? ToMetadata(Core.OrchestrationState? state, bool getInputsAndOutputs)
     {

src/Extensions/AzureBlobPayloads/Converters/BlobPayloadStore.cs

@@ -4,6 +4,8 @@
 using System.Globalization;
 using System.IO.Compression;
 using System.Text;
+using Azure;
+using Azure.Core;
 using Azure.Storage.Blobs;
 using Azure.Storage.Blobs.Models;
 
@@ -16,6 +18,10 @@ namespace Microsoft.DurableTask.Converters;
 public sealed class BlobPayloadStore : IPayloadStore
 {
     const string TokenPrefix = "blob:v1:";
+
+    // Jitter RNG for retry backoff
+    static readonly object RandomLock = new object();
+    static readonly Random SharedRandom = new Random();
     readonly BlobContainerClient containerClient;
     readonly LargePayloadStorageOptions options;
 
@@ -32,47 +38,64 @@ public BlobPayloadStore(LargePayloadStorageOptions options)
         Check.NotNullOrEmpty(options.ConnectionString, nameof(options.ConnectionString));
         Check.NotNullOrEmpty(options.ContainerName, nameof(options.ContainerName));
 
-        BlobServiceClient serviceClient = new(options.ConnectionString);
+        BlobClientOptions clientOptions = new()
+        {
+            Retry =
+            {
+                Mode = RetryMode.Exponential,
+                MaxRetries = 8,
+                Delay = TimeSpan.FromMilliseconds(250),
+                MaxDelay = TimeSpan.FromSeconds(10),
+                NetworkTimeout = TimeSpan.FromMinutes(2),
+            },
+        };
+        BlobServiceClient serviceClient = new(options.ConnectionString, clientOptions);
         this.containerClient = serviceClient.GetBlobContainerClient(options.ContainerName);
     }
 
     /// <inheritdoc/>
     public async Task<string> UploadAsync(ReadOnlyMemory<byte> payloadBytes, CancellationToken cancellationToken)
     {
-        // Ensure container exists
-        await this.containerClient.CreateIfNotExistsAsync(PublicAccessType.None, default, default, cancellationToken);
-
-        // One blob per payload using GUID-based name for uniqueness
+        // One blob per payload using GUID-based name for uniqueness (stable across retries)
         string timestamp = DateTimeOffset.UtcNow.ToString("yyyy/MM/dd/HH/mm/ss", CultureInfo.InvariantCulture);
         string blobName = $"{timestamp}/{Guid.NewGuid():N}";
         BlobClient blob = this.containerClient.GetBlobClient(blobName);
 
         byte[] payloadBuffer = payloadBytes.ToArray();
 
-        // Upload streaming, optionally compressing and marking ContentEncoding
-        if (this.options.CompressPayloads)
+        string token = await WithTransientRetryAsync(
+        async ct =>
         {
-            BlobOpenWriteOptions writeOptions = new()
+            // Ensure container exists (idempotent)
+            await this.containerClient.CreateIfNotExistsAsync(PublicAccessType.None, default, default, ct);
+
+            if (this.options.CompressPayloads)
             {
-                HttpHeaders = new BlobHttpHeaders { ContentEncoding = "gzip" },
-            };
-            using Stream blobStream = await blob.OpenWriteAsync(true, writeOptions, cancellationToken);
-            using GZipStream compressedBlobStream = new(blobStream, CompressionLevel.Optimal, leaveOpen: true);
-            using MemoryStream payloadStream = new(payloadBuffer, writable: false);
-
-            await payloadStream.CopyToAsync(compressedBlobStream, bufferSize: 81920, cancellationToken);
-            await compressedBlobStream.FlushAsync(cancellationToken);
-            await blobStream.FlushAsync(cancellationToken);
-        }
-        else
-        {
-            using Stream blobStream = await blob.OpenWriteAsync(true, default, cancellationToken);
-            using MemoryStream payloadStream = new(payloadBuffer, writable: false);
-            await payloadStream.CopyToAsync(blobStream, bufferSize: 81920, cancellationToken);
-            await blobStream.FlushAsync(cancellationToken);
-        }
+                BlobOpenWriteOptions writeOptions = new()
+                {
+                    HttpHeaders = new BlobHttpHeaders { ContentEncoding = "gzip" },
+                };
+                using Stream blobStream = await blob.OpenWriteAsync(true, writeOptions, ct);
+                using GZipStream compressedBlobStream = new(blobStream, CompressionLevel.Optimal, leaveOpen: true);
+                using MemoryStream payloadStream = new(payloadBuffer, writable: false);
+
+                await payloadStream.CopyToAsync(compressedBlobStream, bufferSize: 81920, ct);
+                await compressedBlobStream.FlushAsync(ct);
+                await blobStream.FlushAsync(ct);
+            }
+            else
+            {
+                using Stream blobStream = await blob.OpenWriteAsync(true, default, ct);
+                using MemoryStream payloadStream = new(payloadBuffer, writable: false);
+                await payloadStream.CopyToAsync(blobStream, bufferSize: 81920, ct);
+                await blobStream.FlushAsync(ct);
+            }
+
+            return EncodeToken(this.containerClient.Name, blobName);
+        },
+        cancellationToken);
 
-        return EncodeToken(this.containerClient.Name, blobName);
+        return token;
     }
 
     /// <inheritdoc/>
@@ -85,20 +108,26 @@ public async Task<string> DownloadAsync(string token, CancellationToken cancella
         }
 
         BlobClient blob = this.containerClient.GetBlobClient(name);
-        using BlobDownloadStreamingResult result = await blob.DownloadStreamingAsync(cancellationToken: cancellationToken);
-        Stream contentStream = result.Content;
-        bool isGzip = string.Equals(
-            result.Details.ContentEncoding, "gzip", StringComparison.OrdinalIgnoreCase);
 
-        if (isGzip)
+        return await WithTransientRetryAsync(
+        async ct =>
         {
-            using GZipStream decompressed = new(contentStream, CompressionMode.Decompress);
-            using StreamReader reader = new(decompressed, Encoding.UTF8);
-            return await reader.ReadToEndAsync();
-        }
+            using BlobDownloadStreamingResult result = await blob.DownloadStreamingAsync(cancellationToken: ct);
+            Stream contentStream = result.Content;
+            bool isGzip = string.Equals(
+                result.Details.ContentEncoding, "gzip", StringComparison.OrdinalIgnoreCase);
 
-        using StreamReader uncompressedReader = new(contentStream, Encoding.UTF8);
-        return await uncompressedReader.ReadToEndAsync();
+            if (isGzip)
+            {
+                using GZipStream decompressed = new(contentStream, CompressionMode.Decompress);
+                using StreamReader reader = new(decompressed, Encoding.UTF8);
+                return await reader.ReadToEndAsync();
+            }
+
+            using StreamReader uncompressedReader = new(contentStream, Encoding.UTF8);
+            return await uncompressedReader.ReadToEndAsync();
+        },
+        cancellationToken);
     }
 
     /// <inheritdoc/>
@@ -130,4 +159,51 @@ public bool IsKnownPayloadToken(string value)
 
         return (rest.Substring(0, sep), rest.Substring(sep + 1));
     }
+
+    static async Task<T> WithTransientRetryAsync<T>(Func<CancellationToken, Task<T>> operation, CancellationToken cancellationToken)
+    {
+        const int maxAttempts = 8;
+        TimeSpan baseDelay = TimeSpan.FromMilliseconds(250);
+        int attempt = 0;
+
+        while (true)
+        {
+            cancellationToken.ThrowIfCancellationRequested();
+            try
+            {
+                return await operation(cancellationToken);
+            }
+            catch (RequestFailedException ex) when (IsTransient(ex) && attempt < maxAttempts - 1)
+            {
+                attempt++;
+                TimeSpan delay = ComputeBackoff(baseDelay, attempt);
+                await Task.Delay(delay, cancellationToken);
+            }
+            catch (IOException) when (attempt < maxAttempts - 1)
+            {
+                attempt++;
+                TimeSpan delay = ComputeBackoff(baseDelay, attempt);
+                await Task.Delay(delay, cancellationToken);
+            }
+        }
+    }
+
+    static bool IsTransient(RequestFailedException ex)
+    {
+        return ex.Status == 503 || ex.Status == 502 || ex.Status == 500 || ex.Status == 429 ||
+            string.Equals(ex.ErrorCode, "ServerBusy", StringComparison.OrdinalIgnoreCase) ||
+            string.Equals(ex.ErrorCode, "OperationTimedOut", StringComparison.OrdinalIgnoreCase);
+    }
+
+    static TimeSpan ComputeBackoff(TimeSpan baseDelay, int attempt)
+    {
+        double factor = Math.Pow(2, Math.Min(attempt, 6));
+        int jitterMs;
+        lock (RandomLock)
+        {
+            jitterMs = SharedRandom.Next(0, 100);
+        }
+        return TimeSpan.FromMilliseconds(Math.Min((baseDelay.TotalMilliseconds * factor) + jitterMs, 10_000));
+    }
+
 }

src/Grpc/orchestrator_service.proto

@@ -482,6 +482,15 @@ message PurgeInstancesResponse {
     google.protobuf.BoolValue isComplete = 2;
 }
 
+message RestartInstanceRequest {
+    string instanceId = 1;
+    bool restartWithNewInstanceId = 2;
+}
+
+message RestartInstanceResponse {
+    string instanceId = 1;
+}
+
 message CreateTaskHubRequest {
     bool recreateIfExists = 1;
 }
@@ -682,6 +691,9 @@ service TaskHubSidecarService {
     // Rewinds an orchestration instance to last known good state and replays from there.
     rpc RewindInstance(RewindInstanceRequest) returns (RewindInstanceResponse);
 
+    // Restarts an orchestration instance.
+    rpc RestartInstance(RestartInstanceRequest) returns (RestartInstanceResponse);
+
     // Waits for an orchestration instance to reach a running or completion state.
     rpc WaitForInstanceStart(GetInstanceRequest) returns (GetInstanceResponse);
 

src/Grpc/versions.txt

@@ -1,2 +1,2 @@
-# The following files were downloaded from branch main at 2025-08-08 16:46:11 UTC
-https://raw.githubusercontent.com/microsoft/durabletask-protobuf/e88acbd07ae38b499dbe8c4e333e9e3feeb2a9cc/protos/orchestrator_service.proto
+# The following files were downloaded from branch main at 2025-09-10 22:50:45 UTC
+https://raw.githubusercontent.com/microsoft/durabletask-protobuf/985035a0890575ae18be0eb2a3ac93c10824498a/protos/orchestrator_service.proto

test/Client/Core.Tests/DependencyInjection/DefaultDurableTaskClientBuilderTests.cs

@@ -144,6 +144,14 @@ public override Task<OrchestrationMetadata> WaitForInstanceStartAsync(
         {
             throw new NotImplementedException();
         }
+
+        public override Task<string> RestartAsync(
+            string instanceId, 
+            bool restartWithNewInstanceId = false, 
+            CancellationToken cancellation = default)
+        {
+            throw new NotImplementedException();
+        }
     }
 
     class CustomDataConverter : DataConverter