Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
8 changes: 8 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,14 @@

## [Unreleased]

### 修复
- **迟到的过期故障重连污染活会话**(v2.3.0 评审 P1-1):垂死旧会话的迟到故障(epoch 已被替换,如 net48 线程池慢导致第二个 `ScheduleReconnect` 延迟到达)此前会在 epoch 校验之前发布 `Retry` 并把 `CurrentSessionId` 归零——完全存活的会话被谎报为重连中、`WaitForConnectedAsync` 悬挂、活会话编号被误判失效。现改为过期故障在任何对外发布之前整体丢弃(gate 内权威复查保留)。回归测试以反射调度过期故障验证。
- **残留的旧会话绑定条目可能在新会话 socket 上错发并报告成功**(评审 P1-2,"绝不跨会话重放"保证的漏洞):`Connected→Connected` 直连拆除(双 `StartAsync` 竞速/Connected 回调内 `Reconnect()` 重入)路径下,旧会话排队条目既不被清扫也无编号校验,会被新 worker 发到新 socket 并以成功收尾。现发送 worker 认领时校验条目 `SessionId`,不属当前会话的绑定条目一律以 `SessionExpiredException` 失败、绝不发送。压测改为按帧内容对账(跨会话错发探测器)。
- 发送失败类型收敛(评审 P2-5/P2-6):停机时清扫先于发送通道完成(队满等待中的条目统一以 `SessionExpiredException` 收尾而非偶发 `ChannelClosedException`);worker 写出失败时 internal 的 `SessionFaultException` 与 `ObjectDisposedException` 统一映射为 `SessionExpiredException`,不再向调用方泄漏非文档类型。
- `Reconnect()` 在 `Start()` 之前调用由 NRE 改为明确的 `InvalidOperationException`(存量问题顺手修复)。
- 未完成帧超时关闭(默认)时不再每读循环拷贝半帧快照(评审 P2-1,2.2.x 路径的小开销)。
- `SendInSessionAsync` 注册竞态窗口的失败以条目携带的同一异常收尾,不再留下未观察的任务异常(评审 P2-4)。

### 变更
- 暂无

Expand Down
2 changes: 1 addition & 1 deletion README.en.md
Original file line number Diff line number Diff line change
Expand Up @@ -166,7 +166,7 @@ A peer that announces a huge frame but never completes it (or a stream with STX
- `MaxIncompleteFrameBufferBytes` (default = frame limit + 4 KB) — a cap in **bytes**: exceeding it disconnects, blocking memory-flooding peers;
- `IncompleteFrameTimeoutMs` (default 0 = off) — a cap in **time**: if a started frame receives no further bytes for this long, the session is torn down and `FrameError` reports `IncompleteFrameTimeout` with a buffer snapshot capped at 8 KB.

The incomplete-frame timeout only counts while a frame is actually in progress: an idle connection with an empty buffer never trips it (each received byte resets it; it clears once a whole frame is cut). It complements `ReceiveIdleTimeoutMs` (see below) — the latter also times total silence, which suits protocols with periodic traffic; for protocols that may stay idle for a long time but must not tolerate a stalled half-frame (e.g. HSMS T8), use the incomplete-frame timeout.
The incomplete-frame timeout only counts while a frame is actually in progress: an idle connection with an empty buffer never trips it (each received byte resets it; it clears once a whole frame is cut). It complements `ReceiveIdleTimeoutMs` (see below) — the latter also times total silence, which suits protocols with periodic traffic; for protocols that may stay idle for a long time but must not tolerate a stalled half-frame (e.g. HSMS T8), use the incomplete-frame timeout. Note its scope is "waiting for further bytes from the network": when `ReceiveQueueCapacity` is set and the consumer stalls completely, the decode loop blocks on the message-channel write and the timeout does not tick during that period (the memory guard `MaxIncompleteFrameBufferBytes` still applies).

### Logging (optional)

Expand Down
2 changes: 1 addition & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -164,7 +164,7 @@ socket 层全量输出(含被丢弃的噪声字节),发送侧按实际写
- `MaxIncompleteFrameBufferBytes`(默认 = 帧上限 + 4KB)——**字节**上限:半帧超过即断线,防内存攻击;
- `IncompleteFrameTimeoutMs`(默认 0 = 关闭)——**时间**上限:半帧开始后连续这么久收不到后续字节即断线,`FrameError` 上报 `IncompleteFrameTimeout` 并携带受 8KB 上限保护的缓冲快照。

未完成帧超时只计"帧已开头、迟迟收不齐"的时间:缓冲为空的静默连接不计时(收到字节即重置,整帧切尽后归零)。它与 `ReceiveIdleTimeoutMs`(见下节)互补——后者在完全没流量时也计时,适合有周期报文的协议;对允许长时间空闲、但半帧卡死必须判死的协议(如 HSMS T8),用未完成帧超时。
未完成帧超时只计"帧已开头、迟迟收不齐"的时间:缓冲为空的静默连接不计时(收到字节即重置,整帧切尽后归零)。它与 `ReceiveIdleTimeoutMs`(见下节)互补——后者在完全没流量时也计时,适合有周期报文的协议;对允许长时间空闲、但半帧卡死必须判死的协议(如 HSMS T8),用未完成帧超时。注意其作用域是"等待网络后续字节"期间:若设置了 `ReceiveQueueCapacity` 且消费端完全停滞,解码循环阻塞在消息通道写入上,此期间不计时(内存防线仍由 `MaxIncompleteFrameBufferBytes` 兜底)。

### 日志(可选)

Expand Down
14 changes: 9 additions & 5 deletions src/StreamFrame/Connection/FrameDecoder.cs
Original file line number Diff line number Diff line change
Expand Up @@ -125,11 +125,15 @@ await _relay.Writer.WriteAsync(
$"未完成帧缓冲 {buffer.Length} 字节超过上限 {_maxIncompleteFrameBytes} 字节。");
}

// 挂表前拷贝半帧快照;缓冲为空(帧刚切尽/尚未开始)时清零,下轮不计时
pendingBytes = buffer.Length;
pendingSnapshot = buffer.IsEmpty
? ReadOnlyMemory<byte>.Empty
: buffer.Slice(0, (int)Math.Min(buffer.Length, ErrorSnapshotBytes)).ToArray();
// 挂表前拷贝半帧快照;缓冲为空(帧刚切尽/尚未开始)时清零,下轮不计时。
// 功能关闭(默认)时完全跳过——半帧挂起期间不为诊断付每次读循环的拷贝开销
if (_incompleteFrameTimeoutMs > 0)
{
pendingBytes = buffer.Length;
pendingSnapshot = buffer.IsEmpty
? ReadOnlyMemory<byte>.Empty
: buffer.Slice(0, (int)Math.Min(buffer.Length, ErrorSnapshotBytes)).ToArray();
}

_reader.AdvanceTo(buffer.Start, buffer.End);

Expand Down
47 changes: 38 additions & 9 deletions src/StreamFrame/Connection/StreamConnection.cs
Original file line number Diff line number Diff line change
Expand Up @@ -368,9 +368,12 @@ private void ConfigureSocket(Socket socket)
}

/// <summary>立即进入重连流程。</summary>
/// <exception cref="InvalidOperationException">连接尚未 <see cref="Start"/>(无可重连的会话)。</exception>
public void Reconnect()
{
ThrowIfDisposed();
if (_lifetimeCts is null)
throw new InvalidOperationException("连接尚未 Start,无法 Reconnect。");
CommunicationStateChanging(ConnectionState.Retry);
}

Expand Down Expand Up @@ -407,6 +410,14 @@ private void ScheduleReconnect(int epoch)

private void CommunicationStateChanging(ConnectionState newState, int? observedEpoch = null)
{
// 迟到的过期故障(会话已被替换,epoch 不匹配):在任何对外可见的发布之前丢弃。
// 若放到 Retry 分支的 gate 内才检查(旧实现),State=Retry 的发布与 CurrentSessionId
// 归零已经污染了完全存活的会话——WaitForConnectedAsync 悬挂、活会话编号被误判失效。
// gate 内的权威复查仍然保留,防检查与发布之间 epoch 变化的窄竞态。
if (newState == ConnectionState.Retry && observedEpoch is { } staleEpoch
&& Volatile.Read(ref _sessionEpoch) != staleEpoch)
return;

// 公共会话编号的线性化点:Connected 对外发布(事件回调/等待器完成)之前完成分配;
// 离开 Connected 的状态(Retry/Disconnected)对外发布之前归零——状态可见时旧会话
// 已不可用于新的会话绑定发送(SendInSessionAsync 以编号不匹配快速失败)。
Expand Down Expand Up @@ -507,7 +518,7 @@ private void StartSession(long sessionId)

_receiveTask = WatchSessionFaultsAsync(ReceiveLoopAsync(pipe.Writer, token, epoch), cts, epoch);
_decodeTask = WatchSessionFaultsAsync(decoder.RunAsync(token), cts, epoch);
_sendWorkerTask = WatchSessionFaultsAsync(SendWorkerAsync(token), cts, epoch);
_sendWorkerTask = WatchSessionFaultsAsync(SendWorkerAsync(token, sessionId), cts, epoch);
}
}

Expand Down Expand Up @@ -536,8 +547,9 @@ private async Task WatchSessionFaultsAsync(Task sessionTask, CancellationTokenSo
/// 会话拆除(Retry/停机)时立即 fault 所有挂起的会话绑定发送:排队中未写出的条目以
/// <see cref="SessionExpiredException"/> 结束,调用方不必空等重连;已认领(正在写出)的条目
/// 由写出路径给出结果。普通条目不受影响(按既有语义留给新会话续发)。
/// 注意:只在真正的拆除点调用——<see cref="StartSession"/> 开头对旧会话的清理不能调用,
/// 否则会误杀"Connected 已发布、新会话任务尚未就绪"窗口内注册的新条目。
/// 只在真正的拆除点调用——<see cref="StartSession"/> 开头对旧会话的清理不能调用,
/// 否则会误杀"Connected 已发布、新会话任务尚未就绪"窗口内注册的新条目;该路径残留的
/// 旧会话条目由发送 worker 认领时的编号校验兜底(见 <see cref="SendWorkerAsync"/>)。
/// </summary>
private void FaultPendingSessionSends()
{
Expand Down Expand Up @@ -669,11 +681,10 @@ private async Task<int> ReceiveWithIdleTimeoutAsync(Socket socket, Memory<byte>
}
}

private async Task SendWorkerAsync(CancellationToken ct)
private async Task SendWorkerAsync(CancellationToken ct, long sessionId)
{
// 发送失败(socket 故障/会话失效)不在此吞掉:上抛给会话守护 → 断线重连,
// 未发送的普通消息留在连接级队列,由重连后的新 worker 继续发送。
// 会话绑定条目已在拆除时被 fault(或调用方已取消):只跳过、不发送、不重放。
var reader = _sendQueue.Reader;
while (await reader.WaitToReadAsync(ct).ConfigureAwait(false))
{
Expand All @@ -685,6 +696,17 @@ private async Task SendWorkerAsync(CancellationToken ct)
continue;
}

// 编号校验(防跨会话错发):绑定旧会话的残留条目绝不能发到当前会话的 socket 上。
// 覆盖 Connected→Connected 直连拆除(StartSession 内的 StopSessionCore 刻意不
// 清扫,保护"刚注册的新条目")等一切条目跨会话存活的路径——
// "会话绑定消息绝不转移到新会话重放"的保证在此闭合
if (entry.SessionId != sessionId)
{
entry.TryExpire(new SessionExpiredException(
entry.SessionId, $"会话 {entry.SessionId} 已失效(条目跨会话残留),消息未写出。"));
continue;
}

// 认领 = 调用方取消的提交点:认领成功后写入只受会话令牌控制
if (!entry.TryClaimForSend())
continue;
Expand All @@ -702,7 +724,12 @@ private async Task SendWorkerAsync(CancellationToken ct)
}
catch (Exception ex)
{
entry.Fault(ex);
// 对调用方只暴露文档承诺的失败类型:internal 的 SessionFaultException
// (会话拆除竞速时 SendRawAsync 的"无可用连接")与 ObjectDisposedException
// (socket 已释放)统一收敛为会话失效
entry.Fault(ex is SessionFaultException or ObjectDisposedException
? new SessionExpiredException(entry.SessionId, $"会话 {entry.SessionId} 在整帧写出前终止。", ex)
: ex);
throw;
}
}
Expand Down Expand Up @@ -808,11 +835,13 @@ public async Task SendInSessionAsync(long sessionId, TMessage message, Cancellat
_pendingSessionSends.TryAdd(entry, 0);
try
{
// 双检:注册与"拆除归零编号"的竞态——先注册、再复核,失效则立即 fault 本条目
// 双检:注册与"拆除归零编号"的竞态——先注册、再复核,失效则以条目携带的同一
// 异常收尾(同时观察 TCS,避免留下未观察的任务异常)
if (Volatile.Read(ref _currentSessionId) != sessionId)
{
entry.TryExpire(NewSessionExpired(sessionId));
throw NewSessionExpired(sessionId);
await entry.CompletionTask.ConfigureAwait(false);
return;
}

// 调用方取消:提交点(worker 认领)之前使条目退出;认领之后无副作用
Expand Down Expand Up @@ -998,8 +1027,8 @@ private void Shutdown()

_lifetimeRegistration.Dispose();
_lifetimeCts?.Cancel(); // 停止连接/监听重试循环
FaultPendingSessionSends(); // 停机:挂起的会话绑定发送全部以会话失效收尾(先于通道完成,统一失败类型)
_sendQueue.Writer.TryComplete();
FaultPendingSessionSends(); // 停机:挂起的会话绑定发送全部以会话失效收尾

StopSession();

Expand Down
3 changes: 2 additions & 1 deletion src/StreamFrame/IStreamConnection.cs
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,8 @@ public interface IStreamConnection<TMessage> : IAsyncDisposable
/// <summary>
/// 帧层诊断事件:帧内容解码失败(<see cref="FrameErrorKind.DecodeFailed"/>)、被定界器
/// 丢弃的噪声字节(<see cref="FrameErrorKind.DiscardedByResync"/>)、未完成帧缓冲超限
/// (<see cref="FrameErrorKind.IncompleteFrameOverflow"/>)。
/// (<see cref="FrameErrorKind.IncompleteFrameOverflow"/>)、未完成帧超时
/// (<see cref="FrameErrorKind.IncompleteFrameTimeout"/>)。
/// 事件携带的 <see cref="FrameErrorEventArgs.Bytes"/> 是已拷贝的字节,回调后可安全留存。
/// </summary>
event EventHandler<FrameErrorEventArgs>? FrameError;
Expand Down
5 changes: 5 additions & 0 deletions src/StreamFrame/StreamConnectionOptions.cs
Original file line number Diff line number Diff line change
Expand Up @@ -91,6 +91,11 @@ public sealed class StreamConnectionOptions
/// 与 <see cref="ReceiveIdleTimeoutMs"/> 的区别:后者在"完全没有字节"时也计时(要求连接
/// 必须有周期流量);本选项只在"帧已开头、迟迟收不齐"时计时——缓冲为空的正常静默不会触发,
/// 一帧完整切出后重新归零。适合允许长时间空闲、但半帧卡死必须判死的长度前缀协议(如 HSMS T8)。
///
/// 作用域限制:计时窗口是解码循环等待网络后续字节期间。若设置了
/// <see cref="ReceiveQueueCapacity"/> 且消费端完全停滞,解码循环会阻塞在消息通道写入上、
/// 不在等待网络字节的状态——此期间本超时不计时(半帧的内存防线仍由
/// <see cref="MaxIncompleteFrameBufferBytes"/> 兜底)。
/// </summary>
public int IncompleteFrameTimeoutMs { get; set; }

Expand Down
Loading
Loading