remove retry

Author: YunchuWang

src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs

@@ -16,20 +16,15 @@ namespace Microsoft.DurableTask;
 /// Azure Blob Storage implementation of <see cref="IPayloadStore"/>.
 /// Stores payloads as blobs and returns opaque tokens in the form "blob:v1:&lt;container&gt;:&lt;blobName&gt;".
 /// </summary>
-internal sealed class BlobPayloadStore : IPayloadStore
+sealed class BlobPayloadStore : IPayloadStore
 {
     const string TokenPrefix = "blob:v1:";
     const string ContentEncodingGzip = "gzip";
     const int DefaultCopyBufferSize = 81920;
     const int MaxRetryAttempts = 8;
     const int BaseDelayMs = 250;
     const int MaxDelayMs = 10_000;
-    const int MaxJitterMs = 100;
     const int NetworkTimeoutMinutes = 2;
-
-    // Jitter RNG for retry backoff
-    static readonly object RandomLock = new object();
-    static readonly Random SharedRandom = new Random();
     readonly BlobContainerClient containerClient;
     readonly LargePayloadStorageOptions options;
 
@@ -65,45 +60,37 @@ public BlobPayloadStore(LargePayloadStorageOptions options)
     public override async Task<string> UploadAsync(ReadOnlyMemory<byte> payloadBytes, CancellationToken cancellationToken)
     {
         // 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}";
+        string blobName = $"{Guid.NewGuid():N}";
         BlobClient blob = this.containerClient.GetBlobClient(blobName);
 
         byte[] payloadBuffer = payloadBytes.ToArray();
 
-        string token = await WithTransientRetryAsync(
-        async ct =>
-        {
-            // Ensure container exists (idempotent)
-            await this.containerClient.CreateIfNotExistsAsync(PublicAccessType.None, default, default, ct);
+        // Ensure container exists (idempotent)
+        await this.containerClient.CreateIfNotExistsAsync(PublicAccessType.None, default, default, cancellationToken);
 
-            if (this.options.CompressPayloads)
-            {
-                BlobOpenWriteOptions writeOptions = new()
-                {
-                    HttpHeaders = new BlobHttpHeaders { ContentEncoding = ContentEncodingGzip },
-                };
-                using Stream blobStream = await blob.OpenWriteAsync(true, writeOptions, ct);
-                using GZipStream compressedBlobStream = new(blobStream, System.IO.Compression.CompressionLevel.Optimal, leaveOpen: true);
-                using MemoryStream payloadStream = new(payloadBuffer, writable: false);
-
-                await payloadStream.CopyToAsync(compressedBlobStream, bufferSize: DefaultCopyBufferSize, ct);
-                await compressedBlobStream.FlushAsync(ct);
-                await blobStream.FlushAsync(ct);
-            }
-            else
+        if (this.options.CompressPayloads)
+        {
+            BlobOpenWriteOptions writeOptions = new()
             {
-                using Stream blobStream = await blob.OpenWriteAsync(true, default, ct);
-                using MemoryStream payloadStream = new(payloadBuffer, writable: false);
-                await payloadStream.CopyToAsync(blobStream, bufferSize: DefaultCopyBufferSize, ct);
-                await blobStream.FlushAsync(ct);
-            }
-
-            return EncodeToken(this.containerClient.Name, blobName);
-        },
-        cancellationToken);
+                HttpHeaders = new BlobHttpHeaders { ContentEncoding = ContentEncodingGzip },
+            };
+            using Stream blobStream = await blob.OpenWriteAsync(true, writeOptions, cancellationToken);
+            using GZipStream compressedBlobStream = new(blobStream, System.IO.Compression.CompressionLevel.Optimal, leaveOpen: true);
+            using MemoryStream payloadStream = new(payloadBuffer, writable: false);
+
+            await payloadStream.CopyToAsync(compressedBlobStream, bufferSize: DefaultCopyBufferSize, 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: DefaultCopyBufferSize, cancellationToken);
+            await blobStream.FlushAsync(cancellationToken);
+        }
 
-        return token;
+        return EncodeToken(this.containerClient.Name, blobName);
     }
 
     /// <inheritdoc/>
@@ -117,25 +104,20 @@ public override async Task<string> DownloadAsync(string token, CancellationToken
 
         BlobClient blob = this.containerClient.GetBlobClient(name);
 
-        return await WithTransientRetryAsync(
-        async ct =>
+        using BlobDownloadStreamingResult result = await blob.DownloadStreamingAsync(cancellationToken: cancellationToken);
+        Stream contentStream = result.Content;
+        bool isGzip = string.Equals(
+            result.Details.ContentEncoding, ContentEncodingGzip, StringComparison.OrdinalIgnoreCase);
+
+        if (isGzip)
         {
-            using BlobDownloadStreamingResult result = await blob.DownloadStreamingAsync(cancellationToken: ct);
-            Stream contentStream = result.Content;
-            bool isGzip = string.Equals(
-                result.Details.ContentEncoding, ContentEncodingGzip, StringComparison.OrdinalIgnoreCase);
+            using GZipStream decompressed = new(contentStream, CompressionMode.Decompress);
+            using StreamReader reader = new(decompressed, Encoding.UTF8);
+            return await reader.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);
+        using StreamReader uncompressedReader = new(contentStream, Encoding.UTF8);
+        return await uncompressedReader.ReadToEndAsync();
     }
 
     /// <inheritdoc/>
@@ -167,49 +149,4 @@ public override 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 = MaxRetryAttempts;
-        TimeSpan baseDelay = TimeSpan.FromMilliseconds(BaseDelayMs);
-        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;
-    }
-
-    static TimeSpan ComputeBackoff(TimeSpan baseDelay, int attempt)
-    {
-        double factor = Math.Pow(2, Math.Min(attempt, 6));
-        int jitterMs;
-        lock (RandomLock)
-        {
-            jitterMs = SharedRandom.Next(0, MaxJitterMs);
-        }
-
-        return TimeSpan.FromMilliseconds(Math.Min((baseDelay.TotalMilliseconds * factor) + jitterMs, MaxDelayMs));
-    }
-}
+}