fix(link): reconnect after RPC send failures [CI]
This commit is contained in:
1 parent
6cf6b23585
commit
fe049c97fa
2 files changed
+38
-4
No files matched your search
@@ -214,7 +214,7 @@ public sealed class FleetLink
|
|||||||
|
|
||||||
Task cancellation = Task.Delay(Timeout.Infinite, ct);
|
Task cancellation = Task.Delay(Timeout.Infinite, ct);
|
||||||
Task finished = await Task.WhenAny(run, pump, heartbeat, cancellation);
|
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)
|
if (ct.IsCancellationRequested)
|
||||||
{
|
{
|
||||||
try
|
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)
|
if (finished.IsFaulted)
|
||||||
{
|
{
|
||||||
@@ -417,6 +417,11 @@ public sealed class FleetLink
|
|||||||
return $"{taskName} failed: {finished.Exception?.GetBaseException().Message ?? "unknown error"}";
|
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";
|
return finished == run ? "closed by server" : "link task stopped";
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -1,4 +1,5 @@
|
|||||||
using System.Collections.Concurrent;
|
using System.Collections.Concurrent;
|
||||||
|
using System.Net.WebSockets;
|
||||||
using System.Text.Json;
|
using System.Text.Json;
|
||||||
using System.Text.Json.Serialization.Metadata;
|
using System.Text.Json.Serialization.Metadata;
|
||||||
using System.Threading.Channels;
|
using System.Threading.Channels;
|
||||||
@@ -58,6 +59,7 @@ public sealed class RpcSession<TContext> : IAsyncDisposable
|
|||||||
private long _lastReceive;
|
private long _lastReceive;
|
||||||
private int _violations;
|
private int _violations;
|
||||||
private bool _violated;
|
private bool _violated;
|
||||||
|
private string? _transportFailure;
|
||||||
|
|
||||||
public RpcSession(IRpcTransport transport, RpcRouter<TContext> router, TContext context, TimeProvider time, RpcSessionOptions? options = null)
|
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>
|
/// <summary>对端上一次发来任何帧到现在的时间。静默死亡(断电、进程被冻结)的对端不会关闭连接,调用方据此设置空闲超时。</summary>
|
||||||
public TimeSpan IdleFor => _time.GetElapsedTime(Volatile.Read(ref _lastReceive));
|
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>
|
/// <summary>读帧并分发,直到连接关闭或被取消。</summary>
|
||||||
public async Task RunAsync(CancellationToken ct)
|
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);
|
_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) =>
|
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);
|
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)
|
private static RpcMessage? Parse(ReadOnlyMemory<byte> data)
|
||||||
{
|
{
|
||||||
try
|
try
|
||||||
|
|||||||
Reference in new issue
Block a user