Fix GrpcChannel handle leak in AzureManaged backend

src/Client/AzureManaged/DurableTaskSchedulerClientExtensions.cs

@@ -1,7 +1,10 @@
 // Copyright (c) Microsoft Corporation.
 // Licensed under the MIT License.
 
+using System.Collections.Concurrent;
+using System.Linq;
 using Azure.Core;
+using Grpc.Net.Client;
 using Microsoft.DurableTask.Client.Grpc;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.DependencyInjection.Extensions;
@@ -99,11 +102,23 @@ static void ConfigureSchedulerOptions(
     /// <summary>
     /// Configuration class that sets up gRPC channels for client options
     /// using the provided Durable Task Scheduler options.
+    /// Channels are cached per configuration key and disposed when the service provider is disposed.
     /// </summary>
-    /// <param name="schedulerOptions">Monitor for accessing the current scheduler options configuration.</param>
-    class ConfigureGrpcChannel(IOptionsMonitor<DurableTaskSchedulerClientOptions> schedulerOptions) :
-        IConfigureNamedOptions<GrpcDurableTaskClientOptions>
+    sealed class ConfigureGrpcChannel : IConfigureNamedOptions<GrpcDurableTaskClientOptions>, IAsyncDisposable
     {
+        readonly IOptionsMonitor<DurableTaskSchedulerClientOptions> schedulerOptions;
+        readonly ConcurrentDictionary<string, Lazy<GrpcChannel>> channels = new();
+        int disposed;
+
+        /// <summary>
+        /// Initializes a new instance of the <see cref="ConfigureGrpcChannel"/> class.
+        /// </summary>
+        /// <param name="schedulerOptions">Monitor for accessing the current scheduler options configuration.</param>
+        public ConfigureGrpcChannel(IOptionsMonitor<DurableTaskSchedulerClientOptions> schedulerOptions)
+        {
+            this.schedulerOptions = schedulerOptions;
+        }
+
         /// <summary>
         /// Configures the default named options instance.
         /// </summary>
@@ -117,8 +132,77 @@ class ConfigureGrpcChannel(IOptionsMonitor<DurableTaskSchedulerClientOptions> sc
         /// <param name="options">The options instance to configure.</param>
         public void Configure(string? name, GrpcDurableTaskClientOptions options)
         {
-            DurableTaskSchedulerClientOptions source = schedulerOptions.Get(name ?? Options.DefaultName);
-            options.Channel = source.CreateChannel();
+#if NET7_0_OR_GREATER
+            ObjectDisposedException.ThrowIf(this.disposed == 1, this);
+#else
+            if (this.disposed == 1)
+            {
+                throw new ObjectDisposedException(nameof(ConfigureGrpcChannel));
+            }
+#endif
+
+            string optionsName = name ?? Options.DefaultName;
+            DurableTaskSchedulerClientOptions source = this.schedulerOptions.Get(optionsName);
+
+            // Create a cache key that includes all properties that affect CreateChannel behavior.
+            // This ensures channels are reused for the same configuration
+            // but separate channels are created when any relevant property changes.
+            // Use a delimiter character (\u001F) that will not appear in typical endpoint URIs.
+            string credentialType = source.Credential?.GetType().FullName ?? "null";
+            string retryOptionsKey = source.RetryOptions != null
+                ? $"{source.RetryOptions.MaxRetries}|{source.RetryOptions.InitialBackoffMs}|{source.RetryOptions.MaxBackoffMs}|{source.RetryOptions.BackoffMultiplier}|{(source.RetryOptions.RetryableStatusCodes != null ? string.Join(",", source.RetryOptions.RetryableStatusCodes) : string.Empty)}"
+                : "null";
+            string cacheKey = $"{optionsName}\u001F{source.EndpointAddress}\u001F{source.TaskHubName}\u001F{source.ResourceId}\u001F{credentialType}\u001F{source.AllowInsecureCredentials}\u001F{retryOptionsKey}";
+            options.Channel = this.channels.GetOrAdd(
+                cacheKey,
+                _ => new Lazy<GrpcChannel>(source.CreateChannel)).Value;
+        }
+
+        /// <inheritdoc/>
+        public async ValueTask DisposeAsync()
+        {
+            if (Interlocked.Exchange(ref this.disposed, 1) == 1)
+            {
+                return;
+            }
+
+            List<Exception>? exceptions = null;
+            foreach (Lazy<GrpcChannel> channel in this.channels.Values.Where(lazy => lazy.IsValueCreated))
+            {
+                try
+                {
+                    await DisposeChannelAsync(channel.Value).ConfigureAwait(false);
+                }
+                catch (Exception ex)
+                {
+                    exceptions ??= new List<Exception>();
+                    exceptions.Add(ex);
+                }
+            }
+
+            this.channels.Clear();
+            GC.SuppressFinalize(this);
+
+            if (exceptions is { Count: > 0 })
+            {
+                throw new AggregateException(exceptions);
+            }
+        }
+
+        static async Task DisposeChannelAsync(GrpcChannel channel)
+        {
+            // ShutdownAsync is the graceful way to close a gRPC channel.
+            using (channel)
+            {
+                try
+                {
+                    await channel.ShutdownAsync().ConfigureAwait(false);
+                }
+                catch (Exception ex) when (ex is OperationCanceledException or ObjectDisposedException)
+                {
+                    // Ignore expected shutdown/disposal errors
+                }
+            }
         }
     }
 }

src/Worker/AzureManaged/DurableTaskSchedulerWorkerExtensions.cs

@@ -1,7 +1,10 @@
 // Copyright (c) Microsoft Corporation.
 // Licensed under the MIT License.
 
+using System.Collections.Concurrent;
+using System.Linq;
 using Azure.Core;
+using Grpc.Net.Client;
 using Microsoft.DurableTask.Worker.Grpc;
 using Microsoft.DurableTask.Worker.Grpc.Internal;
 using Microsoft.Extensions.DependencyInjection;
@@ -101,11 +104,23 @@ static void ConfigureSchedulerOptions(
     /// <summary>
     /// Configuration class that sets up gRPC channels for worker options
     /// using the provided Durable Task Scheduler options.
+    /// Channels are cached per configuration key and disposed when the service provider is disposed.
     /// </summary>
-    /// <param name="schedulerOptions">Monitor for accessing the current scheduler options configuration.</param>
-    class ConfigureGrpcChannel(IOptionsMonitor<DurableTaskSchedulerWorkerOptions> schedulerOptions) :
-        IConfigureNamedOptions<GrpcDurableTaskWorkerOptions>
+    sealed class ConfigureGrpcChannel : IConfigureNamedOptions<GrpcDurableTaskWorkerOptions>, IAsyncDisposable
     {
+        readonly IOptionsMonitor<DurableTaskSchedulerWorkerOptions> schedulerOptions;
+        readonly ConcurrentDictionary<string, Lazy<GrpcChannel>> channels = new();
+        int disposed;
+
+        /// <summary>
+        /// Initializes a new instance of the <see cref="ConfigureGrpcChannel"/> class.
+        /// </summary>
+        /// <param name="schedulerOptions">Monitor for accessing the current scheduler options configuration.</param>
+        public ConfigureGrpcChannel(IOptionsMonitor<DurableTaskSchedulerWorkerOptions> schedulerOptions)
+        {
+            this.schedulerOptions = schedulerOptions;
+        }
+
         /// <summary>
         /// Configures the default named options instance.
         /// </summary>
@@ -119,9 +134,75 @@ class ConfigureGrpcChannel(IOptionsMonitor<DurableTaskSchedulerWorkerOptions> sc
         /// <param name="options">The options instance to configure.</param>
         public void Configure(string? name, GrpcDurableTaskWorkerOptions options)
         {
-            DurableTaskSchedulerWorkerOptions source = schedulerOptions.Get(name ?? Options.DefaultName);
-            options.Channel = source.CreateChannel();
+#if NET7_0_OR_GREATER
+            ObjectDisposedException.ThrowIf(this.disposed == 1, this);
+#else
+            if (this.disposed == 1)
+            {
+                throw new ObjectDisposedException(nameof(ConfigureGrpcChannel));
+            }
+#endif
+
+            string optionsName = name ?? Options.DefaultName;
+            DurableTaskSchedulerWorkerOptions source = this.schedulerOptions.Get(optionsName);
+
+            // Create a cache key that includes all properties that affect CreateChannel behavior.
+            // This ensures channels are reused for the same configuration
+            // but separate channels are created when any relevant property changes.
+            // Use a delimiter character (\u001F) that will not appear in typical endpoint URIs.
+            string credentialType = source.Credential?.GetType().FullName ?? "null";
+            string cacheKey = $"{optionsName}\u001F{source.EndpointAddress}\u001F{source.TaskHubName}\u001F{source.ResourceId}\u001F{credentialType}\u001F{source.AllowInsecureCredentials}\u001F{source.WorkerId}";
+            options.Channel = this.channels.GetOrAdd(
+                cacheKey,
+                _ => new Lazy<GrpcChannel>(source.CreateChannel)).Value;
             options.ConfigureForAzureManaged();
         }
+
+        /// <inheritdoc/>
+        public async ValueTask DisposeAsync()
+        {
+            if (Interlocked.Exchange(ref this.disposed, 1) == 1)
+            {
+                return;
+            }
+
+            List<Exception>? exceptions = null;
+            foreach (Lazy<GrpcChannel> channel in this.channels.Values.Where(lazy => lazy.IsValueCreated))
+            {
+                try
+                {
+                    await DisposeChannelAsync(channel.Value).ConfigureAwait(false);
+                }
+                catch (Exception ex) when (ex is not OperationCanceledException and not ObjectDisposedException)
+                {
+                    exceptions ??= new List<Exception>();
+                    exceptions.Add(ex);
+                }
+            }
+
+            this.channels.Clear();
+            GC.SuppressFinalize(this);
+
+            if (exceptions is { Count: > 0 })
+            {
+                throw new AggregateException(exceptions);
+            }
+        }
+
+        static async Task DisposeChannelAsync(GrpcChannel channel)
+        {
+            // ShutdownAsync is the graceful way to close a gRPC channel.
+            using (channel)
+            {
+                try
+                {
+                    await channel.ShutdownAsync().ConfigureAwait(false);
+                }
+                catch (Exception ex) when (ex is OperationCanceledException or ObjectDisposedException)
+                {
+                    // Ignore expected shutdown/disposal errors
+                }
+            }
+        }
     }
 }