Fix race condition when finishing streaming and getting trailers (#876)

Author: JamesNK

src/Grpc.Net.Client/Internal/GrpcCall.cs

@@ -210,6 +210,11 @@ private void FinishResponseAndCleanUp(Status status)
         /// <param name="status">The completed response status code.</param>
         public void ResponseStreamEnded(Status status)
         {
+            // Set response finished immediately rather than set it in logic resumed
+            // from the callTcs to avoid race condition.
+            // e.g. response stream finished and then immediately call GetTrailers().
+            ResponseFinished = true;
+
             _callTcs.TrySetResult(status);
         }
 
@@ -546,7 +551,7 @@ private async ValueTask RunCall(HttpRequestMessage request, TimeSpan? timeout)
                             status = await CallTask.ConfigureAwait(false);
 
                             finished = FinishCall(request, diagnosticSourceEnabled, activity, status.Value);
-                            FinishResponseAndCleanUp(status.Value);
+                            Cleanup(status.Value);
                         }
                     }
                 }

test/FunctionalTests/Client/StreamingTests.cs

@@ -19,6 +19,7 @@
 using System;
 using System.Diagnostics;
 using System.IO;
+using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using Google.Protobuf;
@@ -424,6 +425,39 @@ async Task UnaryDeadlineExceeded(IAsyncStreamReader<DataMessage> requestStream,
             Assert.IsFalse(await call2.ResponseStream.MoveNext().DefaultTimeout());
         }
 
+        [Test]
+        public async Task ServerStreaming_GetTrailersAndStatus_Success()
+        {
+            async Task ServerStreamingWithTrailers(DataMessage request, IServerStreamWriter<DataMessage> responseStream, ServerCallContext context)
+            {
+                await responseStream.WriteAsync(new DataMessage());
+                context.ResponseTrailers.Add("my-trailer", "value");
+            }
+
+            // Arrange
+            var method = Fixture.DynamicGrpc.AddServerStreamingMethod<DataMessage, DataMessage>(ServerStreamingWithTrailers);
+
+            var channel = CreateChannel();
+
+            var client = TestClientFactory.Create(channel, method);
+
+            // Act
+            var call = client.ServerStreamingCall(new DataMessage());
+
+            // Assert
+            Assert.IsTrue(await call.ResponseStream.MoveNext().DefaultTimeout());
+
+            Assert.AreEqual(0, call.ResponseStream.Current.Data.Length);
+
+            Assert.IsFalse(await call.ResponseStream.MoveNext().DefaultTimeout());
+
+            var trailers = call.GetTrailers();
+            Assert.AreEqual(1, trailers.Count);
+            Assert.AreEqual("value", trailers.First(e => e.Key == "my-trailer").Value);
+
+            Assert.AreEqual(StatusCode.OK, call.GetStatus().StatusCode);
+        }
+
         private static byte[] CreateTestData(int size)
         {
             var data = new byte[size];