diff --git a/CHANGELOG.md b/CHANGELOG.md index e0aa634..7434cb5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,9 @@ ## [Unreleased] +### 新增 +- **未完成帧超时（`IncompleteFrameTimeoutMs`，默认 0 = 关闭）**（#38）：帧已开头、缓冲里留着半帧字节，但连续这么久未收到后续字节时判定会话失效并断线重连——与 `ReceiveIdleTimeoutMs`（完全静默也计时，要求周期流量）互补，适合允许长时间空闲、但半帧卡死必须判死的长度前缀协议（如 HSMS T8）。计时只在半帧进行中生效：缓冲为空不计时、收到新字节即重置、整帧切尽后归零；超时经 `FrameError` 上报新类别 `IncompleteFrameTimeout` 并携带受 8KB 上限保护的缓冲快照。纯增量、默认行为不变。 + ### 变更 - LOGO 资产对齐 social preview 定稿：图标符号改为与横幅裸符号完全同构——青色波改为**自左下低位上穿环洞**（旧版为下行穿过且环洞被底色矩形遮盖、波不显于洞内），环壁加粗至约 21.5% 边长、洞圆角收小，背景由纯色 `#4338CA` 改为横幅同款对角渐变 `#1E1B4B → #3730A3`；`docs/logo/` 全套按 social-preview.png 像素实测几何重绘，PNG 由 SVG 母版经 resvg 4096 栅格化降采样产出（单一矢量源），README 双语版经 raw 地址引用自动生效，NuGet 包图标随下个版本发布生效（social preview 横幅本身不变）。 diff --git a/README.en.md b/README.en.md index fe76915..84b9848 100644 --- a/README.en.md +++ b/README.en.md @@ -95,7 +95,7 @@ public interface ICodec - **Auto-reconnect**: `Connecting → Connected → Retry` state machine; optional exponential backoff (`MaxRetryDelayMs`, doubling with cap + ±20% jitter, auto-reset on success); `GetMessages` is a stable stream across reconnections — received messages are not lost and enumeration does not break when the connection drops and recovers - **Start & stop**: the `ct` passed to `Start(ct)` is the connection's **lifetime token** — cancelling it stops connection/reconnection and tears the link down (state enters the terminal `Disconnected`, `GetMessages` completes naturally; create a new connection afterwards). `DisposeAsync` does the same. After shutdown, `SendAsync` throws `ChannelClosedException` - **Waiting for readiness**: `await conn.WaitForConnectedAsync(ct)` — completes immediately when connected; otherwise waits for the next successful connection (or cancellation/dispose). No state polling, no `Task.Delay` guessing -- **Robustness**: payload decode failures, incomplete-frame overflows, send failures, and receive idle timeouts all invalidate the session and rebuild it automatically (no more "connection looks alive while messages silently vanish") +- **Robustness**: payload decode failures, incomplete-frame overflows/timeouts, send failures, and receive idle timeouts all invalidate the session and rebuild it automatically (no more "connection looks alive while messages silently vanish") - **Liveness detection (optional)**: TCP KeepAlive and receive idle timeout to catch half-open connections (peer power loss / unplugged cable) - **Events**: `ConnectionChanged` state changes, `FrameError` frame-level diagnostics, `RawBytesReceived/Sent` raw bytes (HEX debugging) - **Backpressure**: bounded send queue — `SendAsync` waits when full; the receive side buffers without limit by default, or set `ReceiveQueueCapacity` — decoding pauses when the consumer lags, propagating TCP backpressure to the peer and preventing unbounded memory growth @@ -112,6 +112,7 @@ client.FrameError += (_, e) => // e.Kind: DecodeFailed (frame intact, payload parsing failed) // DiscardedByResync (bytes discarded as noise by the framer) // IncompleteFrameOverflow (incomplete-frame buffer over its limit) + // IncompleteFrameTimeout (incomplete frame timed out waiting for the rest) // e.Bytes: copied; safe to retain long-term // e.Exception: the original exception for DecodeFailed Console.WriteLine($"[{e.Kind}] {Convert.ToHexString(e.Bytes.Span)} {e.Exception?.Message}"); @@ -131,7 +132,12 @@ Full output from the socket layer (including noise bytes later discarded by the ### Incomplete-frame protection -A peer that announces a huge frame but never completes it (or a stream with STX but no closing ETX) would hold the buffer forever. `MaxIncompleteFrameBufferBytes` (default = frame limit + 4 KB) puts a hard cap on half-frames: exceeding it disconnects and reports via `FrameError`. +A peer that announces a huge frame but never completes it (or a stream with STX but no closing ETX) would hold the buffer forever. Two complementary guards: + +- `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. ### Logging (optional) @@ -164,6 +170,8 @@ _ = Task.Run(async () => When the peer dies silently (power loss / unplugged cable, no FIN/RST), the idle timeout kills the session and auto-reconnect kicks in; if the peer is unreachable, reconnection keeps failing and the state stays in `Connecting`/`Retry`. +Choosing between the two receive timeouts: `ReceiveIdleTimeoutMs` demands **periodic traffic** — use it when the protocol has heartbeats/periodic reports (it also covers half-open connections); when the protocol may stay silent for long stretches (traffic only expected while a frame is in progress), use `IncompleteFrameTimeoutMs` instead — silence is not a fault, only a stalled half-frame is. + ## Performance Measured with BenchmarkDotNet (details and how to reproduce in [bench/README.md](bench/README.md)): diff --git a/README.md b/README.md index d524a25..fd6b4dd 100644 --- a/README.md +++ b/README.md @@ -95,7 +95,7 @@ public interface ICodec - **自动重连**：`Connecting → Connected → Retry` 状态机；可选指数退避（`MaxRetryDelayMs`，连续失败倍增封顶 + ±20% 抖动，连接成功自动复位）；`GetMessages` 是跨重连的稳定消息流——断线重连后已收消息不丢、枚举不中断 - **启动与停止**：`Start(ct)` 的 `ct` 是连接的**生命周期令牌**——取消它会停止连接/重连并拆线（状态进入 `Disconnected` 终态，`GetMessages` 自然结束，之后需新建连接）；`DisposeAsync` 与之等效。停机后 `SendAsync` 抛 `ChannelClosedException` - **等待连接就绪**：`await conn.WaitForConnectedAsync(ct)`——已连接立即完成，未连接时等到下次连接成功（或取消/Dispose），不用轮询状态、不用 `Task.Delay` 盲等 -- **健壮性**：帧内容解码失败、未完成帧超限、发送失败、接收空闲超时都会判定会话失效并自动重建（不再产生"连接看似存活、消息静默消失"的假活） +- **健壮性**：帧内容解码失败、未完成帧超限/超时、发送失败、接收空闲超时都会判定会话失效并自动重建（不再产生"连接看似存活、消息静默消失"的假活） - **活性探测（可选）**：TCP KeepAlive 与接收空闲超时，兜底半开连接（对端断电/拔线） - **事件**：`ConnectionChanged` 状态变化、`FrameError` 帧层诊断、`RawBytesReceived/Sent` 原始字节（HEX 调试） - **发送背压**：有界发送队列，队列满时 `SendAsync` 自动等待；接收侧默认无上限缓冲，可用 `ReceiveQueueCapacity` 设上限——消费慢时解码暂停、TCP 背压自然传导到对端，防内存无限增长 @@ -112,6 +112,7 @@ client.FrameError += (_, e) => // e.Kind: DecodeFailed（帧完整但内容解析失败） // DiscardedByResync（被定界器当作噪声丢弃的字节） // IncompleteFrameOverflow（未完成帧缓冲超限） + // IncompleteFrameTimeout（未完成帧超时：半帧迟迟收不齐） // e.Bytes: 已拷贝，可安全长期留存 // e.Exception: DecodeFailed 时的原始异常 Console.WriteLine($"[{e.Kind}] {Convert.ToHexString(e.Bytes.Span)} {e.Exception?.Message}"); @@ -131,7 +132,12 @@ socket 层全量输出（含被丢弃的噪声字节），发送侧按实际写 ### 未完成帧防护 -对端声明一个超长帧却永远不补齐（或 STX/ETX 流中只有 STX 没有闭合），会无限占用缓冲。`MaxIncompleteFrameBufferBytes`（默认 = 帧上限 + 4KB）给"等不齐的半帧"设了硬上限，超限即断线并通过 `FrameError` 上报。 +对端声明一个超长帧却永远不补齐（或 STX/ETX 流中只有 STX 没有闭合），会无限占用缓冲。两条互补的防线： + +- `MaxIncompleteFrameBufferBytes`（默认 = 帧上限 + 4KB）——**字节**上限：半帧超过即断线，防内存攻击； +- `IncompleteFrameTimeoutMs`（默认 0 = 关闭）——**时间**上限：半帧开始后连续这么久收不到后续字节即断线，`FrameError` 上报 `IncompleteFrameTimeout` 并携带受 8KB 上限保护的缓冲快照。 + +未完成帧超时只计"帧已开头、迟迟收不齐"的时间：缓冲为空的静默连接不计时（收到字节即重置，整帧切尽后归零）。它与 `ReceiveIdleTimeoutMs`（见下节）互补——后者在完全没流量时也计时，适合有周期报文的协议；对允许长时间空闲、但半帧卡死必须判死的协议（如 HSMS T8），用未完成帧超时。 ### 日志（可选） @@ -163,6 +169,8 @@ _ = Task.Run(async () => 对端"猝死"（断电/拔线，无 FIN/RST）时静默超限 → 会话判定死亡 → 自动重连；对端不可达时重连持续失败，状态停留在 `Connecting`/`Retry`。 +两种接收超时的取舍：`ReceiveIdleTimeoutMs` 要求连接**必须有周期流量**——协议有心跳/周期上报时用它（还能兜底半开连接）；协议允许长时间静默（只在有帧进行中才该有流量）时改用 `IncompleteFrameTimeoutMs`，静默不算故障、半帧卡死才判死。 + ## 性能 BenchmarkDotNet 实测（详见 [bench/README.md](bench/README.md)，可本地复现）： diff --git a/src/StreamFrame/Connection/FrameDecoder.cs b/src/StreamFrame/Connection/FrameDecoder.cs index be2f47c..7adb4da 100644 --- a/src/StreamFrame/Connection/FrameDecoder.cs +++ b/src/StreamFrame/Connection/FrameDecoder.cs @@ -9,23 +9,26 @@ namespace StreamFrame; /// 半包靠 AdvanceTo 保留未消费字节；粘包靠循环切尽所有完整帧。 /// /// 生命周期约定一：本类型不负责消息通道（relay）的完成——通道归连接所有， -/// 跨会话复用，仅在连接 Dispose 时完成。会话级失败（解码失败、不完整帧超限）以 -/// 上抛，由连接层决定断线重连。 +/// 跨会话复用，仅在连接 Dispose 时完成。会话级失败（解码失败、不完整帧超限、 +/// 未完成帧超时）以 上抛，由连接层决定断线重连。 /// /// 生命周期约定二：ReadAsync 不绑定会话取消令牌（取消中的 ReadAsync 会把 Pipe 留在 /// "读取进行中"状态，无法再 TryRead）。退出由流结束（writer.CompleteAsync）或 /// CancelPendingRead 驱动；退出前把已缓冲的完整帧全部投递，保证"收到的字节必达"。 +/// 唯一例外：未完成帧超时使用独立的超时令牌（不链会话取消），超时后本循环直接以 +/// 会话故障收尾、不再复用该 PipeReader——因此不会有"进行中的读取残留到下一次读取"。 /// internal sealed class FrameDecoder { - /// 上报 IncompleteFrameOverflow 时携带的缓冲快照上限，避免事件本身复制超长数据。 - private const int OverflowSnapshotBytes = 8192; + /// 上报 IncompleteFrameOverflow / IncompleteFrameTimeout 时携带的缓冲快照上限，避免事件本身复制超长数据。 + private const int ErrorSnapshotBytes = 8192; private readonly PipeReader _reader; private readonly IFramer _framing; private readonly ICodec _codec; private readonly Channel _relay; private readonly int _maxIncompleteFrameBytes; + private readonly int _incompleteFrameTimeoutMs; private readonly DecodeErrorPolicy _decodeErrorPolicy; private readonly Action? _onFrameError; @@ -35,6 +38,7 @@ public FrameDecoder( ICodec codec, Channel relay, int maxIncompleteFrameBytes, + int incompleteFrameTimeoutMs, DecodeErrorPolicy decodeErrorPolicy, Action? onFrameError) { @@ -43,17 +47,42 @@ public FrameDecoder( _codec = codec; _relay = relay; _maxIncompleteFrameBytes = maxIncompleteFrameBytes; + _incompleteFrameTimeoutMs = incompleteFrameTimeoutMs; _decodeErrorPolicy = decodeErrorPolicy; _onFrameError = onFrameError; } public async Task RunAsync(CancellationToken ct) { + // 未完成帧状态：上次 AdvanceTo 时缓冲里留存的半帧字节（快照在挂表前拷贝—— + // AdvanceTo 之后旧 sequence 不再保证有效）。缓冲为空 = 没有进行中的帧，不计时。 + var pendingBytes = 0L; + var pendingSnapshot = ReadOnlyMemory.Empty; + try { while (true) { - var result = await _reader.ReadAsync(CancellationToken.None).ConfigureAwait(false); + ReadResult result; + if (pendingBytes > 0 && _incompleteFrameTimeoutMs > 0) + { + // 计时窗口 = 本次 ReadAsync：新字节到达即返回、循环重来（计时自然重置）； + // 令牌刻意不链会话取消（生命周期约定二），只有超时会触发它。 + using var timeoutCts = new CancellationTokenSource(_incompleteFrameTimeoutMs); + try + { + result = await _reader.ReadAsync(timeoutCts.Token).ConfigureAwait(false); + } + catch (OperationCanceledException) when (!ct.IsCancellationRequested) + { + throw TimeoutFault(pendingBytes, pendingSnapshot); + } + } + else + { + result = await _reader.ReadAsync(CancellationToken.None).ConfigureAwait(false); + } + var buffer = result.Buffer; // 切尽当前缓冲内的所有完整帧；已缓冲的字节即便会话正在停止也要投递完 @@ -86,12 +115,18 @@ public async Task RunAsync(CancellationToken ct) if (buffer.Length > _maxIncompleteFrameBytes) { // 未完成帧永远等不齐或对端蓄意喂半帧：判定流不可恢复，交由连接层断线。 - var snapshot = buffer.Slice(0, Math.Min(buffer.Length, OverflowSnapshotBytes)).ToArray(); + var snapshot = buffer.Slice(0, Math.Min(buffer.Length, ErrorSnapshotBytes)).ToArray(); RaiseFrameError(new FrameErrorEventArgs(FrameErrorKind.IncompleteFrameOverflow, snapshot)); throw new SessionFaultException( $"未完成帧缓冲 {buffer.Length} 字节超过上限 {_maxIncompleteFrameBytes} 字节。"); } + // 挂表前拷贝半帧快照；缓冲为空（帧刚切尽/尚未开始）时清零，下轮不计时 + pendingBytes = buffer.Length; + pendingSnapshot = buffer.IsEmpty + ? ReadOnlyMemory.Empty + : buffer.Slice(0, (int)Math.Min(buffer.Length, ErrorSnapshotBytes)).ToArray(); + _reader.AdvanceTo(buffer.Start, buffer.End); if (result.IsCompleted) @@ -102,6 +137,11 @@ public async Task RunAsync(CancellationToken ct) // CancelPendingRead（会话停止时唤醒）：缓冲已投递完，正常退出 if (ct.IsCancellationRequested) break; + + // 非会话取消的 IsCanceled 只能来自未完成帧超时令牌的闩锁路径 + // （数据与取消同时到达时 ReadAsync 不抛异常、以 IsCanceled 结果返回） + if (pendingBytes > 0 && _incompleteFrameTimeoutMs > 0) + throw TimeoutFault(pendingBytes, pendingSnapshot); } } } @@ -112,6 +152,14 @@ public async Task RunAsync(CancellationToken ct) } } + /// 未完成帧超时的会话故障：先上报带快照的 FrameError，再抛出（由连接层断线重连）。 + private SessionFaultException TimeoutFault(long pendingBytes, ReadOnlyMemory snapshot) + { + RaiseFrameError(new FrameErrorEventArgs(FrameErrorKind.IncompleteFrameTimeout, snapshot)); + return new SessionFaultException( + $"未完成帧已缓冲 {pendingBytes} 字节，{_incompleteFrameTimeoutMs}ms 内未收齐后续字节，判定会话失效。"); + } + /// 切一帧；定界器支持丢弃上报时，把被丢弃的字节作为诊断事件上抛。 private bool TryDecodeFrame(ref ReadOnlySequence buffer, out ReadOnlySequence payload) { diff --git a/src/StreamFrame/Connection/StreamConnection.cs b/src/StreamFrame/Connection/StreamConnection.cs index 7b23115..f9b854a 100644 --- a/src/StreamFrame/Connection/StreamConnection.cs +++ b/src/StreamFrame/Connection/StreamConnection.cs @@ -29,7 +29,7 @@ public sealed class StreamConnection : IStreamConnection /// public event EventHandler? ConnectionChanged; - /// 帧层诊断事件：解码失败、被定界器丢弃的字节、不完整帧超限。字节已拷贝、可留存。 + /// 帧层诊断事件：解码失败、被定界器丢弃的字节、不完整帧超限、未完成帧超时。字节已拷贝、可留存。 public event EventHandler? FrameError; /// @@ -465,7 +465,8 @@ private void StartSession() var decoder = new FrameDecoder( pipe.Reader, _framing, _codec, _messageRelay, - maxIncompleteFrameBytes, _options.DecodeErrorPolicy, RaiseFrameError); + maxIncompleteFrameBytes, _options.IncompleteFrameTimeoutMs, + _options.DecodeErrorPolicy, RaiseFrameError); _receiveTask = WatchSessionFaultsAsync(ReceiveLoopAsync(pipe.Writer, token, epoch), cts, epoch); _decodeTask = WatchSessionFaultsAsync(decoder.RunAsync(token), cts, epoch); @@ -475,7 +476,8 @@ private void StartSession() /// /// 会话任务守护：会话停止引发的取消视为正常退出；其余异常（socket 故障、帧解码失败、 - /// 不完整帧超限、发送失败）统一转为断线重连（按 防止过期故障误杀新会话）。 + /// 不完整帧超限、未完成帧超时、发送失败）统一转为断线重连（按 + /// 防止过期故障误杀新会话）。 /// private async Task WatchSessionFaultsAsync(Task sessionTask, CancellationTokenSource sessionCts, int epoch) { diff --git a/src/StreamFrame/FrameErrorKind.cs b/src/StreamFrame/FrameErrorKind.cs index 4a412e6..c192541 100644 --- a/src/StreamFrame/FrameErrorKind.cs +++ b/src/StreamFrame/FrameErrorKind.cs @@ -11,4 +11,7 @@ public enum FrameErrorKind /// 未完成帧的已缓冲字节超过上限，判定流不可恢复。 IncompleteFrameOverflow, + + /// 未完成帧超时：帧已开头、缓冲里留着半帧字节，但在 IncompleteFrameTimeoutMs 内未收齐后续字节。 + IncompleteFrameTimeout, } diff --git a/src/StreamFrame/StreamConnectionOptions.cs b/src/StreamFrame/StreamConnectionOptions.cs index 7885645..664d1b2 100644 --- a/src/StreamFrame/StreamConnectionOptions.cs +++ b/src/StreamFrame/StreamConnectionOptions.cs @@ -83,6 +83,17 @@ public sealed class StreamConnectionOptions /// public int ReceiveIdleTimeoutMs { get; set; } + /// + /// 未完成帧超时（毫秒）：缓冲里已有半帧字节、但后续字节连续这么久未到达，即判定会话失效， + /// 断线重连（并通过 FrameError 事件上报 ， + /// 携带受上限保护的缓冲快照）。默认 0 = 关闭。 + /// + /// 与 的区别：后者在"完全没有字节"时也计时（要求连接 + /// 必须有周期流量）；本选项只在"帧已开头、迟迟收不齐"时计时——缓冲为空的正常静默不会触发， + /// 一帧完整切出后重新归零。适合允许长时间空闲、但半帧卡死必须判死的长度前缀协议（如 HSMS T8）。 + /// + public int IncompleteFrameTimeoutMs { get; set; } + /// /// 校验所有参数取值的合法性（由 构造时调用）。 /// 非法值在构造时立即失败，而不是在运行中的重连/收发路径深处抛出难定位的异常。 @@ -108,6 +119,8 @@ internal void Validate() throw new ArgumentOutOfRangeException(nameof(ReceiveQueueCapacity), ReceiveQueueCapacity, "不能为负数（0 = 不限制）。"); if (ReceiveIdleTimeoutMs < 0) throw new ArgumentOutOfRangeException(nameof(ReceiveIdleTimeoutMs), ReceiveIdleTimeoutMs, "不能为负数（0 = 关闭）。"); + if (IncompleteFrameTimeoutMs < 0) + throw new ArgumentOutOfRangeException(nameof(IncompleteFrameTimeoutMs), IncompleteFrameTimeoutMs, "不能为负数（0 = 关闭）。"); if (TcpKeepAlive && KeepAliveTimeMs <= 0) throw new ArgumentOutOfRangeException(nameof(KeepAliveTimeMs), KeepAliveTimeMs, "TcpKeepAlive 开启时必须为正数。"); if (TcpKeepAlive && KeepAliveIntervalMs <= 0) diff --git a/test/StreamFrame.Tests/FrameDecoderTests.cs b/test/StreamFrame.Tests/FrameDecoderTests.cs index 6293d7e..c78e17a 100644 --- a/test/StreamFrame.Tests/FrameDecoderTests.cs +++ b/test/StreamFrame.Tests/FrameDecoderTests.cs @@ -37,7 +37,8 @@ private static DecoderHarness CreateDecoder( IFramer framing, DecodeErrorPolicy policy = DecodeErrorPolicy.Disconnect, int? maxIncompleteFrameBytes = null, - ICodec? codec = null) + ICodec? codec = null, + int incompleteFrameTimeoutMs = 0) { var errors = new List(); var relay = Channel.CreateUnbounded(new UnboundedChannelOptions @@ -49,6 +50,7 @@ private static DecoderHarness CreateDecoder( var decoder = new FrameDecoder( pipe.Reader, framing, codec ?? StringCodec.Instance, relay, maxIncompleteFrameBytes ?? framing.MaxPayloadBytes + 4096, + incompleteFrameTimeoutMs, policy, args => { lock (errors) errors.Add(args); }); return new DecoderHarness(decoder, relay, pipe, errors, new List(), new CancellationTokenSource()); diff --git a/test/StreamFrame.Tests/IncompleteFrameTimeoutTests.cs b/test/StreamFrame.Tests/IncompleteFrameTimeoutTests.cs new file mode 100644 index 0000000..2500609 --- /dev/null +++ b/test/StreamFrame.Tests/IncompleteFrameTimeoutTests.cs @@ -0,0 +1,377 @@ +using System.Buffers.Binary; +using System.Net; +using System.Net.Sockets; +using System.Text; +using StreamFrame; + +namespace StreamFrame.Tests; + +/// +/// 未完成帧超时（IncompleteFrameTimeoutMs）的连接级端到端测试（真实 TCP 回环）。 +/// 语义：仅在缓冲里已有半帧字节时计时；缓冲为空的静默连接不触发；新字节到达即重置； +/// 超时判定会话失效并断线重连，FrameError 上报 IncompleteFrameTimeout + 受上限保护的快照。 +/// +public class IncompleteFrameTimeoutTests +{ + private static int GetFreePort() + { + var listener = new TcpListener(IPAddress.Loopback, 0); + listener.Start(); + var port = ((IPEndPoint)listener.LocalEndpoint).Port; + listener.Stop(); + return port; + } + + private static byte[] Frame(string payload) + { + var bytes = Encoding.UTF8.GetBytes(payload); + var frame = new byte[4 + bytes.Length]; + BinaryPrimitives.WriteInt32BigEndian(frame.AsSpan(0, 4), bytes.Length); + bytes.CopyTo(frame.AsSpan(4)); + return frame; + } + + private static async Task WaitForStateAsync( + StreamConnection connection, + Func predicate, + int timeoutMs = 5000) + { + var deadline = TestClock.TickCount64 + timeoutMs; + while (TestClock.TickCount64 < deadline) + { + if (predicate(connection.State)) + return; + await Task.Delay(20); + } + + Assert.True(predicate(connection.State), $"等待连接状态超时（{timeoutMs}ms），当前 {connection.State}。"); + } + + private static async Task WaitForAsync(Func condition, int timeoutMs = 5000, string? what = null) + { + var deadline = TestClock.TickCount64 + timeoutMs; + while (TestClock.TickCount64 < deadline) + { + if (condition()) + return; + await Task.Delay(20); + } + + Assert.True(condition(), $"等待条件超时（{timeoutMs}ms）：{what ?? "未描述"}。"); + } + + private static StreamConnection CreateServer(int port, StreamConnectionOptions options) + => new( + new LengthPrefixFramer(), + StringCodec.Instance, + IPAddress.Loopback, + port, + isActive: false, + options); + + /// + /// 连接服务端（被动模式重连窗口内监听可能尚未就绪，短暂 SocketException 需重试）。 + /// + private static async Task ConnectWithRetryAsync(TcpClient client, int port) + { + for (var attempt = 0; ; attempt++) + { + try + { + await client.ConnectAsync(IPAddress.Loopback, port); + return; + } + catch (SocketException) when (attempt < 10) + { + await Task.Delay(200); + } + } + } + + /// 记录 FrameError 事件（拷贝字节，回调返回后可安全断言）。 + private static List<(FrameErrorKind Kind, byte[] Bytes)> AttachErrorRecorder(StreamConnection server) + { + var errors = new List<(FrameErrorKind, byte[])>(); + server.FrameError += (_, e) => + { + lock (errors) + errors.Add((e.Kind, e.Bytes.ToArray())); + }; + return errors; + } + + /// 后台排空消息通道（与 Resilience 测试同款模式），返回已收消息列表。 + private static List StartMessageDrain(StreamConnection server) + { + var received = new List(); + var drainCts = new CancellationTokenSource(TimeSpan.FromSeconds(15)); + _ = Task.Run(async () => + { + try + { + await foreach (var message in server.GetMessages(drainCts.Token)) + lock (received) + received.Add(message); + } + catch (OperationCanceledException) + { + } + }); + return received; + } + + [Fact] + public void NegativeTimeout_ThrowsAtConstruction() + { + var port = GetFreePort(); + Assert.Throws(() => + CreateServer(port, new StreamConnectionOptions { IncompleteFrameTimeoutMs = -1 })); + } + + [Fact] + public async Task IdleConnection_NeverReceivedBytes_DoesNotTrigger() + { + var port = GetFreePort(); + await using var server = CreateServer(port, new StreamConnectionOptions { IncompleteFrameTimeoutMs = 200 }); + var errors = AttachErrorRecorder(server); + server.Start(CancellationToken.None); + + using (var client = new TcpClient()) + { + await client.ConnectAsync(IPAddress.Loopback, port); + await WaitForStateAsync(server, s => s == ConnectionState.Connected); + + // 连接后一个字节都不发：即使超过未完成帧超时时长，也不能触发（没有进行中的帧） + await Task.Delay(600); + Assert.Equal(ConnectionState.Connected, server.State); + } + + Assert.Empty(errors); + } + + [Fact] + public async Task CompleteFrameThenLongIdle_DoesNotTrigger() + { + var port = GetFreePort(); + await using var server = CreateServer(port, new StreamConnectionOptions { IncompleteFrameTimeoutMs = 200 }); + var errors = AttachErrorRecorder(server); + var received = StartMessageDrain(server); + server.Start(CancellationToken.None); + + using (var client = new TcpClient()) + { + await client.ConnectAsync(IPAddress.Loopback, port); + await WaitForStateAsync(server, s => s == ConnectionState.Connected); + + // 一帧完整送达后长时间静默：帧已切尽、缓冲为空，不应计时 + await client.GetStream().WriteAsync(Frame("done")); + await WaitForAsync(() => { lock (received) return received.Count == 1; }); + await Task.Delay(600); + + Assert.Equal(ConnectionState.Connected, server.State); + } + + Assert.Empty(errors); + lock (received) + Assert.Equal(new[] { "done" }, received); + } + + [Fact] + public async Task PartialLengthHeader_TimesOut_EndsSessionAndReconnects() + { + var port = GetFreePort(); + await using var server = CreateServer(port, new StreamConnectionOptions { IncompleteFrameTimeoutMs = 200 }); + var errors = AttachErrorRecorder(server); + var received = StartMessageDrain(server); + server.Start(CancellationToken.None); + + using (var client1 = new TcpClient()) + { + await client1.ConnectAsync(IPAddress.Loopback, port); + await WaitForStateAsync(server, s => s == ConnectionState.Connected); + + // 只发长度头的前 2 字节（完整头是 4 字节大端），半帧永远等不齐 + await client1.GetStream().WriteAsync(new byte[] { 0x00, 0x00 }); + + await WaitForAsync(() => { lock (errors) return errors.Count == 1; }, timeoutMs: 3000); + lock (errors) + { + var (kind, bytes) = Assert.Single(errors); + Assert.Equal(FrameErrorKind.IncompleteFrameTimeout, kind); + Assert.Equal(new byte[] { 0x00, 0x00 }, bytes); + } + + // 超时判定会话失效：离开 Connected 进入重连 + await WaitForStateAsync(server, s => s != ConnectionState.Connected); + } + + // 重连后新会话不受迟到故障影响：新客户端接入驱动回到 Connected，正常收发 + using (var client2 = new TcpClient()) + { + await ConnectWithRetryAsync(client2, port); + await WaitForStateAsync(server, s => s == ConnectionState.Connected, timeoutMs: 8000); + await Task.Delay(300); + await client2.GetStream().WriteAsync(Frame("after-timeout")); + } + + await WaitForAsync(() => { lock (received) return received.Count == 1; }, timeoutMs: 8000); + lock (received) + Assert.Equal(new[] { "after-timeout" }, received); + } + + [Fact] + public async Task CompleteHeaderPartialBody_TimesOut() + { + var port = GetFreePort(); + await using var server = CreateServer(port, new StreamConnectionOptions { IncompleteFrameTimeoutMs = 200 }); + var errors = AttachErrorRecorder(server); + server.Start(CancellationToken.None); + + using (var client = new TcpClient()) + { + await client.ConnectAsync(IPAddress.Loopback, port); + await WaitForStateAsync(server, s => s == ConnectionState.Connected); + + // 长度头完整（声明 10 字节正文）但正文只到 3 字节 + var partial = new byte[] { 0x00, 0x00, 0x00, 0x0A, 0x41, 0x42, 0x43 }; + await client.GetStream().WriteAsync(partial); + + await WaitForAsync(() => { lock (errors) return errors.Count == 1; }, timeoutMs: 3000); + lock (errors) + { + var (kind, bytes) = Assert.Single(errors); + Assert.Equal(FrameErrorKind.IncompleteFrameTimeout, kind); + Assert.Equal(partial, bytes); + } + } + } + + [Fact] + public async Task TrickleWithinDeadline_TimerResets_FrameDelivered() + { + var port = GetFreePort(); + await using var server = CreateServer(port, new StreamConnectionOptions { IncompleteFrameTimeoutMs = 1000 }); + var errors = AttachErrorRecorder(server); + var received = StartMessageDrain(server); + server.Start(CancellationToken.None); + + using (var client = new TcpClient()) + { + await client.ConnectAsync(IPAddress.Loopback, port); + await WaitForStateAsync(server, s => s == ConnectionState.Connected); + var stream = client.GetStream(); + + // 每段间隔 250ms（< 1000ms 超时）：每次续传都应重置计时，最终整帧交付 + var header = new byte[4]; + BinaryPrimitives.WriteInt32BigEndian(header, 8); + await stream.WriteAsync(header); + await Task.Delay(250); + await stream.WriteAsync(Encoding.UTF8.GetBytes("123")); + await Task.Delay(250); + await stream.WriteAsync(Encoding.UTF8.GetBytes("456")); + await Task.Delay(250); + await stream.WriteAsync(Encoding.UTF8.GetBytes("78")); + + await WaitForAsync(() => { lock (received) return received.Count == 1; }); + await Task.Delay(400); // 交付后再静默一会儿，确认没有迟到的超时 + + Assert.Equal(ConnectionState.Connected, server.State); + } + + Assert.Empty(errors); + lock (received) + Assert.Equal(new[] { "12345678" }, received); + } + + [Fact] + public async Task RepeatedTimeoutCycles_RemainIdempotent() + { + var port = GetFreePort(); + await using var server = CreateServer(port, new StreamConnectionOptions { IncompleteFrameTimeoutMs = 200 }); + var errors = AttachErrorRecorder(server); + var received = StartMessageDrain(server); + server.Start(CancellationToken.None); + + // 连续两轮"半帧超时 → 重连"，验证 epoch 防护下迟到故障不会误杀后续会话。 + // 客户端必须等超时事件落地后再释放：提前 dispose 会让 FIN 抢先触发重连（走对端关闭路径）； + // 被动模式重连由下一个客户端接入驱动完成。 + for (var cycle = 0; cycle < 2; cycle++) + { + using var client = new TcpClient(); + await ConnectWithRetryAsync(client, port); + await WaitForStateAsync(server, s => s == ConnectionState.Connected); + await client.GetStream().WriteAsync(new byte[] { 0x00, 0x0F }); + + var expected = cycle + 1; + await WaitForAsync( + () => { lock (errors) return errors.Count == expected; }, + timeoutMs: 3000, what: $"第 {cycle + 1} 轮未完成帧超时事件"); + await WaitForStateAsync(server, s => s != ConnectionState.Connected); + } + + lock (errors) + Assert.Equal(2, errors.Count); + + // 第三轮：正常客户端照常工作 + using (var client = new TcpClient()) + { + await ConnectWithRetryAsync(client, port); + await WaitForStateAsync(server, s => s == ConnectionState.Connected, timeoutMs: 8000); + await Task.Delay(300); + await client.GetStream().WriteAsync(Frame("third-session")); + } + await WaitForAsync(() => { lock (received) return received.Count == 1; }, timeoutMs: 8000); + lock (received) + Assert.Equal(new[] { "third-session" }, received); + } + + [Fact] + public async Task SnapshotIsCappedAtLimit() + { + var port = GetFreePort(); + await using var server = CreateServer(port, new StreamConnectionOptions { IncompleteFrameTimeoutMs = 200 }); + var errors = AttachErrorRecorder(server); + server.Start(CancellationToken.None); + + using (var client = new TcpClient()) + { + await client.ConnectAsync(IPAddress.Loopback, port); + await WaitForStateAsync(server, s => s == ConnectionState.Connected); + + // 声明 20000 字节正文但只发 9000 字节：快照必须被截到上限（8192），而不是全量拷贝 + var partial = new byte[4 + 9000]; + BinaryPrimitives.WriteInt32BigEndian(partial, 20000); + await client.GetStream().WriteAsync(partial); + + await WaitForAsync(() => { lock (errors) return errors.Count == 1; }, timeoutMs: 3000); + lock (errors) + { + var (kind, bytes) = Assert.Single(errors); + Assert.Equal(FrameErrorKind.IncompleteFrameTimeout, kind); + Assert.Equal(8192, bytes.Length); + } + } + } + + [Fact] + public async Task DisposeDuringTimeoutReconnect_CompletesCleanly() + { + var port = GetFreePort(); + var server = CreateServer(port, new StreamConnectionOptions { IncompleteFrameTimeoutMs = 200 }); + _ = AttachErrorRecorder(server); + server.Start(CancellationToken.None); + + using (var client = new TcpClient()) + { + await client.ConnectAsync(IPAddress.Loopback, port); + await WaitForStateAsync(server, s => s == ConnectionState.Connected); + await client.GetStream().WriteAsync(new byte[] { 0x00, 0x2A }); + + // 超时触发、重连流程进行中即 Dispose：必须干净停机、不抛异常 + await WaitForStateAsync(server, s => s != ConnectionState.Connected); + } + + await server.DisposeAsync(); + Assert.True(server.IsDisposed); + } +}