[Backport] [SignalR] Avoid unobserved tasks in the .NET client (#43882)

Author: BrennanConroy

eng/targets/CSharp.Common.targets

@@ -116,7 +116,7 @@
      -->
     <When Condition=" ('$(Nullable)' == 'annotations' OR '$(Nullable)' == 'enable') AND
         '$(SuppressNullableAttributesImport)' != 'true' AND
-        (('$(TargetFrameworkIdentifier)' == '.NETStandard' AND $([MSBuild]::VersionLessThanOrEquals('$(TargetFrameworkVersion)', '2.0'))) OR '$(TargetFrameworkIdentifier)' == '.NETFramework')">
+        (('$(TargetFrameworkIdentifier)' == '.NETStandard' AND $([MSBuild]::VersionLessThanOrEquals('$(TargetFrameworkVersion)', '2.1'))) OR '$(TargetFrameworkIdentifier)' == '.NETFramework')">
       <PropertyGroup>
         <DefineConstants>$(DefineConstants),INTERNAL_NULLABLE_ATTRIBUTES</DefineConstants>
         <NoWarn>$(NoWarn);nullable</NoWarn>

src/Shared/Nullable/NullableAttributes.cs

@@ -5,6 +5,8 @@
 
 namespace System.Diagnostics.CodeAnalysis
 {
+// Attributes added in netstandard2.1
+#if !NETSTANDARD2_1_OR_GREATER
     /// <summary>Specifies that null is allowed as an input even if the corresponding type disallows it.</summary>
     [AttributeUsage(AttributeTargets.Field | AttributeTargets.Parameter | AttributeTargets.Property, Inherited = false)]
 #if INTERNAL_NULLABLE_ATTRIBUTES
@@ -126,7 +128,10 @@ sealed class DoesNotReturnIfAttribute : Attribute
         /// <summary>Gets the condition parameter value.</summary>
         public bool ParameterValue { get; }
     }
+#endif
 
+// Attributes added in 5.0
+#if NETSTANDARD || NETFRAMEWORK
     /// <summary>Specifies that the method or property will ensure that the listed field and property members have not-null values.</summary>
     [AttributeUsage(AttributeTargets.Method | AttributeTargets.Property, Inherited = false, AllowMultiple = true)]
 #if INTERNAL_NULLABLE_ATTRIBUTES
@@ -193,4 +198,5 @@ public MemberNotNullWhenAttribute(bool returnValue, params string[] members)
         /// <summary>Gets field or property member names.</summary>
         public string[] Members { get; }
     }
+#endif
 }

src/SignalR/clients/csharp/Client.Core/src/HubConnection.Log.cs

@@ -10,7 +10,7 @@ namespace Microsoft.AspNetCore.SignalR.Client
 {
     public partial class HubConnection
     {
-        private static class Log
+        private static partial class Log
         {
             private static readonly LogDefineOptions SkipEnabledCheckLogOptions = new() { SkipEnabledCheck = true };
 
@@ -254,6 +254,11 @@ private static class Log
             private static readonly Action<ILogger, string, Exception?> _erroredStream =
                 LoggerMessage.Define<string>(LogLevel.Trace, new EventId(84, "ErroredStream"), "Client threw an error for stream '{StreamId}'.");
 
+            [LoggerMessage(87, LogLevel.Trace, "Completion message for stream '{StreamId}' was not sent because the connection is closed.", EventName = "CompletingStreamNotSent")]
+            public static partial void CompletingStreamNotSent(ILogger logger, string streamId);
+
+            // EventId 88 used in 7.0+
+
             public static void PreparingNonBlockingInvocation(ILogger logger, string target, int count)
             {
                 _preparingNonBlockingInvocation(logger, target, count, null);

src/SignalR/clients/csharp/Client.Core/src/HubConnection.cs

@@ -782,11 +782,26 @@ private async Task CommonStreaming(ConnectionState connectionState, string strea
                 responseError = $"Stream errored by client: '{ex}'";
             }
 
-            Log.CompletingStream(_logger, streamId);
-
             // Don't use cancellation token here
             // this is triggered by a cancellation token to tell the server that the client is done streaming
-            await SendWithLock(connectionState, CompletionMessage.WithError(streamId, responseError), cancellationToken: default);
+            await _state.WaitConnectionLockAsync(token: default).ConfigureAwait(false);
+            try
+            {
+                // Avoid sending when the connection isn't active, likely happens if there is an active stream when the connection closes
+                if (_state.IsConnectionActive())
+                {
+                    Log.CompletingStream(_logger, streamId);
+                    await SendHubMessage(connectionState, CompletionMessage.WithError(streamId, responseError), cancellationToken: default).ConfigureAwait(false);
+                }
+                else
+                {
+                    Log.CompletingStreamNotSent(_logger, streamId);
+                }
+            }
+            finally
+            {
+                _state.ReleaseConnectionLock();
+            }
         }
 
         private async Task<object?> InvokeCoreAsyncCore(string methodName, Type returnType, object?[] args, CancellationToken cancellationToken)
@@ -2007,7 +2022,7 @@ public async Task<ConnectionState> WaitForActiveConnectionAsync(string methodNam
             {
                 await WaitConnectionLockAsync(token, methodName);
 
-                if (CurrentConnectionStateUnsynchronized == null || CurrentConnectionStateUnsynchronized.Stopping)
+                if (!IsConnectionActive())
                 {
                     ReleaseConnectionLock(methodName);
                     throw new InvalidOperationException($"The '{methodName}' method cannot be called if the connection is not active");
@@ -2016,6 +2031,13 @@ public async Task<ConnectionState> WaitForActiveConnectionAsync(string methodNam
                 return CurrentConnectionStateUnsynchronized;
             }
 
+            [MemberNotNullWhen(true, nameof(CurrentConnectionStateUnsynchronized))]
+            public bool IsConnectionActive()
+            {
+                AssertInConnectionLock();
+                return CurrentConnectionStateUnsynchronized is not null && !CurrentConnectionStateUnsynchronized.Stopping;
+            }
+
             public void ReleaseConnectionLock([CallerMemberName] string? memberName = null,
                 [CallerFilePath] string? filePath = null, [CallerLineNumber] int lineNumber = 0)
             {

src/SignalR/clients/csharp/Client/test/UnitTests/HubConnectionTests.cs

@@ -544,6 +544,40 @@ public async Task UploadStreamErrorSendsStreamComplete()
             }
         }
 
+        [Fact]
+        [LogLevel(LogLevel.Trace)]
+        public async Task ActiveUploadStreamWhenConnectionClosesObservesException()
+        {
+            using (StartVerifiableLog())
+            {
+                var connection = new TestConnection();
+                var hubConnection = CreateHubConnection(connection, loggerFactory: LoggerFactory);
+                await hubConnection.StartAsync().DefaultTimeout();
+
+                var channel = Channel.CreateUnbounded<int>();
+                var invokeTask = hubConnection.InvokeAsync<object>("UploadMethod", channel.Reader);
+
+                var invokeMessage = await connection.ReadSentJsonAsync().DefaultTimeout();
+                Assert.Equal(HubProtocolConstants.InvocationMessageType, invokeMessage["type"]);
+
+                // Not sure how to test for unobserved task exceptions, best I could come up with is to check that we log where there once was an unobserved task exception
+                var tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+                TestSink.MessageLogged += wc =>
+                {
+                    if (wc.EventId.Name == "CompletingStreamNotSent")
+                    {
+                        tcs.SetResult();
+                    }
+                };
+
+                await hubConnection.StopAsync();
+
+                await Assert.ThrowsAsync<TaskCanceledException>(() => invokeTask).DefaultTimeout();
+
+                await tcs.Task.DefaultTimeout();
+            }
+        }
+
         [Fact]
         [LogLevel(LogLevel.Trace)]
         public async Task InvocationCanCompleteBeforeStreamCompletes()