Fix client timeout when error in custom CallCredential (#1388)

JamesNK · web-flow · commit 15158af86d44 · 2021-08-24T07:52:27.000+12:00
diff --git a/src/Grpc.Net.Client/Balancer/Internal/SocketConnectivitySubchannelTransport.cs b/src/Grpc.Net.Client/Balancer/Internal/SocketConnectivitySubchannelTransport.cs
@@ -21,6 +21,7 @@
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.IO;
+using System.Linq;
 using System.Net;
 using System.Net.Http;
 using System.Net.Sockets;
@@ -51,7 +52,7 @@ internal class SocketConnectivitySubchannelTransport : ISubchannelTransport, IDi
         private readonly ILogger _logger;
         private readonly Subchannel _subchannel;
         private readonly TimeSpan _socketPingInterval;
-        internal readonly List<(DnsEndPoint EndPoint, Socket Socket, Stream? Stream)> _activeStreams;
+        private readonly List<(DnsEndPoint EndPoint, Socket Socket, Stream? Stream)> _activeStreams;
         private readonly Timer _socketConnectedTimer;
 
         private int _lastEndPointIndex;
@@ -73,6 +74,15 @@ public SocketConnectivitySubchannelTransport(Subchannel subchannel, TimeSpan soc
         public DnsEndPoint? CurrentEndPoint => _currentEndPoint;
         public bool HasStream { get; }
 
+        // For testing. Take a copy under lock for thread-safety.
+        internal IReadOnlyList<(DnsEndPoint EndPoint, Socket Socket, Stream? Stream)> GetActiveStreams()
+        {
+            lock (Lock)
+            {
+                return _activeStreams.ToList();
+            }
+        }
+
         public void Disconnect()
         {
             lock (Lock)
diff --git a/src/Grpc.Net.Client/Internal/GrpcCall.cs b/src/Grpc.Net.Client/Internal/GrpcCall.cs
@@ -415,18 +415,20 @@ private async Task RunCall(HttpRequestMessage request, TimeSpan? timeout)
             {
                 var (diagnosticSourceEnabled, activity) = InitializeCall(request, timeout);
 
-                if (Options.Credentials != null || Channel.CallCredentials?.Count > 0)
-                {
-                    await ReadCredentials(request).ConfigureAwait(false);
-                }
-
                 // Unset variable to check that FinishCall is called in every code path
                 bool finished;
 
                 Status? status = null;
 
                 try
                 {
+                    // User supplied call credentials could error and so must be run
+                    // inside try/catch to handle errors.
+                    if (Options.Credentials != null || Channel.CallCredentials?.Count > 0)
+                    {
+                        await ReadCredentials(request).ConfigureAwait(false);
+                    }
+
                     // Fail early if deadline has already been exceeded
                     _callCts.Token.ThrowIfCancellationRequested();
 
diff --git a/test/FunctionalTests/Balancer/ConnectionTests.cs b/test/FunctionalTests/Balancer/ConnectionTests.cs
@@ -110,12 +110,16 @@ async Task<HelloReply> UnaryMethod(HelloRequest request, ServerCallContext conte
             var balancer = BalancerHelpers.GetInnerLoadBalancer<PickFirstBalancer>(channel)!;
             var subchannel = balancer._subchannel!;
             var transport = (SocketConnectivitySubchannelTransport)subchannel.Transport;
-            var activeStreams = transport._activeStreams;
+            var activeStreams = transport.GetActiveStreams();
 
             // Assert
             Assert.AreEqual(HttpHandlerType.SocketsHttpHandler, channel.HttpHandlerType);
 
-            await TestHelpers.AssertIsTrueRetryAsync(() => activeStreams.Count == 10, "Wait for connections to start.");
+            await TestHelpers.AssertIsTrueRetryAsync(() =>
+            {
+                activeStreams = transport.GetActiveStreams();
+                return activeStreams.Count == 10;
+            }, "Wait for connections to start.");
             foreach (var t in activeStreams)
             {
                 Assert.AreEqual(new DnsEndPoint("127.0.0.1", 50051), t.EndPoint);
@@ -134,7 +138,11 @@ async Task<HelloReply> UnaryMethod(HelloRequest request, ServerCallContext conte
             Logger.LogInformation($"Done sending gRPC calls");
 
             // Assert
-            await TestHelpers.AssertIsTrueRetryAsync(() => activeStreams.Count == 11, "Wait for connections to start.");
+            await TestHelpers.AssertIsTrueRetryAsync(() =>
+            {
+                activeStreams = transport.GetActiveStreams();
+                return activeStreams.Count == 11;
+            }, "Wait for connections to start.");
             Assert.AreEqual(new DnsEndPoint("127.0.0.1", 50051), activeStreams.Last().EndPoint);
 
             tcs.SetResult(null);
@@ -149,7 +157,11 @@ async Task<HelloReply> UnaryMethod(HelloRequest request, ServerCallContext conte
 
             // There are still be 10 HTTP/1.1 connections because they aren't immediately removed
             // when the server is shutdown and connectivity is lost.
-            await TestHelpers.AssertIsTrueRetryAsync(() => activeStreams.Count == 10, "Wait for HTTP/2 connection to end.");
+            await TestHelpers.AssertIsTrueRetryAsync(() =>
+            {
+                activeStreams = transport.GetActiveStreams();
+                return activeStreams.Count == 10;
+            }, "Wait for HTTP/2 connection to end.");
 
             grpcWebHandler.HttpVersion = new Version(1, 1);
 
@@ -160,6 +172,7 @@ async Task<HelloReply> UnaryMethod(HelloRequest request, ServerCallContext conte
             Assert.AreEqual(StatusCode.Unavailable, ex.StatusCode);
 
             // Removed by failed call.
+            activeStreams = transport.GetActiveStreams();
             Assert.AreEqual(0, activeStreams.Count);
             Assert.AreEqual(ConnectivityState.Idle, channel.State);
 
@@ -168,6 +181,7 @@ async Task<HelloReply> UnaryMethod(HelloRequest request, ServerCallContext conte
             Assert.AreEqual("Balancer", reply.Message);
             Assert.AreEqual("127.0.0.1:50052", host);
 
+            activeStreams = transport.GetActiveStreams();
             Assert.AreEqual(1, activeStreams.Count);
             Assert.AreEqual(new DnsEndPoint("127.0.0.1", 50052), activeStreams[0].EndPoint);
         }
diff --git a/test/FunctionalTests/Balancer/PickFirstBalancerTests.cs b/test/FunctionalTests/Balancer/PickFirstBalancerTests.cs
@@ -246,7 +246,7 @@ async Task<HelloReply> UnaryMethod(HelloRequest request, ServerCallContext conte
             var balancer = BalancerHelpers.GetInnerLoadBalancer<PickFirstBalancer>(channel)!;
             var subchannel = balancer._subchannel!;
             var transport = (SocketConnectivitySubchannelTransport)subchannel.Transport;
-            var activeStreams = transport._activeStreams;
+            var activeStreams = transport.GetActiveStreams();
 
             // Assert
             Assert.AreEqual(2, activeStreams.Count);
@@ -263,6 +263,7 @@ async Task<HelloReply> UnaryMethod(HelloRequest request, ServerCallContext conte
 
             await TestHelpers.AssertIsTrueRetryAsync(() =>
             {
+                activeStreams = transport.GetActiveStreams();
                 Logger.LogInformation($"Current active stream endpoints: {string.Join(", ", activeStreams.Select(s => s.EndPoint))}");
                 return activeStreams.Count == 0;
             }, "Active streams removed.", Logger).DefaultTimeout();
@@ -271,6 +272,7 @@ await TestHelpers.AssertIsTrueRetryAsync(() =>
             Assert.AreEqual("Balancer", reply.Message);
             Assert.AreEqual("127.0.0.1:50052", host);
 
+            activeStreams = transport.GetActiveStreams();
             Assert.AreEqual(1, activeStreams.Count);
             Assert.AreEqual(new DnsEndPoint("127.0.0.1", 50052), activeStreams[0].EndPoint);
         }
diff --git a/test/Grpc.Net.Client.Tests/CallCredentialTests.cs b/test/Grpc.Net.Client.Tests/CallCredentialTests.cs
@@ -28,6 +28,8 @@
 using Grpc.Net.Client.Internal;
 using Grpc.Net.Client.Tests.Infrastructure;
 using Grpc.Tests.Shared;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Logging.Testing;
 using NUnit.Framework;
 
@@ -36,6 +38,37 @@ namespace Grpc.Net.Client.Tests
     [TestFixture]
     public class CallCredentialTests
     {
+        [Test]
+        public async Task CallCredentialsWithHttps_WhenAsyncAuthInterceptorThrow_ShouldThrow()
+        {
+            // Arrange
+            var services = new ServiceCollection();
+            services.AddNUnitLogger();
+            var loggerFactory = services.BuildServiceProvider().GetRequiredService<ILoggerFactory>();
+
+            var httpClient = ClientTestHelpers.CreateTestClient(async request =>
+            {
+                var reply = new HelloReply { Message = "Hello world" };
+                var streamContent = await ClientTestHelpers.CreateResponseContent(reply).DefaultTimeout();
+                return ResponseUtils.CreateResponse(HttpStatusCode.OK, streamContent);
+            });
+            var invoker = HttpClientCallInvokerFactory.Create(httpClient, loggerFactory);
+
+            // Act
+            var expectedException = new Exception("Some AsyncAuthInterceptor Exception");
+
+            var callCredentials = CallCredentials.FromInterceptor((context, metadata) =>
+            {
+                return Task.FromException(expectedException);
+            });
+
+            var call = invoker.AsyncUnaryCall<HelloRequest, HelloReply>(ClientTestHelpers.ServiceMethod, string.Empty, new CallOptions(credentials: callCredentials), new HelloRequest());
+            var ex = await ExceptionAssert.ThrowsAsync<RpcException>(() => call.ResponseAsync).DefaultTimeout();
+
+            // Assert
+            Assert.AreSame(expectedException, ex.Status.DebugException);
+        }
+
         [Test]
         public async Task CallCredentialsWithHttps_MetadataOnRequest()
         {
diff --git a/test/Shared/HttpEventSourceListener.cs b/test/Shared/HttpEventSourceListener.cs
@@ -16,7 +16,6 @@
 
 #endregion
 
-using System;
 using System.Diagnostics.Tracing;
 using System.Text;
 using Microsoft.Extensions.Logging;
@@ -26,7 +25,7 @@ namespace Grpc.Tests.Shared
     public sealed class HttpEventSourceListener : EventListener
     {
         private readonly StringBuilder _messageBuilder = new StringBuilder();
-        private readonly ILogger _logger;
+        private readonly ILogger? _logger;
         private readonly object _lock = new object();
         private bool _disposed;
 
@@ -43,7 +42,13 @@ protected override void OnEventSourceCreated(EventSource eventSource)
             if (eventSource.Name.Contains("System.Net.Quic") ||
                 eventSource.Name.Contains("System.Net.Http"))
             {
-                EnableEvents(eventSource, EventLevel.LogAlways, EventKeywords.All);
+                lock (_lock)
+                {
+                    if (!_disposed)
+                    {
+                        EnableEvents(eventSource, EventLevel.LogAlways, EventKeywords.All);
+                    }
+                }
             }
         }
 
@@ -72,7 +77,10 @@ protected override void OnEventWritten(EventWrittenEventArgs eventData)
             {
                 if (!_disposed)
                 {
-                    _logger.LogDebug(message);
+                    // EventListener base constructor subscribes to events.
+                    // It is possible to start getting events before the
+                    // super constructor is run and logger is assigned.
+                    _logger?.LogDebug(message);
                 }
             }
         }
@@ -90,7 +98,7 @@ public override void Dispose()
             {
                 if (!_disposed)
                 {
-                    _logger.LogDebug($"Stopping {nameof(HttpEventSourceListener)}.");
+                    _logger?.LogDebug($"Stopping {nameof(HttpEventSourceListener)}.");
                     _disposed = true;
                 }
             }

Original file line number	Diff line number	Diff line change
`@@ -16,7 +16,6 @@`
`16`	`16`
`17`	`17`	`#endregion`
`18`	`18`
`19`		`-using System;`
`20`	`19`	`using System.Diagnostics.Tracing;`
`21`	`20`	`using System.Text;`
`22`	`21`	`using Microsoft.Extensions.Logging;`
`@@ -26,7 +25,7 @@ namespace Grpc.Tests.Shared`
`26`	`25`	`public sealed class HttpEventSourceListener : EventListener`
`27`	`26`	`{`
`28`	`27`	`private readonly StringBuilder _messageBuilder = new StringBuilder();`
`29`		`- private readonly ILogger _logger;`
	`28`	`+ private readonly ILogger? _logger;`
`30`	`29`	`private readonly object _lock = new object();`
`31`	`30`	`private bool _disposed;`
`32`	`31`
`@@ -43,7 +42,13 @@ protected override void OnEventSourceCreated(EventSource eventSource)`
`43`	`42`	`if (eventSource.Name.Contains("System.Net.Quic") \|\|`
`44`	`43`	`eventSource.Name.Contains("System.Net.Http"))`
`45`	`44`	`{`
`46`		`- EnableEvents(eventSource, EventLevel.LogAlways, EventKeywords.All);`
	`45`	`+ lock (_lock)`
	`46`	`+ {`
	`47`	`+ if (!_disposed)`
	`48`	`+ {`
	`49`	`+ EnableEvents(eventSource, EventLevel.LogAlways, EventKeywords.All);`
	`50`	`+ }`
	`51`	`+ }`
`47`	`52`	`}`
`48`	`53`	`}`
`49`	`54`
`@@ -72,7 +77,10 @@ protected override void OnEventWritten(EventWrittenEventArgs eventData)`
`72`	`77`	`{`
`73`	`78`	`if (!_disposed)`
`74`	`79`	`{`
`75`		`- _logger.LogDebug(message);`
	`80`	`+ // EventListener base constructor subscribes to events.`
	`81`	`+ // It is possible to start getting events before the`
	`82`	`+ // super constructor is run and logger is assigned.`
	`83`	`+ _logger?.LogDebug(message);`
`76`	`84`	`}`
`77`	`85`	`}`
`78`	`86`	`}`
`@@ -90,7 +98,7 @@ public override void Dispose()`
`90`	`98`	`{`
`91`	`99`	`if (!_disposed)`
`92`	`100`	`{`
`93`		`- _logger.LogDebug($"Stopping {nameof(HttpEventSourceListener)}.");`
	`101`	`+ _logger?.LogDebug($"Stopping {nameof(HttpEventSourceListener)}.");`
`94`	`102`	`_disposed = true;`
`95`	`103`	`}`
`96`	`104`	`}`