Prevent block inside ResolveAsync from blocking PollingResolver.Refresh

src/Grpc.Net.Client/Balancer/PollingResolver.cs

@@ -135,7 +135,9 @@ public sealed override void Refresh()
 
             if (_resolveTask.IsCompleted)
             {
-                _resolveTask = ResolveNowAsync(_cts.Token);
+                // Run ResolveAsync in a background task.
+                // This is done to prevent synchronous block inside ResolveAsync from blocking future Refresh calls.
+                _resolveTask = Task.Run(() => ResolveNowAsync(_cts.Token), _cts.Token);
                 _resolveTask.ContinueWith(static (t, state) =>
                 {
                     var pollingResolver = (PollingResolver)state!;

test/Grpc.Net.Client.Tests/Balancer/ConnectionManagerTests.cs

@@ -409,6 +409,7 @@ public async Task PickAsync_DoesNotDeadlockAfterReconnect_WithResolverError()
         services.AddNUnitLogger();
         await using var serviceProvider = services.BuildServiceProvider();
         var loggerFactory = serviceProvider.GetRequiredService<ILoggerFactory>();
+        var logger = loggerFactory.CreateLogger(GetType());
 
         var resolver = new TestResolver(loggerFactory);
 
@@ -427,22 +428,29 @@ public async Task PickAsync_DoesNotDeadlockAfterReconnect_WithResolverError()
             clientChannel));
 
         // Act
+        logger.LogInformation("Client connecting.");
         var connectTask = clientChannel.ConnectAsync(waitForReady: true, cancellationToken: CancellationToken.None);
+
+        logger.LogInformation("Starting pick on connecting channel.");
         var pickTask = clientChannel.PickAsync(
             new PickContext { Request = new HttpRequestMessage() },
             waitForReady: true,
             CancellationToken.None).AsTask();
 
+        logger.LogInformation("Waiting for resolve to complete.");
+        await resolver.HasResolvedTask.DefaultTimeout();
+
         resolver.UpdateAddresses(new List<BalancerAddress>
         {
             new BalancerAddress("localhost", 80)
         });
         await Task.WhenAll(connectTask, pickTask).DefaultTimeout();
 
-        // Simulate transport/network issue
+        logger.LogInformation("Simulate transport/network issue.");
         transportFactory.Transports.ForEach(t => t.Disconnect());
         resolver.UpdateError(new Status(StatusCode.Unavailable, "Test error"));
 
+        logger.LogInformation("Starting pick on disconnected channel.");
         pickTask = clientChannel.PickAsync(
             new PickContext { Request = new HttpRequestMessage() },
             waitForReady: true,
@@ -454,7 +462,10 @@ public async Task PickAsync_DoesNotDeadlockAfterReconnect_WithResolverError()
 
         // Assert
         // Should not timeout (deadlock)
+        logger.LogInformation("Wait for pick task to complete.");
         await pickTask.DefaultTimeout();
+
+        logger.LogInformation("Done.");
     }
 
     [Test]
@@ -489,6 +500,8 @@ public async Task PickAsync_DoesNotDeadlockAfterReconnect_WithZeroAddressResolve
             waitForReady: true,
             CancellationToken.None).AsTask();
 
+        await resolver.HasResolvedTask.DefaultTimeout();
+
         resolver.UpdateAddresses(new List<BalancerAddress>
         {
             new BalancerAddress("localhost", 80)
@@ -560,6 +573,8 @@ public async Task PickAsync_ExecutionContext_DoesNotCaptureAsyncLocalsInConnect(
             waitForReady: true,
             CancellationToken.None).AsTask();
 
+        await resolver.HasResolvedTask.DefaultTimeout();
+
         resolver.UpdateAddresses(new List<BalancerAddress>
         {
             new BalancerAddress("localhost", 80)

test/Grpc.Net.Client.Tests/Balancer/PickFirstBalancerTests.cs

@@ -196,6 +196,8 @@ public async Task ResolverError_HasFailedSubchannel_SubchannelShutdown()
         _ = channel.ConnectAsync();
 
         // Assert
+        await resolver.HasResolvedTask.DefaultTimeout();
+
         var subchannels = channel.ConnectionManager.GetSubchannels();
         Assert.AreEqual(1, subchannels.Count);
 

test/Grpc.Net.Client.Tests/Balancer/ResolverTests.cs

@@ -42,6 +42,71 @@ namespace Grpc.Net.Client.Tests.Balancer;
 [TestFixture]
 public class ResolverTests
 {
+    [Test]
+    public async Task Refresh_BlockInsideResolveAsync_ResolverNotBlocked()
+    {
+        var waitHandle = new ManualResetEvent(false);
+
+        var services = new ServiceCollection();
+        var testSink = new TestSink();
+        services.AddLogging(b =>
+        {
+            b.AddProvider(new TestLoggerProvider(testSink));
+        });
+        services.AddNUnitLogger();
+        var loggerFactory = services.BuildServiceProvider().GetRequiredService<ILoggerFactory>();
+
+        var logger = loggerFactory.CreateLogger<ResolverTests>();
+        logger.LogInformation("Starting.");
+
+        var ss = new LockingPollingResolver(loggerFactory, waitHandle);
+        ss.Start(result => { });
+
+        logger.LogInformation("Refresh call 1. This should block.");
+        var refreshTask1 = Task.Run(ss.Refresh);
+
+        logger.LogInformation("Refresh call 2. This should complete.");
+        var refreshTask2 = Task.Run(ss.Refresh);
+
+        await Task.WhenAny(refreshTask1, refreshTask2);
+
+        logger.LogInformation("Setting wait handle.");
+        waitHandle.Set();
+
+        logger.LogInformation("Finishing.");
+    }
+
+    private class LockingPollingResolver : PollingResolver
+    {
+        private ManualResetEvent? _waitHandle;
+        private readonly object _lock = new();
+
+        public LockingPollingResolver(ILoggerFactory loggerFactory, ManualResetEvent waitHandle) : base(loggerFactory)
+        {
+            _waitHandle = waitHandle;
+        }
+
+        protected override Task ResolveAsync(CancellationToken cancellationToken)
+        {
+            lock (_lock)
+            {
+                // Block the first caller.
+                if (_waitHandle != null)
+                {
+                    _waitHandle.WaitOne();
+                    _waitHandle = null;
+                }
+
+                Listener(ResolverResult.ForResult(new List<BalancerAddress>
+                {
+                    new BalancerAddress("localhost", 80)
+                }));
+
+                return Task.CompletedTask;
+            }
+        }
+    }
+
     [Test]
     public async Task Resolver_ResolveNameFromServices_Success()
     {

test/Grpc.Net.Client.Tests/Balancer/RoundRobinBalancerTests.cs

@@ -187,7 +187,10 @@ public async Task ResolverError_HasFailedSubchannel_SubchannelShutdown()
         _ = channel.ConnectAsync();
 
         // Assert
+        await resolver.HasResolvedTask.DefaultTimeout();
+
         var subchannels = channel.ConnectionManager.GetSubchannels();
+
         Assert.AreEqual(1, subchannels.Count);
 
         Assert.AreEqual(1, subchannels[0]._addresses.Count);

test/Grpc.Net.Client.Tests/Balancer/WaitForReadyTests.cs

@@ -58,21 +58,26 @@ public async Task ResolverReturnsNoAddresses_CallWithWaitForReady_Wait()
         });
 
         var services = new ServiceCollection();
+        services.AddNUnitLogger();
 
-        var resolver = new TestResolver();
-
-        services.AddSingleton<ResolverFactory>(new TestResolverFactory(resolver));
+        services.AddSingleton<TestResolver>();
+        services.AddSingleton<ResolverFactory>(s => new TestResolverFactory(s.GetRequiredService<TestResolver>()));
         services.AddSingleton<ISubchannelTransportFactory>(new TestSubchannelTransportFactory());
+        var serviceProvider = services.BuildServiceProvider();
+
+        var resolver = serviceProvider.GetRequiredService<TestResolver>();
 
         var invoker = HttpClientCallInvokerFactory.Create(testMessageHandler, "test:///localhost", configure: o =>
         {
             o.Credentials = ChannelCredentials.Insecure;
-            o.ServiceProvider = services.BuildServiceProvider();
+            o.ServiceProvider = serviceProvider;
         });
 
         // Act
         var call = invoker.AsyncUnaryCall<HelloRequest, HelloReply>(ClientTestHelpers.ServiceMethod, string.Empty, new CallOptions().WithWaitForReady(), new HelloRequest());
 
+        await resolver.HasResolvedTask.DefaultTimeout();
+
         var responseTask = call.ResponseAsync;
 
         Assert.IsFalse(responseTask.IsCompleted);

test/Shared/TestResolver.cs

@@ -1,4 +1,4 @@
-#region Copyright notice and license
+#region Copyright notice and license
 
 // Copyright 2019 The gRPC Authors
 //
@@ -34,6 +34,7 @@ internal class TestResolver : PollingResolver
 {
     private readonly Func<Task>? _onRefreshAsync;
     private readonly TaskCompletionSource<object?> _hasResolvedTcs;
+    private readonly ILogger _logger;
     private ResolverResult? _result;
 
     public Task HasResolvedTask => _hasResolvedTcs.Task;
@@ -46,15 +47,18 @@ public TestResolver(ILoggerFactory? loggerFactory = null, Func<Task>? onRefreshA
     {
         _onRefreshAsync = onRefreshAsync;
         _hasResolvedTcs = new TaskCompletionSource<object?>(TaskCreationOptions.RunContinuationsAsynchronously);
+        _logger = (ILogger?)loggerFactory?.CreateLogger<TestResolver>() ?? NullLogger.Instance;
     }
 
     public void UpdateAddresses(List<BalancerAddress> addresses, ServiceConfig? serviceConfig = null, Status? serviceConfigStatus = null)
     {
+        _logger.LogInformation("Updating result addresses: {Addresses}", string.Join(", ", addresses));
         UpdateResult(ResolverResult.ForResult(addresses, serviceConfig, serviceConfigStatus));
     }
 
     public void UpdateError(Status status)
     {
+        _logger.LogInformation("Updating result error: {Status}", status);
         UpdateResult(ResolverResult.ForFailure(status));
     }