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
3 changes: 3 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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 横幅本身不变)。

Expand Down
12 changes: 10 additions & 2 deletions README.en.md
Original file line number Diff line number Diff line change
Expand Up @@ -95,7 +95,7 @@ public interface ICodec<TMessage>
- **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
Expand All @@ -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}");
Expand All @@ -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)

Expand Down Expand Up @@ -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)):
Expand Down
12 changes: 10 additions & 2 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -95,7 +95,7 @@ public interface ICodec<TMessage>
- **自动重连**:`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 背压自然传导到对端,防内存无限增长
Expand All @@ -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}");
Expand All @@ -131,7 +132,12 @@ socket 层全量输出(含被丢弃的噪声字节),发送侧按实际写

### 未完成帧防护

对端声明一个超长帧却永远不补齐(或 STX/ETX 流中只有 STX 没有闭合),会无限占用缓冲。`MaxIncompleteFrameBufferBytes`(默认 = 帧上限 + 4KB)给"等不齐的半帧"设了硬上限,超限即断线并通过 `FrameError` 上报。
对端声明一个超长帧却永远不补齐(或 STX/ETX 流中只有 STX 没有闭合),会无限占用缓冲。两条互补的防线:

- `MaxIncompleteFrameBufferBytes`(默认 = 帧上限 + 4KB)——**字节**上限:半帧超过即断线,防内存攻击;
- `IncompleteFrameTimeoutMs`(默认 0 = 关闭)——**时间**上限:半帧开始后连续这么久收不到后续字节即断线,`FrameError` 上报 `IncompleteFrameTimeout` 并携带受 8KB 上限保护的缓冲快照。

未完成帧超时只计"帧已开头、迟迟收不齐"的时间:缓冲为空的静默连接不计时(收到字节即重置,整帧切尽后归零)。它与 `ReceiveIdleTimeoutMs`(见下节)互补——后者在完全没流量时也计时,适合有周期报文的协议;对允许长时间空闲、但半帧卡死必须判死的协议(如 HSMS T8),用未完成帧超时。

### 日志(可选)

Expand Down Expand Up @@ -163,6 +169,8 @@ _ = Task.Run(async () =>

对端"猝死"(断电/拔线,无 FIN/RST)时静默超限 → 会话判定死亡 → 自动重连;对端不可达时重连持续失败,状态停留在 `Connecting`/`Retry`。

两种接收超时的取舍:`ReceiveIdleTimeoutMs` 要求连接**必须有周期流量**——协议有心跳/周期上报时用它(还能兜底半开连接);协议允许长时间静默(只在有帧进行中才该有流量)时改用 `IncompleteFrameTimeoutMs`,静默不算故障、半帧卡死才判死。

## 性能

BenchmarkDotNet 实测(详见 [bench/README.md](bench/README.md),可本地复现):
Expand Down
60 changes: 54 additions & 6 deletions src/StreamFrame/Connection/FrameDecoder.cs
Original file line number Diff line number Diff line change
Expand Up @@ -9,23 +9,26 @@ namespace StreamFrame;
/// 半包靠 AdvanceTo 保留未消费字节;粘包靠循环切尽所有完整帧。
///
/// 生命周期约定一:本类型不负责消息通道(relay)的完成——通道归连接所有,
/// 跨会话复用,仅在连接 Dispose 时完成。会话级失败(解码失败、不完整帧超限)以
/// <see cref="SessionFaultException"/> 上抛,由连接层决定断线重连。
/// 跨会话复用,仅在连接 Dispose 时完成。会话级失败(解码失败、不完整帧超限、
/// 未完成帧超时)以 <see cref="SessionFaultException"/> 上抛,由连接层决定断线重连。
///
/// 生命周期约定二:ReadAsync 不绑定会话取消令牌(取消中的 ReadAsync 会把 Pipe 留在
/// "读取进行中"状态,无法再 TryRead)。退出由流结束(writer.CompleteAsync)或
/// CancelPendingRead 驱动;退出前把已缓冲的完整帧全部投递,保证"收到的字节必达"。
/// 唯一例外:未完成帧超时使用独立的超时令牌(不链会话取消),超时后本循环直接以
/// 会话故障收尾、不再复用该 PipeReader——因此不会有"进行中的读取残留到下一次读取"。
/// </summary>
internal sealed class FrameDecoder<TMessage>
{
/// <summary>上报 IncompleteFrameOverflow 时携带的缓冲快照上限,避免事件本身复制超长数据。</summary>
private const int OverflowSnapshotBytes = 8192;
/// <summary>上报 IncompleteFrameOverflow / IncompleteFrameTimeout 时携带的缓冲快照上限,避免事件本身复制超长数据。</summary>
private const int ErrorSnapshotBytes = 8192;

private readonly PipeReader _reader;
private readonly IFramer _framing;
private readonly ICodec<TMessage> _codec;
private readonly Channel<TMessage> _relay;
private readonly int _maxIncompleteFrameBytes;
private readonly int _incompleteFrameTimeoutMs;
private readonly DecodeErrorPolicy _decodeErrorPolicy;
private readonly Action<FrameErrorEventArgs>? _onFrameError;

Expand All @@ -35,6 +38,7 @@ public FrameDecoder(
ICodec<TMessage> codec,
Channel<TMessage> relay,
int maxIncompleteFrameBytes,
int incompleteFrameTimeoutMs,
DecodeErrorPolicy decodeErrorPolicy,
Action<FrameErrorEventArgs>? onFrameError)
{
Expand All @@ -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<byte>.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;

// 切尽当前缓冲内的所有完整帧;已缓冲的字节即便会话正在停止也要投递完
Expand Down Expand Up @@ -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<byte>.Empty
: buffer.Slice(0, (int)Math.Min(buffer.Length, ErrorSnapshotBytes)).ToArray();

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

if (result.IsCompleted)
Expand All @@ -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);
}
}
}
Expand All @@ -112,6 +152,14 @@ public async Task RunAsync(CancellationToken ct)
}
}

/// <summary>未完成帧超时的会话故障:先上报带快照的 FrameError,再抛出(由连接层断线重连)。</summary>
private SessionFaultException TimeoutFault(long pendingBytes, ReadOnlyMemory<byte> snapshot)
{
RaiseFrameError(new FrameErrorEventArgs(FrameErrorKind.IncompleteFrameTimeout, snapshot));
return new SessionFaultException(
$"未完成帧已缓冲 {pendingBytes} 字节,{_incompleteFrameTimeoutMs}ms 内未收齐后续字节,判定会话失效。");
}

/// <summary>切一帧;定界器支持丢弃上报时,把被丢弃的字节作为诊断事件上抛。</summary>
private bool TryDecodeFrame(ref ReadOnlySequence<byte> buffer, out ReadOnlySequence<byte> payload)
{
Expand Down
8 changes: 5 additions & 3 deletions src/StreamFrame/Connection/StreamConnection.cs
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@ public sealed class StreamConnection<TMessage> : IStreamConnection<TMessage>
/// <inheritdoc />
public event EventHandler<ConnectionState>? ConnectionChanged;

/// <summary>帧层诊断事件:解码失败、被定界器丢弃的字节、不完整帧超限。字节已拷贝、可留存。</summary>
/// <summary>帧层诊断事件:解码失败、被定界器丢弃的字节、不完整帧超限、未完成帧超时。字节已拷贝、可留存。</summary>
public event EventHandler<FrameErrorEventArgs>? FrameError;

/// <inheritdoc />
Expand Down Expand Up @@ -465,7 +465,8 @@ private void StartSession()

var decoder = new FrameDecoder<TMessage>(
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);
Expand All @@ -475,7 +476,8 @@ private void StartSession()

/// <summary>
/// 会话任务守护:会话停止引发的取消视为正常退出;其余异常(socket 故障、帧解码失败、
/// 不完整帧超限、发送失败)统一转为断线重连(按 <paramref name="epoch"/> 防止过期故障误杀新会话)。
/// 不完整帧超限、未完成帧超时、发送失败)统一转为断线重连(按 <paramref name="epoch"/>
/// 防止过期故障误杀新会话)。
/// </summary>
private async Task WatchSessionFaultsAsync(Task sessionTask, CancellationTokenSource sessionCts, int epoch)
{
Expand Down
3 changes: 3 additions & 0 deletions src/StreamFrame/FrameErrorKind.cs
Original file line number Diff line number Diff line change
Expand Up @@ -11,4 +11,7 @@ public enum FrameErrorKind

/// <summary>未完成帧的已缓冲字节超过上限,判定流不可恢复。</summary>
IncompleteFrameOverflow,

/// <summary>未完成帧超时:帧已开头、缓冲里留着半帧字节,但在 IncompleteFrameTimeoutMs 内未收齐后续字节。</summary>
IncompleteFrameTimeout,
}
Loading
Loading