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

Author: YunchuWang

Directory.Build.props

@@ -7,6 +7,11 @@
     <ImplicitUsings>enable</ImplicitUsings>
   </PropertyGroup>
 
+  <PropertyGroup Condition="'$(TargetFramework)' == 'net6.0'">
+    <SuppressTfmSupportBuildWarnings>true</SuppressTfmSupportBuildWarnings>
+    <CheckEolTargetFramework>false</CheckEolTargetFramework>
+  </PropertyGroup>
+
   <!-- Assembly name and namespace -->
   <PropertyGroup Condition="!$(MSBuildProjectName.StartsWith('Microsoft.DurableTask'))">
     <TopLevelNamespace Condition="'$(TopLevelNamespace)' == ''">Microsoft.DurableTask</TopLevelNamespace>

samples/AzureFunctionsUnitTests/SampleUnitTests.cs

@@ -200,9 +200,9 @@ public TestResponse(FunctionContext functionContext) : base(functionContext)
     public class TestLogger : ILogger
     {
         // list of all logs emitted, for validation
-        public IList<string> CapturedLogs {get; set;} = new List<string>();
+        public IList<string> CapturedLogs { get; set; } = new List<string>();
 
-        public IDisposable BeginScope<TState>(TState state) => Mock.Of<IDisposable>();
+        public IDisposable BeginScope<TState>(TState state) where TState : notnull => Mock.Of<IDisposable>();
 
         public bool IsEnabled(LogLevel logLevel)
         {

samples/LargePayloadConsoleApp/Program.cs

@@ -27,7 +27,7 @@
     // Keep threshold small to force externalization for demo purposes
     opts.ExternalizeThresholdBytes = 1024; // 1KB
     opts.ConnectionString = builder.Configuration.GetValue<string>("DURABLETASK_STORAGE") ?? "UseDevelopmentStorage=true";
-    opts.ContainerName = builder.Configuration.GetValue<string>("DURABLETASK_PAYLOAD_CONTAINER");
+    opts.ContainerName = builder.Configuration.GetValue<string>("DURABLETASK_PAYLOAD_CONTAINER") ?? "payloads";
 });
 
 // 2) Configure Durable Task client

src/Client/Grpc/GrpcDurableTaskClient.cs

@@ -110,7 +110,7 @@ public override async Task<string> ScheduleNewOrchestrationInstanceAsync(
 
         DateTimeOffset? startAt = options?.StartAt;
         this.logger.SchedulingOrchestration(
-            request.InstanceId,
+            request.InstanceId ?? string.Empty,
             orchestratorName,
             sizeInBytes: request.Input != null ? Encoding.UTF8.GetByteCount(request.Input) : 0,
             startAt.GetValueOrDefault(DateTimeOffset.UtcNow));

src/Extensions/AzureBlobPayloads/Options/LargePayloadStorageOptions.cs

@@ -63,7 +63,6 @@ public LargePayloadStorageOptions(Uri accountUri, TokenCredential credential)
     /// Gets or sets the threshold in bytes at which payloads are externalized. Default is 900_000 bytes.
     /// Value must not exceed 1 MiB (1,048,576 bytes).
     /// </summary>
-
     public int ExternalizeThresholdBytes
     {
         get => this.externalizeThresholdBytes;

src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs

@@ -89,7 +89,7 @@ public override async Task<string> UploadAsync(string payLoad, CancellationToken
             // using MemoryStream payloadStream = new(payloadBuffer, writable: false);
 
             // await payloadStream.CopyToAsync(compressedBlobStream, bufferSize: DefaultCopyBufferSize, cancellationToken);
-            await compressedBlobStream.WriteAsync(payloadBuffer, 0, payloadBuffer.Length, cancellationToken);
+            await WritePayloadAsync(payloadBuffer, compressedBlobStream, cancellationToken);
             await compressedBlobStream.FlushAsync(cancellationToken);
             await blobStream.FlushAsync(cancellationToken);
         }
@@ -99,7 +99,7 @@ public override async Task<string> UploadAsync(string payLoad, CancellationToken
 
             // using MemoryStream payloadStream = new(payloadBuffer, writable: false);
             // await payloadStream.CopyToAsync(blobStream, bufferSize: DefaultCopyBufferSize, cancellationToken);
-            await blobStream.WriteAsync(payloadBuffer, 0, payloadBuffer.Length, cancellationToken);
+            await WritePayloadAsync(payloadBuffer, blobStream, cancellationToken);
             await blobStream.FlushAsync(cancellationToken);
         }
 
@@ -126,11 +126,11 @@ public override async Task<string> DownloadAsync(string token, CancellationToken
         {
             using GZipStream decompressed = new(contentStream, CompressionMode.Decompress);
             using StreamReader reader = new(decompressed, Encoding.UTF8);
-            return await reader.ReadToEndAsync();
+            return await ReadToEndAsync(reader, cancellationToken);
         }
 
         using StreamReader uncompressedReader = new(contentStream, Encoding.UTF8);
-        return await uncompressedReader.ReadToEndAsync();
+        return await ReadToEndAsync(uncompressedReader, cancellationToken);
     }
 
     /// <inheritdoc/>
@@ -144,6 +144,27 @@ public override bool IsKnownPayloadToken(string value)
         return value.StartsWith(TokenPrefix, StringComparison.Ordinal);
     }
 
+    static async Task WritePayloadAsync(byte[] payloadBuffer, Stream target, CancellationToken cancellationToken)
+    {
+#if NETSTANDARD2_0
+        await target.WriteAsync(payloadBuffer, 0, payloadBuffer.Length, cancellationToken).ConfigureAwait(false);
+#else
+        await target.WriteAsync(payloadBuffer.AsMemory(0, payloadBuffer.Length), cancellationToken).ConfigureAwait(false);
+#endif
+    }
+
+    static async Task<string> ReadToEndAsync(StreamReader reader, CancellationToken cancellationToken)
+    {
+#if NETSTANDARD2_0
+        cancellationToken.ThrowIfCancellationRequested();
+        return await reader.ReadToEndAsync().ConfigureAwait(false);
+#elif NET8_0_OR_GREATER
+        return await reader.ReadToEndAsync(cancellationToken).ConfigureAwait(false);
+#else
+        return await reader.ReadToEndAsync().WaitAsync(cancellationToken).ConfigureAwait(false);
+#endif
+    }
+
     static string EncodeToken(string container, string name) => $"blob:v1:{container}:{name}";
 
     static (string Container, string Name) DecodeToken(string token)

src/InProcessTestHost/DurableTaskTestHost.cs

@@ -22,7 +22,7 @@ namespace Microsoft.DurableTask.Testing;
 /// </summary>
 public sealed class DurableTaskTestHost : IAsyncDisposable
 {
-    readonly IWebHost sidecarHost;
+    readonly IHost sidecarHost;
     readonly IHost workerHost;
     readonly GrpcChannel grpcChannel;
 
@@ -33,7 +33,7 @@ public sealed class DurableTaskTestHost : IAsyncDisposable
     /// <param name="workerHost">The worker host.</param>
     /// <param name="grpcChannel">The gRPC channel.</param>
     /// <param name="client">The durable task client.</param>
-    public DurableTaskTestHost(IWebHost sidecarHost, IHost workerHost, GrpcChannel grpcChannel, DurableTaskClient client)
+    public DurableTaskTestHost(IHost sidecarHost, IHost workerHost, GrpcChannel grpcChannel, DurableTaskClient client)
     {
         this.sidecarHost = sidecarHost;
         this.workerHost = workerHost;
@@ -68,32 +68,37 @@ public static async Task<DurableTaskTestHost> StartAsync(
             ? $"http://localhost:{options.Port.Value}"
             : $"http://localhost:{Random.Shared.Next(30000, 40000)}";
 
-        var sidecarHost = new WebHostBuilder()
-            .UseKestrel(kestrelOptions =>
+        IHost sidecarHost = Host.CreateDefaultBuilder()
+            .ConfigureWebHostDefaults(webBuilder =>
             {
-                // Configure for HTTP/2 (required for gRPC)
-                kestrelOptions.ConfigureEndpointDefaults(listenOptions =>
-                    listenOptions.Protocols = HttpProtocols.Http2);
-            })
-            .UseUrls(address)
-            .ConfigureServices(services =>
-            {
-                services.AddGrpc();
-                services.AddSingleton<IOrchestrationService>(orchestrationService);
-                services.AddSingleton<IOrchestrationServiceClient>(orchestrationService);
-                services.AddSingleton<TaskHubGrpcServer>();
-            })
-            .Configure(app =>
-            {
-                app.UseRouting();
-                app.UseEndpoints(endpoints =>
+                webBuilder.UseUrls(address);
+                webBuilder.ConfigureKestrel(kestrelOptions =>
+                {
+                    // Configure for HTTP/2 (required for gRPC)
+                    kestrelOptions.ConfigureEndpointDefaults(listenOptions =>
+                        listenOptions.Protocols = HttpProtocols.Http2);
+                });
+
+                webBuilder.ConfigureServices(services =>
+                {
+                    services.AddGrpc();
+                    services.AddSingleton<IOrchestrationService>(orchestrationService);
+                    services.AddSingleton<IOrchestrationServiceClient>(orchestrationService);
+                    services.AddSingleton<TaskHubGrpcServer>();
+                });
+
+                webBuilder.Configure(app =>
                 {
-                    endpoints.MapGrpcService<TaskHubGrpcServer>();
+                    app.UseRouting();
+                    app.UseEndpoints(endpoints =>
+                    {
+                        endpoints.MapGrpcService<TaskHubGrpcServer>();
+                    });
                 });
             })
             .Build();
 
-        sidecarHost.Start();
+        await sidecarHost.StartAsync(cancellationToken);
         var grpcChannel = GrpcChannel.ForAddress(address);
 
         // Create worker host with user's orchestrators and activities

src/InProcessTestHost/Sidecar/Grpc/TaskHubGrpcServer.cs

@@ -715,7 +715,13 @@ public override async Task GetWorkItems(P.GetWorkItemsRequest request, IServerSt
         }
     }
 
-    /// <inheritdoc/>
+    /// <summary>
+    /// Streams the instance history for a given orchestration instance to the client in chunked form.
+    /// </summary>
+    /// <param name="request">The history request that identifies the instance.</param>
+    /// <param name="responseStream">The response stream used to write history chunks.</param>
+    /// <param name="context">The server call context for the streaming operation.</param>
+    /// <returns>A task that completes when streaming finishes.</returns>
     public override async Task StreamInstanceHistory(P.StreamInstanceHistoryRequest request, IServerStreamWriter<P.HistoryChunk> responseStream, ServerCallContext context)
     {
         if (this.streamingPastEvents.TryGetValue(request.InstanceId, out List<P.HistoryEvent>? pastEvents))

src/Shared/AzureManaged/DurableTaskVersionUtil.cs

@@ -3,6 +3,8 @@
 
 using System.Diagnostics;
 
+#pragma warning disable CS0436
+
 namespace Microsoft.DurableTask;
 
 /// <summary>
@@ -18,7 +20,7 @@ public static class DurableTaskUserAgentUtil
     /// <summary>
     /// The version of the SDK used in the user agent string.
     /// </summary>
-    static readonly string PackageVersion = FileVersionInfo.GetVersionInfo(typeof(DurableTaskUserAgentUtil).Assembly.Location).FileVersion;
+    static readonly string PackageVersion = FileVersionInfo.GetVersionInfo(typeof(DurableTaskUserAgentUtil).Assembly.Location).FileVersion ?? "unknown";
 
     /// <summary>
     /// Generates the user agent string for the Durable Task SDK based on a fixed name, the package version, and the caller type.
@@ -29,4 +31,5 @@ public static string GetUserAgent(string callerType)
     {
         return $"{SdkName}/{PackageVersion?.ToString() ?? "unknown"} ({callerType})";
     }
-}
+}
+#pragma warning restore CS0436

src/Worker/Grpc/GrpcDurableTaskWorkerOptions.cs

@@ -62,7 +62,7 @@ public int CompleteOrchestrationWorkItemChunkSizeInBytes
             if (value < MinCompleteOrchestrationWorkItemChunkSizeInBytes ||
                 value > MaxCompleteOrchestrationWorkItemChunkSizeInBytes)
             {
-                string message = $"CompleteOrchestrationWorkItemChunkSizeInBytes must be between " +
+                string message = $"{nameof(CompleteOrchestrationWorkItemChunkSizeInBytes)} must be between " +
                     $"{MinCompleteOrchestrationWorkItemChunkSizeInBytes} bytes (1 MB) and " +
                     $"{MaxCompleteOrchestrationWorkItemChunkSizeInBytes} bytes (3.9 MB), inclusive.";
                 throw new ArgumentOutOfRangeException(