Skip to content

Retry stateful reconnect attempts #50745

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
BrennanConroy wants to merge 1 commit into from
Closed

Retry stateful reconnect attempts #50745

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 1 addition & 3 deletions src/SignalR/clients/csharp/Client.Core/src/HubConnection.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2082,7 +2082,7 @@ public bool ShouldProcessMessage(HubMessage message)
{
if (!_messageBuffer.ShouldProcessMessage(message))
{
Log.DroppingMessage(_logger, ((HubInvocationMessage)message).GetType().Name, ((HubInvocationMessage)message).InvocationId);
Log.DroppingMessage(_logger, message.GetType().Name, (message as HubInvocationMessage)?.InvocationId ?? "(null}");
Copy link
Member

@Kahbazi Kahbazi Sep 17, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

❓

Suggested change
Log.DroppingMessage(_logger, message.GetType().Name, (message as HubInvocationMessage)?.InvocationId ?? "(null}");
Log.DroppingMessage(_logger, message.GetType().Name, (message as HubInvocationMessage)?.InvocationId ?? "(null)");

Copy link

Copilot AI Feb 14, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The closing parenthesis in the error message is incorrect. It should be "(null)" instead of "(null}".

Suggested change
Log.DroppingMessage(_logger, message.GetType().Name, (message as HubInvocationMessage)?.InvocationId ?? "(null}");
Log.DroppingMessage(_logger, message.GetType().Name, (message as HubInvocationMessage)?.InvocationId ?? "(null)");

Copilot uses AI. Check for mistakes.
return false;
}
}
Expand Down Expand Up @@ -2168,7 +2168,6 @@ Type IInvocationBinder.GetReturnType(string invocationId)
{
if (!TryGetInvocation(invocationId, out var irq))
{
Log.ReceivedUnexpectedResponse(_logger, invocationId);
throw new KeyNotFoundException($"No invocation with id '{invocationId}' could be found.");
}
return irq.ResultType;
Expand All @@ -2180,7 +2179,6 @@ Type IInvocationBinder.GetStreamItemType(string invocationId)
// literally the same code as the above method
if (!TryGetInvocation(invocationId, out var irq))
{
Log.ReceivedUnexpectedResponse(_logger, invocationId);
throw new KeyNotFoundException($"No invocation with id '{invocationId}' could be found.");
}
return irq.ResultType;
Expand Down
141 changes: 139 additions & 2 deletions src/SignalR/clients/csharp/Client/test/FunctionalTests/HubConnectionTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2543,7 +2543,7 @@ public async Task ServerSentEventsWorksWithHttp2OnlyEndpoint()
public async Task CanReconnectAndSendMessageWhileDisconnected()
{
var protocol = HubProtocols["json"];
await using (var server = await StartServer<Startup>(w => w.EventId.Name == "ReceivedUnexpectedResponse"))
await using (var server = await StartServer<Startup>())
{
var websocket = new ClientWebSocket();
var tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
Expand Down Expand Up @@ -2602,7 +2602,7 @@ public async Task CanReconnectAndSendMessageWhileDisconnected()
public async Task CanReconnectAndSendMessageOnceConnected()
{
var protocol = HubProtocols["json"];
await using (var server = await StartServer<Startup>(w => w.EventId.Name == "ReceivedUnexpectedResponse"))
await using (var server = await StartServer<Startup>())
{
var websocket = new ClientWebSocket();
var tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
Expand Down Expand Up @@ -2910,6 +2910,143 @@ bool ExpectedErrors(WriteContext writeContext)
}
}

[Fact]
public async Task MultipleReconnectAttemptsWhenUsingStatefulReconnectFailure()
{
bool ExpectedErrors(WriteContext writeContext)
{
return writeContext.LoggerName == typeof(HubConnection).FullName &&
(writeContext.EventId.Name == "ShutdownWithError" ||
writeContext.EventId.Name == "ServerDisconnectedWithError");
}

var protocol = HubProtocols["json"];
await using (var server = await StartServer<Startup>(ExpectedErrors))
{
var websocketConnectAttempts = 0;
var websocket = new ClientWebSocket();

const string originalMessage = "SignalR";
var connectionBuilder = new HubConnectionBuilder()
.WithLoggerFactory(LoggerFactory)
.WithUrl(server.Url + "/default", HttpTransportType.WebSockets, o =>
{
o.WebSocketFactory = async (context, token) =>
{
websocketConnectAttempts++;
if (websocketConnectAttempts > 1)
{
throw new Exception("from websocket connect");
}
await websocket.ConnectAsync(context.Uri, token);
return websocket;
};
o.UseStatefulReconnect = true;
});
connectionBuilder.Services.AddSingleton(protocol);
var connection = connectionBuilder.Build();

try
{
await connection.StartAsync().DefaultTimeout();
Assert.Equal(1, websocketConnectAttempts);

var originalConnectionId = connection.ConnectionId;

var result = await connection.InvokeAsync<string>(nameof(TestHub.Echo), originalMessage).DefaultTimeout();

Assert.Equal(originalMessage, result);

var originalWebsocket = websocket;
websocket = new ClientWebSocket();
originalWebsocket.Dispose();

var resultTask = connection.InvokeAsync<string>(nameof(TestHub.Echo), originalMessage).DefaultTimeout();

await Assert.ThrowsAsync<TaskCanceledException>(() => resultTask);

// Initial connection + 3 failed reconnect attempts
Assert.Equal(4, websocketConnectAttempts);
}
catch (Exception ex)
{
LoggerFactory.CreateLogger<HubConnectionTests>().LogError(ex, "{ExceptionType} from test", ex.GetType().FullName);
throw;
}
finally
{
await connection.DisposeAsync().DefaultTimeout();
}
}
}

[Fact]
public async Task MultipleReconnectAttemptsWhenUsingStatefulReconnectSuccessful()
{
var protocol = HubProtocols["json"];
await using (var server = await StartServer<Startup>())
{
var websocketConnectAttempts = 0;
var websocket = new ClientWebSocket();

const string originalMessage = "SignalR";
var connectionBuilder = new HubConnectionBuilder()
.WithLoggerFactory(LoggerFactory)
.WithUrl(server.Url + "/default", HttpTransportType.WebSockets, o =>
{
o.WebSocketFactory = async (context, token) =>
{
websocketConnectAttempts++;
if (websocketConnectAttempts is > 1 and < 4)
{
throw new Exception("from websocket connect");
}
await websocket.ConnectAsync(context.Uri, token);
return websocket;
};
o.UseStatefulReconnect = true;
});
connectionBuilder.Services.AddSingleton(protocol);
var connection = connectionBuilder.Build();

try
{
await connection.StartAsync().DefaultTimeout();
Assert.Equal(1, websocketConnectAttempts);

var originalConnectionId = connection.ConnectionId;

var result = await connection.InvokeAsync<string>(nameof(TestHub.Echo), originalMessage).DefaultTimeout();

Assert.Equal(originalMessage, result);

var originalWebsocket = websocket;
websocket = new ClientWebSocket();
originalWebsocket.Dispose();

var resultTask = connection.InvokeAsync<string>(nameof(TestHub.Echo), originalMessage).DefaultTimeout();

result = await resultTask;
Assert.Equal(originalMessage, result);

// Initial connection + 2 failed reconnect attempts + 1 successful attempt
Assert.Equal(4, websocketConnectAttempts);

result = await connection.InvokeAsync<string>(nameof(TestHub.Echo), originalMessage).DefaultTimeout();
Assert.Equal(originalMessage, result);
}
catch (Exception ex)
{
LoggerFactory.CreateLogger<HubConnectionTests>().LogError(ex, "{ExceptionType} from test", ex.GetType().FullName);
throw;
}
finally
{
await connection.DisposeAsync().DefaultTimeout();
}
}
}

private class OneAtATimeSynchronizationContext : SynchronizationContext, IAsyncDisposable
{
private readonly Channel<(SendOrPostCallback, object)> _taskQueue = Channel.CreateUnbounded<(SendOrPostCallback, object)>();
Expand Down
24 changes: 18 additions & 6 deletions src/SignalR/clients/csharp/Http.Connections.Client/src/Internal/WebSocketsTransport.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -404,15 +404,27 @@ private async Task ProcessSocketAsync(WebSocket socket, Uri url, bool isReconnec
{
if (!_gracefulClose && UpdateConnectionPair())
{
try
Exception? exception = null;
// Arbitrary number of retries, will likely be user configurable in the future
for (var i = 0; i < 3; i++)
{
await StartAsync(url, _webSocketMessageType == WebSocketMessageType.Binary ? TransferFormat.Binary : TransferFormat.Text,
cancellationToken: default).ConfigureAwait(false);
cleanup = false;
try
{
await StartAsync(url, _webSocketMessageType == WebSocketMessageType.Binary ? TransferFormat.Binary : TransferFormat.Text,
cancellationToken: default).ConfigureAwait(false);
exception = null;
cleanup = false;
break;
}
catch (Exception ex)
{
exception = ex;
}
}
catch (Exception ex)

if (exception is not null)
{
throw new InvalidOperationException("Reconnect attempt failed.", innerException: ex);
throw new InvalidOperationException("Reconnect attempt failed.", innerException: exception);
}
}
}
Expand Down
18 changes: 16 additions & 2 deletions src/SignalR/clients/ts/signalr/src/HttpConnection.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -442,8 +442,22 @@ export class HttpConnection implements IConnection {
if (this.features.reconnect) {
try {
this.features.disconnected();
await this.transport!.connect(url, transferFormat);
await this.features.resend();
let error;
for (let i = 0; i < 3; i++) {
try {
await this.transport!.connect(url, transferFormat);
error = undefined;
break;
} catch (ex) {
error = ex;
}
}

if (error) {
callStop = true;
} else {
await this.features.resend();
}
} catch {
callStop = true;
}
Expand Down
19 changes: 12 additions & 7 deletions src/SignalR/clients/ts/signalr/tests/HttpConnection.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1981,13 +1981,18 @@ describe("TransportSendQueue", () => {
TestWebSocket.webSocketSet = new PromiseSource();
TestWebSocket.webSocket.close();

// transport should be trying to connect again
await TestWebSocket.webSocketSet;
await TestWebSocket.webSocket.openSet;
expect(disconnectedCalled).toBe(true);
expect(resendCalled).toBe(false);
// fail to connect
TestWebSocket.webSocket.onclose(new TestEvent());
for (let i = 0; i < 3; i++) {
// transport should be trying to connect again
await TestWebSocket.webSocketSet;
await TestWebSocket.webSocket.openSet;
expect(resendCalled).toBe(false);
expect(disconnectedCalled).toBe(true);

TestWebSocket.webSocketSet = new PromiseSource();

// fail to connect
TestWebSocket.webSocket.onclose(new TestEvent());
}

await onclosePromise;
expect(resendCalled).toBe(false);
Expand Down