refactor

Author: YunchuWang

src/Extensions/AzureBlobPayloads/Converters/BlobPayloadStore.cs

@@ -18,6 +18,12 @@ namespace Microsoft.DurableTask.Converters;
 public 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;
 
     // Jitter RNG for retry backoff
     static readonly object RandomLock = new object();
@@ -43,8 +49,8 @@ public BlobPayloadStore(LargePayloadStorageOptions options)
             Retry =
             {
                 Mode = RetryMode.Exponential,
-                MaxRetries = 8,
-                Delay = TimeSpan.FromMilliseconds(250),
+                MaxRetries = MaxRetryAttempts,
+                Delay = TimeSpan.FromMilliseconds(BaseDelayMs),
                 MaxDelay = TimeSpan.FromSeconds(10),
                 NetworkTimeout = TimeSpan.FromMinutes(2),
             },
@@ -73,21 +79,21 @@ public async Task<string> UploadAsync(ReadOnlyMemory<byte> payloadBytes, Cancell
             {
                 BlobOpenWriteOptions writeOptions = new()
                 {
-                    HttpHeaders = new BlobHttpHeaders { ContentEncoding = "gzip" },
+                    HttpHeaders = new BlobHttpHeaders { ContentEncoding = ContentEncodingGzip },
                 };
                 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 payloadStream.CopyToAsync(compressedBlobStream, bufferSize: DefaultCopyBufferSize, 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 payloadStream.CopyToAsync(blobStream, bufferSize: DefaultCopyBufferSize, ct);
                 await blobStream.FlushAsync(ct);
             }
 
@@ -115,7 +121,7 @@ public async Task<string> DownloadAsync(string token, CancellationToken cancella
             using BlobDownloadStreamingResult result = await blob.DownloadStreamingAsync(cancellationToken: ct);
             Stream contentStream = result.Content;
             bool isGzip = string.Equals(
-                result.Details.ContentEncoding, "gzip", StringComparison.OrdinalIgnoreCase);
+                result.Details.ContentEncoding, ContentEncodingGzip, StringComparison.OrdinalIgnoreCase);
 
             if (isGzip)
             {
@@ -162,8 +168,8 @@ public bool IsKnownPayloadToken(string value)
 
     static async Task<T> WithTransientRetryAsync<T>(Func<CancellationToken, Task<T>> operation, CancellationToken cancellationToken)
     {
-        const int maxAttempts = 8;
-        TimeSpan baseDelay = TimeSpan.FromMilliseconds(250);
+        const int maxAttempts = MaxRetryAttempts;
+        TimeSpan baseDelay = TimeSpan.FromMilliseconds(BaseDelayMs);
         int attempt = 0;
 
         while (true)
@@ -201,9 +207,9 @@ static TimeSpan ComputeBackoff(TimeSpan baseDelay, int attempt)
         int jitterMs;
         lock (RandomLock)
         {
-            jitterMs = SharedRandom.Next(0, 100);
+            jitterMs = SharedRandom.Next(0, MaxJitterMs);
         }
-        return TimeSpan.FromMilliseconds(Math.Min((baseDelay.TotalMilliseconds * factor) + jitterMs, 10_000));
-    }
 
+        return TimeSpan.FromMilliseconds(Math.Min((baseDelay.TotalMilliseconds * factor) + jitterMs, MaxDelayMs));
+    }
 }

src/Extensions/AzureBlobPayloads/Converters/LargePayloadDataConverter.cs

@@ -37,14 +37,14 @@ public sealed class LargePayloadDataConverter(
     [return: NotNullIfNotNull("value")]
     public override string? Serialize(object? value)
     {
-        throw new NotImplementedException();
+        throw new NotSupportedException();
     }
 
     /// <inheritdoc/>
     [return: NotNullIfNotNull("data")]
     public override object? Deserialize(string? data, Type targetType)
     {
-        throw new NotImplementedException();
+        throw new NotSupportedException();
     }
 
     /// <summary>

src/Worker/Core/Shims/DurableTaskShimFactory.cs

@@ -50,7 +50,12 @@ public TaskActivity CreateActivity(TaskName name, ITaskActivity activity)
     {
         Check.NotDefault(name);
         Check.NotNull(activity);
-        return new TaskActivityShim(this.loggerFactory, this.options.DataConverter, name, activity, this.options.EnableLargePayloadSupport);
+        return new TaskActivityShim(
+            this.loggerFactory,
+            this.options.DataConverter,
+            name,
+            activity,
+            this.options.EnableLargePayloadSupport);
     }
 
     /// <summary>
@@ -151,6 +156,11 @@ public TaskEntity CreateEntity(TaskName name, ITaskEntity entity, EntityId entit
         // In the future we may consider caching those shims and reusing them, which can reduce
         // deserialization and allocation overheads.
         ILogger logger = this.loggerFactory.CreateLogger(entity.GetType());
-        return new TaskEntityShim(this.options.DataConverter, entity, entityId, logger, this.options.EnableLargePayloadSupport);
+        return new TaskEntityShim(
+            this.options.DataConverter,
+            entity,
+            entityId,
+            logger,
+            enableLargePayloadSupport: this.options.EnableLargePayloadSupport);
     }
 }