From fe049c97fadae18132a9772dfce183094da6eb4a Mon Sep 17 00:00:00 2001
From: JustMe <mr.toing@outlook.com>
Date: Mon, 12 Oct 2026 00:21:12 +0800
Subject: [PATCH] fix(link): reconnect after RPC send failures [CI]

---
 backend/fleet/src/Fleet.Link/FleetLink.cs |  9 +++++--
 backend/fleet/src/Fleet.Rpc/RpcSession.cs | 33 +++++++++++++++++++++--
 2 files changed, 38 insertions(+), 4 deletions(-)

diff --git a/backend/fleet/src/Fleet.Link/FleetLink.cs b/backend/fleet/src/Fleet.Link/FleetLink.cs
index 3c205d2..07755c2 100644
--- a/backend/fleet/src/Fleet.Link/FleetLink.cs
+++ b/backend/fleet/src/Fleet.Link/FleetLink.cs
@@ -214,7 +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);
+            string finishedReason = DescribeFinishedTask(finished, run, pump, heartbeat, session.TransportFailure);
             if (ct.IsCancellationRequested)
             {
                 try
@@ -409,7 +409,7 @@ public sealed class FleetLink
         }
     }
 
-    private static string DescribeFinishedTask(Task finished, Task run, Task pump, Task heartbeat)
+    private static string DescribeFinishedTask(Task finished, Task run, Task pump, Task heartbeat, string? transportFailure)
     {
         if (finished.IsFaulted)
         {
@@ -417,6 +417,11 @@ public sealed class FleetLink
             return $"{taskName} failed: {finished.Exception?.GetBaseException().Message ?? "unknown error"}";
         }
 
+        if (finished == run && transportFailure is not null)
+        {
+            return "transport send failed: " + transportFailure;
+        }
+
         return finished == run ? "closed by server" : "link task stopped";
     }
 }
diff --git a/backend/fleet/src/Fleet.Rpc/RpcSession.cs b/backend/fleet/src/Fleet.Rpc/RpcSession.cs
index ec1770a..918807a 100644
--- a/backend/fleet/src/Fleet.Rpc/RpcSession.cs
+++ b/backend/fleet/src/Fleet.Rpc/RpcSession.cs
@@ -1,4 +1,5 @@
 using System.Collections.Concurrent;
+using System.Net.WebSockets;
 using System.Text.Json;
 using System.Text.Json.Serialization.Metadata;
 using System.Threading.Channels;
@@ -58,6 +59,7 @@ public sealed class RpcSession<TContext> : IAsyncDisposable
     private long _lastReceive;
     private int _violations;
     private bool _violated;
+    private string? _transportFailure;
 
     public RpcSession(IRpcTransport transport, RpcRouter<TContext> router, TContext context, TimeProvider time, RpcSessionOptions? options = null)
     {
@@ -76,6 +78,9 @@ public sealed class RpcSession<TContext> : IAsyncDisposable
     /// <summary>对端上一次发来任何帧到现在的时间。静默死亡（断电、进程被冻结）的对端不会关闭连接，调用方据此设置空闲超时。</summary>
     public TimeSpan IdleFor => _time.GetElapsedTime(Volatile.Read(ref _lastReceive));
 
+    /// <summary>The first transport send failure observed by this session, if any.</summary>
+    public string? TransportFailure => Volatile.Read(ref _transportFailure);
+
     /// <summary>读帧并分发，直到连接关闭或被取消。</summary>
     public async Task RunAsync(CancellationToken ct)
     {
@@ -197,11 +202,11 @@ public sealed class RpcSession<TContext> : IAsyncDisposable
             _router.CheckSend(message.Method, message.Id is null ? MethodKind.Notification : MethodKind.Request);
         }
 
-        return _transport.SendAsync(new RpcFrame(FrameKind.Text, JsonSerializer.SerializeToUtf8Bytes(message, FleetRpcContext.Default.RpcMessage)), ct);
+        return SendTransportAsync(new RpcFrame(FrameKind.Text, JsonSerializer.SerializeToUtf8Bytes(message, FleetRpcContext.Default.RpcMessage)), ct);
     }
 
     public ValueTask SendBinaryAsync(ReadOnlyMemory<byte> frame, CancellationToken ct) =>
-        _transport.SendAsync(new RpcFrame(FrameKind.Binary, frame), ct);
+        SendTransportAsync(new RpcFrame(FrameKind.Binary, frame), ct);
 
     public ValueTask CloseAsync(string reason, CancellationToken ct) => _transport.CloseAsync(reason, ct);
 
@@ -239,6 +244,30 @@ public sealed class RpcSession<TContext> : IAsyncDisposable
         }
     }
 
+    private async ValueTask SendTransportAsync(RpcFrame frame, CancellationToken ct)
+    {
+        try
+        {
+            await _transport.SendAsync(frame, ct);
+        }
+        catch (Exception ex) when (!ct.IsCancellationRequested && ex is WebSocketException or IOException or TimeoutException or ObjectDisposedException)
+        {
+            Interlocked.CompareExchange(ref _transportFailure, ex.Message, null);
+            try
+            {
+                // A failed send must also stop the receive loop. Otherwise callers that log and swallow this
+                // exception can leave the session marked connected forever on an already-aborted socket.
+                _stop.Cancel();
+            }
+            catch (ObjectDisposedException)
+            {
+                // Disposal raced with the failed send.
+            }
+
+            throw;
+        }
+    }
+
     private static RpcMessage? Parse(ReadOnlyMemory<byte> data)
     {
         try