fix(link): recover stale websocket sessions [CI]
ci/woodpecker/push/dev-image Pipeline is pending
ci/woodpecker/push/web-build Pipeline was canceled
ci/woodpecker/push/verify Pipeline was canceled

This commit is contained in:
justme committed 2026-10-11 19:54:14 +08:00
1 parent b5e275c51f
commit be97e24693
2 files changed
+40 -5

No files matched your search

+19 -3
View File
@@ -140,7 +140,7 @@ public sealed class FleetLink
{
return;
}
catch (Exception ex) when (ex is WebSocketException or IOException or HttpRequestException or InvalidOperationException or TimeoutException)
catch (Exception ex)
{
reason = ex.Message;
}
@@ -214,6 +214,7 @@ public sealed class FleetLink
Task cancellation = Task.Delay(Timeout.Infinite, ct);
Task finished = await Task.WhenAny(run, pump, heartbeat, cancellation);
string finishedReason = DescribeFinishedTask(finished, run, pump, heartbeat);
if (ct.IsCancellationRequested)
{
try
@@ -231,7 +232,7 @@ public sealed class FleetLink
await sessionCts.CancelAsync();
await Task.WhenAll(Quiet(run), Quiet(pump), Quiet(heartbeat));
return finished == run ? "closed by server" : "link task failed";
return finishedReason;
}
finally
{
@@ -398,9 +399,24 @@ public sealed class FleetLink
{
await task;
}
catch (Exception ex) when (ex is OperationCanceledException or WebSocketException or IOException or ObjectDisposedException)
catch (OperationCanceledException)
{
// Session cancellation is expected during reconnect and shutdown.
}
catch (Exception ex) when (ex is WebSocketException or IOException or ObjectDisposedException or TimeoutException or InvalidOperationException)
{
// 会话结束时的正常异常
}
}
private static string DescribeFinishedTask(Task finished, Task run, Task pump, Task heartbeat)
{
if (finished.IsFaulted)
{
string taskName = finished == run ? "receive" : finished == pump ? "outbox send" : "heartbeat";
return $"{taskName} failed: {finished.Exception?.GetBaseException().Message ?? "unknown error"}";
}
return finished == run ? "closed by server" : "link task stopped";
}
}
+21 -2
View File
@@ -7,6 +7,7 @@ namespace Fleet.Link;
/// <summary>WebSocket 传输(节点 → 服务端):文本帧是 JSON-RPC,二进制帧是发件箱。</summary>
public sealed class WsTransport : IRpcTransport
{
private static readonly TimeSpan s_sendTimeout = TimeSpan.FromSeconds(20);
private readonly WebSocket _socket;
private readonly int _maxFrameBytes;
private readonly SemaphoreSlim _sendLock = new(1, 1);
@@ -19,16 +20,34 @@ public sealed class WsTransport : IRpcTransport
public async ValueTask SendAsync(RpcFrame frame, CancellationToken ct)
{
await _sendLock.WaitAsync(ct);
using var timeout = new CancellationTokenSource(s_sendTimeout);
using CancellationTokenSource linked = CancellationTokenSource.CreateLinkedTokenSource(ct, timeout.Token);
bool lockTaken = false;
try
{
await _socket.SendAsync(frame.Data, frame.Kind == FrameKind.Text ? WebSocketMessageType.Text : WebSocketMessageType.Binary, endOfMessage: true, ct);
await _sendLock.WaitAsync(linked.Token);
lockTaken = true;
if (_socket.State != WebSocketState.Open)
{
throw new WebSocketException($"WebSocket is not open (state: {_socket.State})");
}
await _socket.SendAsync(frame.Data, frame.Kind == FrameKind.Text ? WebSocketMessageType.Text : WebSocketMessageType.Binary, endOfMessage: true, linked.Token);
}
catch (OperationCanceledException) when (timeout.IsCancellationRequested && !ct.IsCancellationRequested)
{
// A wedged send otherwise holds the lock and prevents heartbeat/reconnect from making progress.
_socket.Abort();
throw new TimeoutException($"WebSocket send timed out after {s_sendTimeout.TotalSeconds:0} seconds");
}
finally
{
if (lockTaken)
{
_sendLock.Release();
}
}
}
public async ValueTask<RpcFrame?> ReceiveAsync(CancellationToken ct)
{