Repository: grpc/grpc-dotnet (csharp)

Issue:
TaskCanceledException when reconnecting after disconnect
<!--

This form is for bug reports and feature requests ONLY!
For general questions and troubleshooting, please ask/look for answers here:
- Documentation: https://docs.microsoft.com/aspnet/core/grpc
- Troubleshooting: https://docs.microsoft.com/aspnet/core/grpc/troubleshoot
- StackOverflow, with "grpc" tag: https://stackoverflow.com/questions/tagged/grpc

Issues specific to Grpc.Core (C-core server and client), or code generation with Grpc.Tools should be created at https://github.com/grpc/grpc/issues/new
-->

### What version of gRPC and what language are you using?
Grpc.Net.Client 2.70.0, C#


### What operating system (Linux, Windows,...) and version?
Debian (Docker image mcr.microsoft.com/dotnet/aspnet:8.0)

### What runtime / compiler are you using (e.g. .NET Core SDK version `dotnet --info`)
Docker image mcr.microsoft.com/dotnet/sdk:8.0

### What did you do?
If possible, provide a recipe for reproducing the error. Try being specific and include code snippets if helpful.

### Application setup
Our GrpcClient is registered simply as such using `Grpc.Net.ClientFactory` version `2.66`:
```
 services.AddGrpcClient<Client>(o =>
        {
            o.Address = new Uri(<url>);
        });
```

### Issue description

Our application is having intermittent RPC errors with exception `System.Threading.Tasks.TaskCanceledException: A task was canceled.`. Investigating the logs it appears to happen when requests are picking a subchannel that has previously disconnected.

Full log example:
```
2025-03-17 17:01:24.684	Req X: Successfully picked subchannel id '"1-1"' with address "authorization-api.opa:8081". Transport status: ActiveStream
2025-03-17 17:01:26.777	 Subchannel id '"1-1"' state changed to Idle. Detail: '"Disconnected."'.
2025-03-17 17:01:26.777	 Channel state updated to Idle.
2025-03-17 17:01:26.777	 Channel picker updated.
2025-03-17 17:02:07.983	Req1: Picked subchannel id '"1-1"' doesn't have a current address.
2025-03-17 17:02:07.983	Req2: Subchannel id '"1-1"' state changed to Connecting. Detail: '"Connection requested."'.
2025-03-17 17:02:07.983	Req2: Channel state updated to Connecting.
2025-03-17 17:02:07.983	Req2: Channel picker updated.
2025-03-17 17:02:07.983	Req2: Picked subchannel id '"1-1"' doesn't have a current address.
2025-03-17 17:02:07.983	Req3: Picked queued.
2025-03-17 17:02:07.983	Req1: Picked queued.
2025-03-17 17:02:07.983	Req2: Picked queued.
2025-03-17 17:02:07.983	 Subchannel id '"1-1"' connecting to transport.
2025-03-17 17:02:07.984	 Subchannel id '"1-1"' canceling connect.
2025-03-17 17:02:07.984	 Subchannel id '"1-1"' queuing connect because a connect is already in progress.
2025-03-17 17:02:07.985	 Subchannel id '"1-1"' error connecting to socket "authorization-api.opa:8081".
2025-03-17 17:02:07.985	 Subchannel id '"1-1"' state changed to TransientFailure. Detail: '"Error connecting to subchannel."'.
2025-03-17 17:02:07.985	 Channel state updated to TransientFailure.
2025-03-17 17:02:07.985	 Channel picker updated.
2025-03-17 17:02:07.985	 Subchannel id '"1-1"' connect canceled.
2025-03-17 17:02:07.985	 Subchannel id '"1-1"' connecting to transport.
2025-03-17 17:02:07.985	 Subchannel id '"1-1"' state changed to Connecting. Detail: '"Connecting to socket."'.
2025-03-17 17:02:07.985	 Channel state updated to Connecting.
2025-03-17 17:02:07.985	 Channel picker updated.
2025-03-17 17:02:07.988	 Subchannel id '"1-1"' connected to socket "authorization-api.opa:8081".
2025-03-17 17:02:07.988	Req3: Picked failure with status: "Status(StatusCode=\"Unavailable\", Detail=\"Error connecting to subchannel.\", DebugException=\"System.Threading.Tasks.TaskCanceledException: A task was canceled.\")"
2025-03-17 17:02:07.988	Req2: Picked failure with status: "Status(StatusCode=\"Unavailable\", Detail=\"Error connecting to subchannel.\", DebugException=\"System.Threading.Tasks.TaskCanceledException: A task was canceled.\")"
2025-03-17 17:02:07.988	Req1: Picked failure with status: "Status(StatusCode=\"Unavailable\", Detail=\"Error connecting to subchannel.\", DebugException=\"System.Threading.Tasks.TaskCanceledException: A task was canceled.\")"
2025-03-17 17:02:07.988	 Subchannel id '"1-1"' state changed to Ready. Detail: '"Successfully connected to socket."'.
2025-03-17 17:02:07.988	 Channel state updated to Ready.
2025-03-17 17:02:07.988	 Channel picker updated.
2025-03-17 17:02:07.993	Req3: Call failed with gRPC error status. Status code: 'Unavailable', Message: '"Error connecting to subchannel."'.
2025-03-17 17:02:07.993	Req2: Call failed with gRPC error status. Status code: 'Unavailable', Message: '"Error connecting to subchannel."'.
2025-03-17 17:02:07.993	Req1: Call failed with gRPC error status. Status code: 'Unavailable', Message: '"Error connecting to subchannel."'.
2025-03-17 17:02:07.995	Req1: SAS Authorization failed because there's been a RpcException.
2025-03-17 17:02:07.995	Req3: SAS Authorization failed because there's been a RpcException.
2025-03-17 17:02:07.995	Req2: SAS Authorization failed because there's been a RpcException.

```

I've marked logs logged from separate HTTP contexts as `Req1` `Req2` and `Req3`.

Having briefly looked at the source code corresponding to the logs, my understanding of what is happening is:

```
2025-03-17 17:01:24.684	Req X: Successfully picked subchannel id '"1-1"' with address "authorization-api.opa:8081". Transport status: ActiveStream
2025-03-17 17:01:26.777	 Subchannel id '"1-1"' state changed to Idle. Detail: '"Disconnected."'.
2025-03-17 17:01:26.777	 Channel state updated to Idle.
```
The subchannel was connected but disconnects for whatever reason. This is fine.

```
2025-03-17 17:02:07.983	Req1: Picked subchannel id '"1-1"' doesn't have a current address.
2025-03-17 17:02:07.983	Req2: Subchannel id '"1-1"' state changed to Connecting. Detail: '"Connection requested."'.
2025-03-17 17:02:07.983	Req2: Channel state updated to Connecting.
2025-03-17 17:02:07.983	Req2: Channel picker updated.
2025-03-17 17:02:07.983	Req2: Picked subchannel id '"1-1"' doesn't have a current address.
2025-03-17 17:02:07.983	Req3: Picked queued.
2025-03-17 17:02:07.983	Req1: Picked queued.
2025-03-17 17:02:07.983	Req2: Picked queued.
2025-03-17 17:02:07.983	 Subchannel id '"1-1"' connecting to transport.
```
One of the three requests requests a new connection on the subchannel. All 3 requests queue up for the picker.

```
2025-03-17 17:02:07.983	 Subchannel id '"1-1"' connecting to transport.
2025-03-17 17:02:07.984	 Subchannel id '"1-1"' canceling connect.
2025-03-17 17:02:07.984	 Subchannel id '"1-1"' queuing connect because a connect is already in progress.
2025-03-17 17:02:07.985	 Subchannel id '"1-1"' error connecting to socket "authorization-api.opa:8081".
2025-03-17 17:02:07.985	 Subchannel id '"1-1"' state changed to TransientFailure. Detail: '"Error connecting to subchannel."'.
2025-03-17 17:02:07.985	 Channel state updated to TransientFailure.
2025-03-17 17:02:07.985	 Channel picker updated.
2025-03-17 17:02:07.985	 Subchannel id '"1-1"' connect canceled.
2025-03-17 17:02:07.985	 Subchannel id '"1-1"' connecting to transport.
2025-03-17 17:02:07.985	 Subchannel id '"1-1"' state changed to Connecting. Detail: '"Connecting to socket."'.
2025-03-17 17:02:07.985	 Channel state updated to Connecting.
2025-03-17 17:02:07.985	 Channel picker updated.
2025-03-17 17:02:07.988	 Subchannel id '"1-1"' connected to socket "authorization-api.opa:8081".
2025-03-17 17:02:07.988	Req3: Picked failure with status: "Status(StatusCode=\"Unavailable\", Detail=\"Error connecting to subchannel.\", DebugException=\"System.Threading.Tasks.TaskCanceledException: A task was canceled.\")"
2025-03-17 17:02:07.988	Req2: Picked failure with status: "Status(StatusCode=\"Unavailable\", Detail=\"Error connecting to subchannel.\", DebugException=\"System.Threading.Tasks.TaskCanceledException: A task was canceled.\")"
2025-03-17 17:02:07.988	Req1: Picked failure with status: "Status(StatusCode=\"Unavailable\", Detail=\"Error connecting to subchannel.\", DebugException=\"System.Threading.Tasks.TaskCanceledException: A task was canceled.\")"
```

The second line shows a connection being cancelled, followed by `Channel state updated to TransientFailure.` and `Channel picker updated.`. I believe this is how the `TaskCanceledException` error is propagated to the 3 enqueued requests. But why is the connection cancelled? My best guess is [this line](https://github.com/grpc/grpc-dotnet/blob/master/src/Grpc.Net.Client/Balancer/Subchannel.cs#L312) shows the previous connection is always cancelled, even if it is in an idle/disconnected state, which is causing the queued requests to fail.

I didn't spend a lot of time studying the code, so apologies if I am misunderstanding. I believe it is not intentional for these requests to fail with `TaskCanceledException`. I know we can set the `WaitForReady` CallOptions flag to retry on connection failure, but I don't believe it should be necessary since the subchannel canceling an idle connection doesn't constitute a failure in my eyes.

An additional observation: Every instance of this error in our logs appeared in two or more errors happening at the same millisecond. There were no single errors like this. That smells like a race condition to me

Pull request description:
Fixes https://github.com/grpc/grpc-dotnet/issues/2612

A previous fix to stop deadlocks - https://github.com/grpc/grpc-dotnet/pull/2601 - caused transient request errors. If a subcannel's not connected, and multiple gRPC calls queue at the same time, it's possible for a connection to be rapidly started and canceled. Although the state of the subchannel is quickly resolved, the queued calls could get an error result.

The fix is to check if a connection request is in progress later in the connection pipeline. If there is already a request then interrupt the delay if it is waiting and exit.

Patch:
diff --git a/src/Grpc.Net.Client/Balancer/Subchannel.cs b/src/Grpc.Net.Client/Balancer/Subchannel.cs
index d80d5adeb..638876aa8 100644
--- a/src/Grpc.Net.Client/Balancer/Subchannel.cs
+++ b/src/Grpc.Net.Client/Balancer/Subchannel.cs
@@ -266,10 +266,10 @@ public void RequestConnection()
             }
         }
 
-        if (connectionRequested)
-        {
-            UpdateConnectivityState(ConnectivityState.Connecting, "Connection requested.");
-        }
+        Debug.Assert(connectionRequested, "Ensure that only expected state made it to this point.");
+
+        SubchannelLog.StartingConnectionRequest(_logger, Id);
+        UpdateConnectivityState(ConnectivityState.Connecting, "Connection requested.");
 
         // Don't capture the current ExecutionContext and its AsyncLocals onto the connect
         var restoreFlow = false;
@@ -327,6 +327,15 @@ private async Task ConnectTransportAsync()
                 return;
             }
 
+            // There is already a connect in-progress on this transport.
+            // Don't cancel and start again as that causes queued requests waiting on the connect to fail.
+            if (_connectContext != null && !_connectContext.Disposed)
+            {
+                SubchannelLog.ConnectionRequestedInNonIdleState(_logger, Id, _state);
+                _delayInterruptTcs?.TrySetResult(null);
+                return;
+            }
+
             connectContext = GetConnectContextUnsynchronized();
 
             // Use a semaphore to limit one connection attempt at a time. This is done to prevent a race conditional where a canceled connect
@@ -633,7 +642,11 @@ public static void AddressesUpdated(ILogger logger, string subchannelId, IReadOn
             AddressesUpdated(logger, subchannelId, addressesText);
         }
     }
+
     [LoggerMessage(Level = LogLevel.Debug, EventId = 20, EventName = "QueuingConnect", Message = "Subchannel id '{SubchannelId}' queuing connect because a connect is already in progress.")]
     public static partial void QueuingConnect(ILogger logger, string subchannelId);
+
+    [LoggerMessage(Level = LogLevel.Trace, EventId = 21, EventName = "StartingConnectionRequest", Message = "Subchannel id '{SubchannelId}' starting connection request.")]
+    public static partial void StartingConnectionRequest(ILogger logger, string subchannelId);
 }
 #endif


Test patch:
diff --git a/test/Grpc.Net.Client.Tests/Balancer/ConnectionManagerTests.cs b/test/Grpc.Net.Client.Tests/Balancer/ConnectionManagerTests.cs
index 7a5221728..7e807aeb8 100644
--- a/test/Grpc.Net.Client.Tests/Balancer/ConnectionManagerTests.cs
+++ b/test/Grpc.Net.Client.Tests/Balancer/ConnectionManagerTests.cs
@@ -620,6 +620,128 @@ public async Task PickAsync_UpdateAddressesWhileRequestingConnection_DoesNotDead
         }
     }
 
+    [Test]
+    public async Task PickAsync_MultipleRequestsRequestConnect_SingleConnectAttempt()
+    {
+        var services = new ServiceCollection();
+        services.AddNUnitLogger();
+
+        var testSink = new TestSink();
+        var testProvider = new TestLoggerProvider(testSink);
+
+        services.AddLogging(b =>
+        {
+            b.AddProvider(testProvider);
+        });
+
+        await using var serviceProvider = services.BuildServiceProvider();
+        var loggerFactory = serviceProvider.GetRequiredService<ILoggerFactory>();
+        var logger = loggerFactory.CreateLogger(nameof(PickAsync_MultipleRequestsRequestConnect_SingleConnectAttempt));
+
+        var requestConnectionChannel = Channel.CreateUnbounded<SyncPoint>();
+        var requestConnectionSyncPoint1 = new SyncPoint(runContinuationsAsynchronously: true);
+        var requestConnectionSyncPoint2 = new SyncPoint(runContinuationsAsynchronously: true);
+        requestConnectionChannel.Writer.TryWrite(requestConnectionSyncPoint1);
+        requestConnectionChannel.Writer.TryWrite(requestConnectionSyncPoint2);
+
+        var connectingSyncPoint = new SyncPoint(runContinuationsAsynchronously: true);
+
+        var resolver = new TestResolver(loggerFactory);
+        resolver.UpdateAddresses(new List<BalancerAddress>
+        {
+            new BalancerAddress("localhost", 80)
+        });
+
+        var channelOptions = new GrpcChannelOptions();
+        var acting = false;
+        var transportFactory = TestSubchannelTransportFactory.Create(async (subChannel, attempt, cancellationToken) =>
+        {
+            cancellationToken.Register(() =>
+            {
+                logger.LogError("Connect cancellation token canceled.");
+            });
+
+            if (!acting)
+            {
+                return new TryConnectResult(ConnectivityState.Ready);
+            }
+
+            await connectingSyncPoint.WaitToContinue().WaitAsync(cancellationToken);
+
+            Assert.IsFalse(cancellationToken.IsCancellationRequested, "Cancellation token should not be canceled.");
+
+            return new TryConnectResult(ConnectivityState.Ready);
+        });
+        var clientChannel = CreateConnectionManager(loggerFactory, resolver, transportFactory, new[] { new PickFirstBalancerFactory() });
+        // Configure balancer similar to how GrpcChannel constructor does it
+        clientChannel.ConfigureBalancer(c => new ChildHandlerLoadBalancer(
+            c,
+            channelOptions.ServiceConfig,
+            clientChannel));
+
+        await clientChannel.ConnectAsync(waitForReady: true, cancellationToken: CancellationToken.None);
+
+        transportFactory.Transports.ForEach(t => t.Disconnect());
+
+        testSink.MessageLogged += (w) =>
+        {
+            if (w.EventId.Name == "StartingConnectionRequest")
+            {
+                if (!requestConnectionChannel.Reader.TryRead(out var syncPoint))
+                {
+                    throw new InvalidOperationException("Channel should have sync point.");
+                }
+                syncPoint.WaitToContinue().Wait();
+            }
+        };
+
+        acting = true;
+
+        logger.LogInformation("Start first pick.");
+        var pickTask1 = Task.Run(() => clientChannel.PickAsync(
+            new PickContext { Request = new HttpRequestMessage() },
+            waitForReady: true,
+            CancellationToken.None).AsTask());
+
+        logger.LogInformation("Wait for first pick to request connection.");
+        await requestConnectionSyncPoint1.WaitForSyncPoint().DefaultTimeout();
+
+        logger.LogInformation("Start second pick.");
+        var pickTask2 = Task.Run(() => clientChannel.PickAsync(
+            new PickContext { Request = new HttpRequestMessage() },
+            waitForReady: true,
+            CancellationToken.None).AsTask());
+
+        logger.LogInformation("Wait for second pick to request connection.");
+        await requestConnectionSyncPoint2.WaitForSyncPoint().DefaultTimeout();
+
+        logger.LogInformation("Allow first pick to start connecting.");
+        requestConnectionSyncPoint1.Continue();
+        await connectingSyncPoint.WaitForSyncPoint();
+
+        var connectionRequestedInNonIdleStateTcs = new TaskCompletionSource<object?>(TaskCreationOptions.RunContinuationsAsynchronously);
+        testSink.MessageLogged += (w) =>
+        {
+            if (w.EventId.Name == "ConnectionRequestedInNonIdleState")
+            {
+                connectionRequestedInNonIdleStateTcs.TrySetResult(null);
+            }
+        };
+
+        logger.LogInformation("Allow second pick to wait for connecting to complete.");
+        requestConnectionSyncPoint2.Continue();
+
+        logger.LogInformation("Wait for second pick to report that there is already a connection requested.");
+        await connectionRequestedInNonIdleStateTcs.Task.DefaultTimeout();
+
+        logger.LogInformation("Allow first pick connecting to complete.");
+        connectingSyncPoint.Continue();
+
+        logger.LogInformation("Wait for both picks to complete successfully.");
+        await pickTask1.DefaultTimeout();
+        await pickTask2.DefaultTimeout();
+    }
+
     [Test]
     public async Task PickAsync_ExecutionContext_DoesNotCaptureAsyncLocalsInConnect()
     {