From be97e246931a4fc324545980e46ab9925f0ce866 Mon Sep 17 00:00:00 2001
From: JustMe <mr.toing@outlook.com>
Date: Sun, 11 Oct 2026 19:54:14 +0800
Subject: [PATCH] fix(link): recover stale websocket sessions [CI]

---
 backend/fleet/src/Fleet.Link/FleetLink.cs  | 22 ++++++++++++++++---
 backend/fleet/src/Fleet.Link/Transports.cs | 25 +++++++++++++++++++---
 2 files changed, 41 insertions(+), 6 deletions(-)

diff --git a/backend/fleet/src/Fleet.Link/FleetLink.cs b/backend/fleet/src/Fleet.Link/FleetLink.cs
index 5d7fbe6..3c205d2 100644
--- a/backend/fleet/src/Fleet.Link/FleetLink.cs
+++ b/backend/fleet/src/Fleet.Link/FleetLink.cs
@@ -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";
+    }
 }
diff --git a/backend/fleet/src/Fleet.Link/Transports.cs b/backend/fleet/src/Fleet.Link/Transports.cs
index 6b712c1..14d864f 100644
--- a/backend/fleet/src/Fleet.Link/Transports.cs
+++ b/backend/fleet/src/Fleet.Link/Transports.cs
@@ -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,14 +20,32 @@ 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
         {
-            _sendLock.Release();
+            if (lockTaken)
+            {
+                _sendLock.Release();
+            }
         }
     }