SignalR MessageBuffer fixes (#50315)

* SignalR MessageBuffer fixes

* pool

* version

* Apply suggestions from code review

* fb

Author: BrennanConroy

eng/Dependencies.props

@@ -17,6 +17,7 @@ and are generated based on the last package release.
     <LatestPackageReference Include="Microsoft.AspNetCore.Mvc.Razor.Extensions" />
     <LatestPackageReference Include="Microsoft.Azure.SignalR" />
     <LatestPackageReference Include="Microsoft.Bcl.HashCode" />
+    <LatestPackageReference Include="Microsoft.Bcl.TimeProvider" />
     <LatestPackageReference Include="Microsoft.Css.Parser" />
     <LatestPackageReference Include="Microsoft.CodeAnalysis.Common" />
     <LatestPackageReference Include="Microsoft.CodeAnalysis.CSharp.Workspaces" />
@@ -63,6 +64,7 @@ and are generated based on the last package release.
     <LatestPackageReference Include="Microsoft.Extensions.Options" />
     <LatestPackageReference Include="Microsoft.Extensions.Primitives" />
     <LatestPackageReference Include="Microsoft.Extensions.Telemetry.Testing" />
+    <LatestPackageReference Include="Microsoft.Extensions.TimeProvider.Testing" />
     <LatestPackageReference Include="Microsoft.Win32.Registry" />
     <LatestPackageReference Include="Microsoft.CodeAnalysis.CSharp.Analyzer.Testing.XUnit" />
     <LatestPackageReference Include="Microsoft.CodeAnalysis.CSharp.CodeFix.Testing.XUnit" />

eng/Version.Details.xml

@@ -397,6 +397,10 @@
       <Uri>https://github.com/dotnet/extensions</Uri>
       <Sha>7c6fa3e84ea0b3d08998726c7cac30e3117ed318</Sha>
     </Dependency>
+    <Dependency Name="Microsoft.Extensions.TimeProvider.Testing" Version="8.0.0-rc.2.23461.3">
+      <Uri>https://github.com/dotnet/extensions</Uri>
+      <Sha>7c6fa3e84ea0b3d08998726c7cac30e3117ed318</Sha>
+    </Dependency>
     <Dependency Name="NuGet.Frameworks" Version="6.2.4">
       <Uri>https://github.com/nuget/nuget.client</Uri>
       <Sha>8fef55f5a55a3b4f2c96cd1a9b5ddc51d4b927f8</Sha>

eng/Versions.props

@@ -138,6 +138,7 @@
     <SystemRuntimeCachingVersion>8.0.0-rc.2.23457.7</SystemRuntimeCachingVersion>
     <!-- Packages from dotnet/extensions -->
     <MicrosoftExtensionsTelemetryTestingVersion>8.0.0-rc.2.23461.3</MicrosoftExtensionsTelemetryTestingVersion>
+    <MicrosoftExtensionsTimeProviderTestingVersion>8.0.0-rc.2.23461.3</MicrosoftExtensionsTimeProviderTestingVersion>
     <!-- Packages from dotnet/efcore -->
     <dotnetefVersion>8.0.0-rc.2.23462.14</dotnetefVersion>
     <MicrosoftEntityFrameworkCoreInMemoryVersion>8.0.0-rc.2.23462.14</MicrosoftEntityFrameworkCoreInMemoryVersion>

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

@@ -1006,7 +1006,7 @@ private async Task SendHubMessage(ConnectionState connectionState, HubMessage hu
 
         if (connectionState.UsingAcks())
         {
-            await connectionState.WriteAsync(new SerializedHubMessage(hubMessage), cancellationToken).ConfigureAwait(false);
+            await connectionState.WriteAsync(hubMessage, cancellationToken).ConfigureAwait(false);
         }
         else
         {
@@ -1133,7 +1133,7 @@ private async Task SendWithLock(ConnectionState expectedConnectionState, HubMess
                 break;
             case AckMessage ackMessage:
                 Log.ReceivedAckMessage(_logger, ackMessage.SequenceId);
-                connectionState.Ack(ackMessage);
+                await connectionState.AckAsync(ackMessage).ConfigureAwait(false);
                 break;
             case SequenceMessage sequenceMessage:
                 Log.ReceivedSequenceMessage(_logger, sequenceMessage.SequenceId);
@@ -1943,7 +1943,7 @@ public ConnectionState(ConnectionContext connection, HubConnection hubConnection
             {
                 _messageBuffer = new MessageBuffer(connection, hubConnection._protocol,
                     _hubConnection._serviceProvider.GetService<IOptions<HubConnectionOptions>>()?.Value.StatefulReconnectBufferSize
-                        ?? DefaultStatefulReconnectBufferSize);
+                        ?? DefaultStatefulReconnectBufferSize, _logger);
 
                 feature.OnReconnected(_messageBuffer.ResendAsync);
             }
@@ -2071,7 +2071,7 @@ public async Task TimerLoop(TimerAwaitable timer)
             }
         }
 
-        public ValueTask<FlushResult> WriteAsync(SerializedHubMessage message, CancellationToken cancellationToken)
+        public ValueTask<FlushResult> WriteAsync(HubMessage message, CancellationToken cancellationToken)
         {
             Debug.Assert(_messageBuffer is not null);
             return _messageBuffer.WriteAsync(message, cancellationToken);
@@ -2090,12 +2090,14 @@ public bool ShouldProcessMessage(HubMessage message)
             return true;
         }
 
-        public void Ack(AckMessage ackMessage)
+        public Task AckAsync(AckMessage ackMessage)
         {
             if (UsingAcks())
             {
-                _messageBuffer.Ack(ackMessage);
+                return _messageBuffer.AckAsync(ackMessage);
             }
+
+            return Task.CompletedTask;
         }
 
         public void ResetSequence(SequenceMessage sequenceMessage)

src/SignalR/clients/csharp/Client.Core/src/Microsoft.AspNetCore.SignalR.Client.Core.csproj

@@ -34,6 +34,10 @@
     <Reference Include="System.Threading.Channels" />
   </ItemGroup>
 
+  <ItemGroup Condition="'$(TargetFramework)' != '$(DefaultNetCoreTargetFramework)'">
+    <Reference Include="Microsoft.Bcl.TimeProvider" />
+  </ItemGroup>
+
   <ItemGroup>
     <InternalsVisibleTo Include="Microsoft.AspNetCore.SignalR.Client.FunctionalTests" />
     <InternalsVisibleTo Include="Microsoft.AspNetCore.SignalR.Client.Tests" />

src/SignalR/clients/csharp/Http.Connections.Client/src/Internal/WebSocketsTransport.cs

@@ -616,7 +616,7 @@ private async Task StartSending(WebSocket socket, bool ignoreFirstCanceled)
 
             if (_gracefulClose || !_useStatefulReconnect)
             {
-                _application.Input.Complete(error);
+                _application.Input.Complete();
             }
             else
             {

src/SignalR/common/Http.Connections/src/Internal/HttpConnectionDispatcher.cs

@@ -295,8 +295,9 @@ private async Task ExecuteAsync(HttpContext context, ConnectionDelegate connecti
                 {
                     // If false then the transport was ungracefully closed, this can mean a temporary network disconnection
                     // We'll mark the connection as inactive and allow the connection to reconnect if that's the case.
-                    // TODO: If acks aren't enabled we can close the connection immediately (not LongPolling)
-                    if (await connection.TransportTask!)
+                    if (await connection.TransportTask!
+                        // If acks aren't enabled we can close the connection immediately (not LongPolling)
+                        || !connection.ClientReconnectExpected())
                     {
                         await _manager.DisposeAndRemoveAsync(connection, closeGracefully: true, HttpConnectionStopStatus.NormalClosure);
                     }

src/SignalR/common/Http.Connections/src/Internal/Transports/WebSocketsServerTransport.cs

@@ -288,9 +288,10 @@ private async Task StartSending(WebSocket socket, bool ignoreFirstCancel)
 
             if (_gracefulClose)
             {
-                _application.Input.Complete(error);
+                _application.Input.Complete();
             }
-            else if (error is not null)
+
+            if (error is not null)
             {
                 Log.SendErrored(_logger, error);
             }