diff --git a/CHANGELOG.md b/CHANGELOG.md index 457ebef..5f9281b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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）。 + ### 变更 - 暂无 diff --git a/README.en.md b/README.en.md index 27bffdf..85d375d 100644 --- a/README.en.md +++ b/README.en.md @@ -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) diff --git a/README.md b/README.md index 538b431..810b599 100644 --- a/README.md +++ b/README.md @@ -164,7 +164,7 @@ socket 层全量输出（含被丢弃的噪声字节），发送侧按实际写 - `MaxIncompleteFrameBufferBytes`（默认 = 帧上限 + 4KB）——**字节**上限：半帧超过即断线，防内存攻击； - `IncompleteFrameTimeoutMs`（默认 0 = 关闭）——**时间**上限：半帧开始后连续这么久收不到后续字节即断线，`FrameError` 上报 `IncompleteFrameTimeout` 并携带受 8KB 上限保护的缓冲快照。 -未完成帧超时只计"帧已开头、迟迟收不齐"的时间：缓冲为空的静默连接不计时（收到字节即重置，整帧切尽后归零）。它与 `ReceiveIdleTimeoutMs`（见下节）互补——后者在完全没流量时也计时，适合有周期报文的协议；对允许长时间空闲、但半帧卡死必须判死的协议（如 HSMS T8），用未完成帧超时。 +未完成帧超时只计"帧已开头、迟迟收不齐"的时间：缓冲为空的静默连接不计时（收到字节即重置，整帧切尽后归零）。它与 `ReceiveIdleTimeoutMs`（见下节）互补——后者在完全没流量时也计时，适合有周期报文的协议；对允许长时间空闲、但半帧卡死必须判死的协议（如 HSMS T8），用未完成帧超时。注意其作用域是"等待网络后续字节"期间：若设置了 `ReceiveQueueCapacity` 且消费端完全停滞，解码循环阻塞在消息通道写入上，此期间不计时（内存防线仍由 `MaxIncompleteFrameBufferBytes` 兜底）。 ### 日志（可选） diff --git a/src/StreamFrame/Connection/FrameDecoder.cs b/src/StreamFrame/Connection/FrameDecoder.cs index 9a62978..e64b303 100644 --- a/src/StreamFrame/Connection/FrameDecoder.cs +++ b/src/StreamFrame/Connection/FrameDecoder.cs @@ -125,11 +125,15 @@ await _relay.Writer.WriteAsync( $"未完成帧缓冲 {buffer.Length} 字节超过上限 {_maxIncompleteFrameBytes} 字节。"); } - // 挂表前拷贝半帧快照；缓冲为空（帧刚切尽/尚未开始）时清零，下轮不计时 - pendingBytes = buffer.Length; - pendingSnapshot = buffer.IsEmpty - ? ReadOnlyMemory.Empty - : buffer.Slice(0, (int)Math.Min(buffer.Length, ErrorSnapshotBytes)).ToArray(); + // 挂表前拷贝半帧快照；缓冲为空（帧刚切尽/尚未开始）时清零，下轮不计时。 + // 功能关闭（默认）时完全跳过——半帧挂起期间不为诊断付每次读循环的拷贝开销 + if (_incompleteFrameTimeoutMs > 0) + { + pendingBytes = buffer.Length; + pendingSnapshot = buffer.IsEmpty + ? ReadOnlyMemory.Empty + : buffer.Slice(0, (int)Math.Min(buffer.Length, ErrorSnapshotBytes)).ToArray(); + } _reader.AdvanceTo(buffer.Start, buffer.End); diff --git a/src/StreamFrame/Connection/StreamConnection.cs b/src/StreamFrame/Connection/StreamConnection.cs index 1c043aa..cb4592c 100644 --- a/src/StreamFrame/Connection/StreamConnection.cs +++ b/src/StreamFrame/Connection/StreamConnection.cs @@ -368,9 +368,12 @@ private void ConfigureSocket(Socket socket) } /// 立即进入重连流程。 + /// 连接尚未 （无可重连的会话）。 public void Reconnect() { ThrowIfDisposed(); + if (_lifetimeCts is null) + throw new InvalidOperationException("连接尚未 Start，无法 Reconnect。"); CommunicationStateChanging(ConnectionState.Retry); } @@ -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 以编号不匹配快速失败）。 @@ -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); } } @@ -536,8 +547,9 @@ private async Task WatchSessionFaultsAsync(Task sessionTask, CancellationTokenSo /// 会话拆除（Retry/停机）时立即 fault 所有挂起的会话绑定发送：排队中未写出的条目以 /// 结束，调用方不必空等重连；已认领（正在写出）的条目 /// 由写出路径给出结果。普通条目不受影响（按既有语义留给新会话续发）。 - /// 注意：只在真正的拆除点调用—— 开头对旧会话的清理不能调用， - /// 否则会误杀"Connected 已发布、新会话任务尚未就绪"窗口内注册的新条目。 + /// 只在真正的拆除点调用—— 开头对旧会话的清理不能调用， + /// 否则会误杀"Connected 已发布、新会话任务尚未就绪"窗口内注册的新条目；该路径残留的 + /// 旧会话条目由发送 worker 认领时的编号校验兜底（见 ）。 /// private void FaultPendingSessionSends() { @@ -669,11 +681,10 @@ private async Task ReceiveWithIdleTimeoutAsync(Socket socket, Memory } } - 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)) { @@ -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; @@ -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; } } @@ -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 认领）之前使条目退出；认领之后无副作用 @@ -998,8 +1027,8 @@ private void Shutdown() _lifetimeRegistration.Dispose(); _lifetimeCts?.Cancel(); // 停止连接/监听重试循环 + FaultPendingSessionSends(); // 停机：挂起的会话绑定发送全部以会话失效收尾（先于通道完成，统一失败类型） _sendQueue.Writer.TryComplete(); - FaultPendingSessionSends(); // 停机：挂起的会话绑定发送全部以会话失效收尾 StopSession(); diff --git a/src/StreamFrame/IStreamConnection.cs b/src/StreamFrame/IStreamConnection.cs index c06c3b3..7ec61c6 100644 --- a/src/StreamFrame/IStreamConnection.cs +++ b/src/StreamFrame/IStreamConnection.cs @@ -29,7 +29,8 @@ public interface IStreamConnection : IAsyncDisposable /// /// 帧层诊断事件：帧内容解码失败（）、被定界器 /// 丢弃的噪声字节（）、未完成帧缓冲超限 - /// （）。 + /// （）、未完成帧超时 + /// （）。 /// 事件携带的 是已拷贝的字节，回调后可安全留存。 /// event EventHandler? FrameError; diff --git a/src/StreamFrame/StreamConnectionOptions.cs b/src/StreamFrame/StreamConnectionOptions.cs index 664d1b2..2baf03e 100644 --- a/src/StreamFrame/StreamConnectionOptions.cs +++ b/src/StreamFrame/StreamConnectionOptions.cs @@ -91,6 +91,11 @@ public sealed class StreamConnectionOptions /// 与 的区别：后者在"完全没有字节"时也计时（要求连接 /// 必须有周期流量）；本选项只在"帧已开头、迟迟收不齐"时计时——缓冲为空的正常静默不会触发， /// 一帧完整切出后重新归零。适合允许长时间空闲、但半帧卡死必须判死的长度前缀协议（如 HSMS T8）。 + /// + /// 作用域限制：计时窗口是解码循环等待网络后续字节期间。若设置了 + /// 且消费端完全停滞，解码循环会阻塞在消息通道写入上、 + /// 不在等待网络字节的状态——此期间本超时不计时（半帧的内存防线仍由 + /// 兜底）。 /// public int IncompleteFrameTimeoutMs { get; set; } diff --git a/test/StreamFrame.Tests/SessionAwareConnectionTests.cs b/test/StreamFrame.Tests/SessionAwareConnectionTests.cs index 5ba77fb..2e8e0be 100644 --- a/test/StreamFrame.Tests/SessionAwareConnectionTests.cs +++ b/test/StreamFrame.Tests/SessionAwareConnectionTests.cs @@ -261,11 +261,11 @@ public async Task SessionTeardown_PendingBoundSends_FailPromptly() server.Reconnect(); // 会话拆除：无需等重连完成，挂起发送必须立即失败 + // 排队中的条目只经拆除清扫终结：失败类型必须精确为会话失效 foreach (var task in tasks) { var ex = await AwaitFailureAsync(task, 3000, "挂起的会话绑定发送及时失败"); - Assert.True(ex is SessionExpiredException or SocketException or OperationCanceledException, - $"失败类型应为会话失效/socket 故障，实际 {ex.GetType().Name}。"); + Assert.IsType(ex); } // 正在写出的条目是拆除竞争的合法双结局：整帧恰好写完 → 成功；否则会话失效失败。 @@ -303,8 +303,7 @@ public async Task SessionSwitch_BoundSendFails_NoReplay_PlainSendStillReplays() foreach (var task in bound) { var ex = await AwaitFailureAsync(task, 3000, "会话切换后旧绑定发送失败"); - Assert.True(ex is SessionExpiredException or SocketException or OperationCanceledException, - $"失败类型意外：{ex.GetType().Name}。"); + Assert.IsType(ex); // 排队条目只经拆除清扫终结 } // 正在写出的条目：拆除竞争的合法双结局（整帧恰好写完→成功；否则失效失败），不悬挂即可 @@ -363,8 +362,7 @@ public async Task Dispose_PendingBoundSends_AllFail_Promptly() foreach (var task in tasks) { var ex = await AwaitFailureAsync(task, 3000, "Dispose 后挂起发送异常收尾"); - Assert.True(ex is SessionExpiredException or SocketException or OperationCanceledException, - $"失败类型应为会话失效/socket 故障，实际 {ex.GetType().Name}。"); + Assert.IsType(ex); // 排队条目只经停机清扫终结（清扫先于通道完成） } // 正在写出的条目：拆除竞争的合法双结局（写完→成功；否则失效失败），不悬挂即可 @@ -409,6 +407,47 @@ await Assert.ThrowsAnyAsync( await AssertNoBytesAsync(stream); } + [Fact] + public async Task StaleGhostReconnect_DoesNotPolluteLiveSession() + { + // 评审 P1-1 回归：垂死旧会话的迟到故障（epoch 已过期）必须被整体丢弃—— + // 不得把活会话的 State 污染成 Retry、不得把 CurrentSessionId 归零。 + // 这种延迟到达只能经反射调度模拟（公共 API 无法构造） + var port = GetFreePort(); + await using var server = CreateServer(port); + server.Start(CancellationToken.None); + + long id1; + using (var client1 = new TcpClient()) + { + await ConnectWithRetryAsync(client1, port); + await WaitForStateAsync(server, s => s == ConnectionState.Connected); + id1 = server.CurrentSessionId; + } + + await WaitForStateAsync(server, s => s != ConnectionState.Connected); + using (var client2 = new TcpClient()) + { + await ConnectWithRetryAsync(client2, port); + await WaitForStateAsync(server, s => s == ConnectionState.Connected, timeoutMs: 8000); + var id2 = server.CurrentSessionId; + Assert.True(id2 > id1); + + var schedule = typeof(StreamConnection).GetMethod( + "ScheduleReconnect", System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance); + Assert.NotNull(schedule); + schedule!.Invoke(server, new object[] { 1 }); // 旧会话（epoch=1）的幽灵故障，早已过期 + + await Task.Delay(300); + Assert.Equal(ConnectionState.Connected, server.State); // 不被谎报为 Retry + Assert.Equal(id2, server.CurrentSessionId); // 活会话编号不被归零 + + var task = server.SendInSessionAsync(id2, "alive"); + Assert.Equal("alive", await ReadFrameAsync(client2.GetStream())); + await AwaitDoneAsync(task, 3000, "活会话发送"); + } + } + [Fact] public async Task CallerCancel_AfterCommit_HasNoEffect() { @@ -653,7 +692,7 @@ public async Task TeardownVersusRegister_Stress_AllTasksEnd_NoCrossSessionSend() }); var sends = Enumerable.Range(0, SendsPerRound) - .Select(_ => server.SendInSessionAsync(server.CurrentSessionId, $"r{round}")) + .Select(i => server.SendInSessionAsync(server.CurrentSessionId, $"r{round}-{i}")) .ToArray(); // 与拆除并发：不等发送结果，直接切换会话（按轮次交替两种拆除路径） @@ -681,7 +720,8 @@ public async Task TeardownVersusRegister_Stress_AllTasksEnd_NoCrossSessionSend() // 成功核对（仅 Reconnect 轮次：对端全程在读，成功帧必然可达且完整）： // 客户端主动 Dispose 轮次不对账——本地关闭时未读数据随 RST 丢弃属 TCP 语义， - // "写入本机 socket"之外的投递结果不在本 API 的公共保证内 + // "写入本机 socket"之外的投递结果不在本 API 的公共保证内。 + // 按帧内容对账：每个完整帧的负载必须带本轮标签——检出任何跨会话/跨轮次的错发 if (round % 2 == 0) { lock (peerBytes) @@ -695,6 +735,8 @@ public async Task TeardownVersusRegister_Stress_AllTasksEnd_NoCrossSessionSend() Assert.True(length is > 0 and < 1024, $"帧长度异常：{length}。"); if (offset + 4 + length > bytes.Length) break; // 拆除时刻正在写的半帧（其任务必然已失败） + var payload = Encoding.UTF8.GetString(bytes, offset + 4, length); + Assert.StartsWith($"r{round}-", payload, StringComparison.Ordinal); // 错发探测器（P1-2 场景） offset += 4 + length; frames++; }