diff --git a/CHANGELOG.md b/CHANGELOG.md
index a63ecd5..0f01d31 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,8 @@
## [Unreleased]
### 修复
+- **终态竞速守门与发布段原子化（[#68](https://github.com/CSJ608/StreamFrame/issues/68)）**：并发 Dispose 与 Reconnect/连接成功竞速时，已通过停机检查但尚未写入 State 的发布被抢占，停机写入 Disconnected 后迟到的 Retry/Connected 会把终态翻转（State 停在 Retry/Connected）；Retry 派发还会对停机回收的 `_lifetimeCts` 空引用。状态发布收敛为发布段（`_publishGate` 锁内原子：停机/过期纪元判定 + 编号纪元簿记 + State 写入）与通知段（锁外用户回调/等待器/派发，回调可安全重入）；停机后到达的非终态发布整体丢弃；传输建立与停机竞速时新 socket 在守门处当场释放（此前泄漏到终结器）。新增迟到发布确定性门控、连接成功与停机竞速、并发关闭/重连终态不变量回归。
+- **回调重入会话撕裂（[#68](https://github.com/CSJ608/StreamFrame/issues/68)）**：Connected 回调内同步 `Reconnect()` 时，迟到发布帧的 `StartSession` 以过期编号/纪元"兜底重建"僵尸会话，其拆除会关掉新会话刚建立的 socket（编号复用 + 会话失活，实测可复现）。改为纪元现行性门控：过期帧放弃创建，连接由取代者循环完成；`_socket` 被新连接覆盖时旧值当场释放（直连迁移路径不再泄漏）。新增回调重入（Reconnect/DisposeAsync）收敛性、合法状态转移序列与编号线性化回归。
- **基准工作量对齐（[#70](https://github.com/CSJ608/StreamFrame/issues/70)）**：单向不再回显；直接 TCP 与框架共享负载、定界、Codec、物化校验及完成点，覆盖双 Framer、64B/1KB/64KB 和字节/两种字符串编码，往返逐条等待回显。增加全矩阵连续批次验证；原始结果与环境随仓库保存，撤回不同工作量相减所得的框架税结论，修正指标单位和筛选命令，不修改框架实现。
- **NativeAOT 运行验收（[#71](https://github.com/CSJ608/StreamFrame/issues/71)）**：CI 发布后执行 linux-x64 原生产物；有界等待连接与三条消息，校验会话绑定/普通发送及内容，观察消费任务并释放资源。动态选择端口并仅在连接阶段有限重选；连接失败、少收、消费异常和内容错误以非零退出，通过受控故障反证检查有效。保留警告为错误和非必需 AOT 检查。
- **Soak 队列续发与收尾（[#67](https://github.com/CSJ608/StreamFrame/issues/67)）**：发送 worker 在每次出队前检查会话取消，避免上一帧写完后拆除的旧 worker 取走并丢弃待续发普通条目；新增确定性门控回归。长流测试精确排空后关闭读 socket 并观察任务，修复 net48 仅取消令牌时读任务不收敛。
@@ -16,6 +18,7 @@
- **Soak 目标矩阵（[#66](https://github.com/CSJ608/StreamFrame/issues/66)）**：显式选择 Ubuntu net8.0/net10.0、Windows net8.0/net10.0/net48，避免 Ubuntu 启动 .NET Framework 测试；job 名、详细日志与始终尝试上传的 TRX 产物按 OS/TFM 区分，测试断言失败继续使 job 失败。普通 CI 必需检查名不变。
### 变更
+- **状态机与所有权规范（[#68](https://github.com/CSJ608/StreamFrame/issues/68) PR-1）**：DESIGN 增补权威清单——状态转换全表与发布段/通知段规则、资源所有权表（会话 socket/监听器/会话五元组/消息通道）、锁与同步原语清单、允许重入点（回调内 Reconnect/Dispose、停机后调用、迟到旧会话任务）及已知限制（拆除同步等待持锁、`_socket` 字段语义——指向后续内部 Session 对象重构）；与新增的 `StateOwnershipBaselineTests` 共同构成 #68 会话所有权重构的行为基线。
- 依赖升级（Dependabot，[PR #59](https://github.com/CSJ608/StreamFrame/pull/59)）：soak 工作流 `actions/setup-dotnet` v4→v6（Node 24 运行时，与其余工作流对齐）。
- **驱动所有权与并发文档（[#69](https://github.com/CSJ608/StreamFrame/issues/69)）**：补全 ICodec/IFramer/流式定界的缓冲借用及同连接收发、新旧会话、跨连接共享约束；双语 README 增加拥有独立数据的 byte[]/ReadOnlyMemory 示例、有界队列生产配置及单客户端边界，澄清终态等待、KeepAlive 粒度和入队非确认。示例纳入三目标编译并以主动覆写缓冲验证异步消费；不改变默认值或实现公共契约。
- **Soak 可观测协议（[#67](https://github.com/CSJ608/StreamFrame/issues/67)）**：保持 SendAsync 仅入队契约，记录可重放 seed、动作、尝试 ID、会话及发送/读取阶段；业务测试使用 ACK、重试、去重并精确校验，重试前独立检查队列续发与会话边界。所有阶段有界等待，未启用长测报告 Skip；完整 600 秒五目标连续三次验收另行跟踪，见 docs/SOAK.md。
diff --git a/docs/DESIGN.md b/docs/DESIGN.md
index 449f6da..3a7daa1 100644
--- a/docs/DESIGN.md
+++ b/docs/DESIGN.md
@@ -142,6 +142,45 @@ socket --ReceiveAsync--> pipe.Writer --[Pipe]--> FrameDecoder.RunAsync
**原始字节与诊断（1.2.0）**：`RawBytesReceived/Sent` 全量输出（含被定界器丢弃的噪声字节），发送侧按实际写出分片回调；两者与 `ConnectionChanged`、`FrameError` 的用户回调异常均被隔离，不反噬会话。`IFrameDiscardReporting`（可选接口，内置两个 codec 实现）让定界器精确上报重同步丢弃的字节，经 `FrameError(DiscardedByResync)` 交上层调试。
+### 状态机与所有权规范（#68 行为基线）
+
+> 本节是 `StreamConnection` 状态转换、资源所有权、锁范围与允许重入点的权威清单，
+> 与 `StateOwnershipBaselineTests` 共同构成后续会话所有权重构的行为底线。
+
+**状态转换全表**（`CommunicationStateChanging` 是唯一发布入口；`State` 只在发布段写入）：
+
+| 目标状态 | 触发源 | 发布点 |
+|---|---|---|
+| `Connecting` | `Start`/重连的 `StartAsync` 循环头（每轮尝试前） | 连接循环 |
+| `Connected` | 传输建立成功（connect/accept 返回后、过停机守门） | 连接循环（唯一发布点） |
+| `Retry` | 用户 `Reconnect()`；会话任务故障经 `ScheduleReconnect`（携带 epoch，过期即丢） | 用户线程 / 线程池 |
+| `Disconnected` | `Shutdown`（Dispose 或生命周期令牌取消）唯一来源 | 停机路径 |
+
+发布段规则（`_publishGate` 锁内，原子）：① 停机后到达的非终态发布**整体丢弃**——终态必须是最后对外可见的状态（并发 Reconnect/连接成功与 Dispose 竞速时不被翻转）；② 过期纪元的故障发布在任何对外可见效果之前丢弃；③ 会话编号分配/归零与纪元递增先于 `State` 写入（线性化点：`ConnectionChanged` 回调与 `WaitForConnectedAsync` 完成时刻读取 `CurrentSessionId` 必得一致值）。通知段（锁外）：用户回调（异常隔离）、等待器完成、会话创建/拆除派发——回调可能重入，不得持有发布锁。
+
+**会话创建的纪元现行性门控**：`StartSession` 只在"发布帧的纪元仍是最新、编号非零"时创建会话。Connected 回调内同步 `Reconnect()` 或发布窗口内的现行纪元故障会让位于取代者（其循环重建连接），迟到帧不再"兜底重建"——那曾以过期编号复用、拆除新会话刚建立的 socket（撕裂会话）。被取代帧的 socket 在新连接覆盖 `_socket` 时当场释放。
+
+**资源所有权**：
+
+| 资源 | 创建者 | 释放者 | 备注 |
+|---|---|---|---|
+| `_socket`（会话 socket） | 连接循环（connect/accept） | 会话拆除（`StopSessionCore`）；新连接覆盖时；`Shutdown` 末尾兜底 | 赋值与关闭互斥于 `_sessionGate`；停机守门失败的连接当场释放 |
+| `_server`（监听 socket） | `InitServer`（Bind/Listen 全部成功才发布） | 单客户端接受后关闭；`Shutdown` 释放；bind 失败自清理 | 归当代接受循环所有（代次门控，#47） |
+| 会话五元组（`_sessionCts`/`_pipe`/接收、解码、发送三任务） | `StartSession` | `StopSessionCore`（cancel → 唤醒 → 置空字段 → 有界等待 ≤2s → 关 socket → 释放 cts） | 终态后不再创建（`StartSession` 停机守门） |
+| `_messageRelay`（消息通道） | 构造 | 仅 `Shutdown`（连接级、跨会话存活） | 会话重建不完成它（1.2.0 假活修复） |
+| `_sendQueue` / 挂起的会话绑定条目 | 业务侧入队 | `Shutdown` 完成通道；拆除时 `FaultPendingSessionSends` | 条目三态 CAS：Queued/Claimed/Cancelled/Expired |
+
+**锁与同步原语清单**：`_publishGate`（状态发布段，锁内无 await/回调）；`_sessionGate`（会话创建/拆除、`_socket` 赋值互斥；拆除等待最长 2s——已知同步等待，见"已知限制"）；`_listenerGate`（监听器发布与停机释放互斥）；`_acceptLock`（接受尝试串行化 + 代次门控落点）；`_sendLock`（socket 写串行化）；CAS 字段：`_retryInProgress`（重拆防重入）、`_acceptLoopId`（接受循环代次）、`_disposeStage`/`_started`、发送条目状态字。
+
+**允许的重入点**（用户回调可调用任意公共 API，异常一律隔离）：
+
+- `Connected` 回调内 `Reconnect()`：合法（病态但支持）。回调内 Retry 先行（纪元递增 + 派发新循环），原发布帧的会话创建被纪元门控放弃；新连接由新循环完成，编号恒非零且不复用。
+- `Connected` 回调内 `DisposeAsync()`：合法。停机同步完成（会话尚未创建则无拆除等待）；原发布帧随后的会话创建被停机守门放弃；挂起等待器以取消收尾。
+- 停机后的任何调用：`Reconnect()`/`SendInSessionAsync` 等抛 `ObjectDisposedException` 或以会话失效收尾；迟到的内部状态发布被发布段丢弃。
+- 迟到的旧会话任务/故障：入口过期纪元检查丢弃，不误杀新会话（1.2.0/2.3.1 既有保证，重入路径由纪元门控补齐）。
+
+**已知限制（后续 PR 方向）**：① `StopSessionCore` 持 `_sessionGate` 同步等待会话任务最长 2 秒——线程池饥饿时放大停机/重连延迟（高并发压力下可观测），#68 后续以"先中止 I/O、异步等待任务"改造；② `_socket`/会话五元组是"当前值"字段语义而非会话对象所有——并发双循环的直连迁移路径靠覆盖释放与拆除收敛，集中所有权（内部 Session 对象）是重构目标。
+
### netstandard2.0 兼容层（2.1.0）
多目标 `netstandard2.0;net8.0;net10.0`：
diff --git a/src/StreamFrame/Connection/StreamConnection.cs b/src/StreamFrame/Connection/StreamConnection.cs
index 0216d22..fa782f6 100644
--- a/src/StreamFrame/Connection/StreamConnection.cs
+++ b/src/StreamFrame/Connection/StreamConnection.cs
@@ -79,9 +79,11 @@ public string? RemoteIpAddress
#if NET9_0_OR_GREATER
private readonly Lock _listenerGate = new(); // 监听器发布与停机释放互斥，不跨 await 持有。
private readonly Lock _sessionGate = new(); // System.Threading.Lock（net9+，比 monitor 锁更轻量）
+ private readonly Lock _publishGate = new(); // 状态发布段互斥（见 CommunicationStateChanging）
#else
private readonly object _listenerGate = new();
private readonly object _sessionGate = new();
+ private readonly object _publishGate = new();
#endif
private Pipe? _pipe;
@@ -238,10 +240,29 @@ private async Task StartAsync(CancellationToken ct, int loopId)
if (!IsActive && Volatile.Read(ref _acceptLoopId) != loopId)
return; // 已被更新的接受循环取代：静默退出，不再竞争 accept
- _socket = IsActive ? await ConnectAsync(ct).ConfigureAwait(false) : await AcceptAsync(ct, loopId).ConfigureAwait(false);
- if (_socket is null)
+ var socket = IsActive ? await ConnectAsync(ct).ConfigureAwait(false) : await AcceptAsync(ct, loopId).ConfigureAwait(false);
+ if (socket is null)
return; // 排队等到锁时已被取代（AcceptAsync 内部的代次检查）
+ // 停机与"连接已建立"的竞速守门（#68）：Shutdown 可能在传输建立完成后、
+ // 发布前的窗口内完成——终态已广播、会话已拆除。此时发布 Connected 会翻转
+ // 终态并重建会话；把 socket 留进 _socket 则无人负责关闭（Shutdown 已走过
+ // 关闭点），泄漏到终结器。判定与赋值同锁于 Shutdown 末尾的关闭块：窗口闭合。
+ lock (_sessionGate)
+ {
+ if (ct.IsCancellationRequested || IsDisposed)
+ {
+ socket.Dispose();
+ return;
+ }
+
+ // 被取代的旧 socket（回调内 Reconnect 重入、并发双循环的直连迁移）：
+ // 新连接接管字段时旧值已无人认领（其会话从未建立或已被拆除），
+ // 当场释放不泄漏。正常路径此处恒为 null（会话拆除时已清理）。
+ _socket?.Dispose();
+ _socket = socket;
+ }
+
_retryScheduler.Reset();
CommunicationStateChanging(ConnectionState.Connected);
connected = true;
@@ -491,32 +512,47 @@ 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 对外发布（事件回调/等待器完成）之前完成分配；
- // 会话纪元（epoch）同样在发布之前递增——发布后到达的旧纪元幽灵故障在函数入口
- // 即被判为过期丢弃，不会卷进"已发布、任务未创建"的新生会话（评审附带发现收口）。
- if (newState == ConnectionState.Connected)
+ // 发布段（_publishGate 锁内）：迟到/过期判定 + 编号纪元簿记 + State 写入。
+ // 判定与写入必须原子：否则"已通过停机检查、尚未写入 State"的发布线程被抢占，
+ // 停机线程在此间隙写入 Disconnected 后，迟到的 Retry/Connected 会把终态翻转
+ // （并发 Reconnect 与 Dispose 竞速实测可命中）。锁内无 await、无用户回调。
+ lock (_publishGate)
{
- var id = checked(Interlocked.Increment(ref _sessionCounter));
- Volatile.Write(ref _currentSessionId, id);
- Interlocked.Increment(ref _sessionEpoch);
- }
- else if (newState is ConnectionState.Retry or ConnectionState.Disconnected)
- {
- Volatile.Write(ref _currentSessionId, 0);
- if (newState == ConnectionState.Retry)
- _metrics.Reconnect(); // 过期幽灵已在函数入口被丢弃，不会计入
- }
+ // 停机后到达的迟到发布（竞速中的连接成功、用户 Reconnect、会话故障重连）整体
+ // 丢弃：其后任何非终态发布都会把对外 State 翻成 Connected/Retry、重分配/归零
+ // 会话编号——终态必须是最后对外可见的状态。Shutdown 自身的 Disconnected 发布
+ // 不受影响（停机标记先于本调用置位，且停机是 Disconnected 的唯一来源）。
+ if (IsDisposed && newState != ConnectionState.Disconnected)
+ return;
+
+ // 迟到的过期故障（会话已被替换，epoch 不匹配）：在任何对外可见的发布之前丢弃。
+ // 若放到 Retry 分支的 gate 内才检查（旧实现），State=Retry 的发布与 CurrentSessionId
+ // 归零已经污染了完全存活的会话——WaitForConnectedAsync 悬挂、活会话编号被误判失效。
+ // gate 内的权威复查仍然保留，防检查与发布之间 epoch 变化的窄竞态。
+ if (newState == ConnectionState.Retry && observedEpoch is { } staleEpoch
+ && Volatile.Read(ref _sessionEpoch) != staleEpoch)
+ return;
- State = newState;
+ // 公共会话编号的线性化点：Connected 对外发布（事件回调/等待器完成）之前完成分配；
+ // 会话纪元（epoch）同样在发布之前递增——发布后到达的旧纪元幽灵故障在函数入口
+ // 即被判为过期丢弃，不会卷进"已发布、任务未创建"的新生会话（评审附带发现收口）。
+ if (newState == ConnectionState.Connected)
+ {
+ var id = checked(Interlocked.Increment(ref _sessionCounter));
+ Volatile.Write(ref _currentSessionId, id);
+ Interlocked.Increment(ref _sessionEpoch);
+ }
+ else if (newState is ConnectionState.Retry or ConnectionState.Disconnected)
+ {
+ Volatile.Write(ref _currentSessionId, 0);
+ if (newState == ConnectionState.Retry)
+ _metrics.Reconnect(); // 过期幽灵已在锁内被丢弃，不会计入
+ }
+ State = newState;
+ }
+
+ // 通知段（锁外）：用户回调可能重入（回调内 Reconnect/Dispose），不得持有发布锁。
try
{
ConnectionChanged?.Invoke(this, newState);
@@ -550,6 +586,11 @@ private void CommunicationStateChanging(ConnectionState newState, int? observedE
{
lock (_sessionGate)
{
+ // 取锁等待期间停机可能已完成：不拆除、不派发新循环（_lifetimeCts
+ // 已被 Shutdown 置空，继续执行会对 null 解引用）
+ if (IsDisposed)
+ return;
+
if (observedEpoch is { } epoch && Volatile.Read(ref _sessionEpoch) != epoch)
return; // 故障会话已被替换，避免误杀新会话
@@ -558,7 +599,18 @@ private void CommunicationStateChanging(ConnectionState newState, int? observedE
StopSessionCore();
}
- StartCore(_lifetimeCts!.Token);
+ // 取锁后停机仍可能已完成（_lifetimeCts 被 Shutdown 回收）：放弃派发，
+ // 静默让位终态；令牌源在取消与回收之间的窗口取 Token 可能抛 ODE，一并容忍。
+ var lifetime = _lifetimeCts;
+ if (lifetime is null)
+ return;
+ try
+ {
+ StartCore(lifetime.Token);
+ }
+ catch (ObjectDisposedException)
+ {
+ }
}
finally
{
@@ -573,17 +625,23 @@ private void StartSession(long sessionId, int epoch)
{
lock (_sessionGate)
{
- StopSessionCore();
+ // 停机守门（#68）：本方法唯一的迟到路径是"Connected 回调内同步 Dispose"——
+ // 发布已过守门、停机随后完成，外层才轮到创建会话。终态后不再创建会话任务
+ //（此时 _sendLock/_acceptLock 已释放，迟到的 worker 只会在已释放的锁上故障）。
+ if (IsDisposed)
+ return;
- // 病态重入兜底：Connected 回调里同步触发 Reconnect() 会让编号/纪元在 StartSession
- // 执行前被归零/递增——此时重新分配两者，保证会话任务拿到的编号恒非零、
- // 纪元为最新（旧纪元的任务故障会被入口的过期检查丢弃）。
- if (sessionId == 0)
- {
- sessionId = checked(Interlocked.Increment(ref _sessionCounter));
- Volatile.Write(ref _currentSessionId, sessionId);
- epoch = Interlocked.Increment(ref _sessionEpoch);
- }
+ // 纪元现行性守门（#68）：本发布帧的纪元已被更新的重连取代——典型是
+ // Connected 回调内同步 Reconnect()（回调里 Retry 已递增纪元并派发新循环），
+ // 或发布窗口内现行纪元的故障重连。迟到帧不得以过期编号/纪元重建会话：
+ // 编号会被复用、StopSessionCore 会拆掉新会话刚建立的 socket（旧"兜底重建"
+ // 曾在该路径撕裂会话）。放弃创建，连接由取代者的循环完成；被取代帧的
+ // socket 在新循环覆盖 _socket 时释放。sessionId==0 同理：编号已被归零
+ // 说明拆除正在并发进行，等待取代者重建。
+ if (Volatile.Read(ref _sessionEpoch) != epoch || sessionId == 0)
+ return;
+
+ StopSessionCore();
var cts = new CancellationTokenSource();
var token = cts.Token;
@@ -1182,19 +1240,24 @@ private void Shutdown()
// 消息通道归连接所有：停机是它唯一的完成点（正常完成，枚举端自然结束）
_messageRelay.Writer.TryComplete();
- if (_socket is { } socket)
+ // 与 StartAsync 的"停机守门 + 赋值"共用 _sessionGate：连接建立与停机互斥，
+ // 刚建立的 socket 不会漏进已完成的停机（#68）。
+ lock (_sessionGate)
{
- try
+ if (_socket is { } socket)
{
- if (socket.Connected)
- socket.Shutdown(SocketShutdown.Both);
- }
- catch (Exception ex)
- {
- _logger.LogDebug(ex, "Socket shutdown on disconnect failed.");
+ try
+ {
+ if (socket.Connected)
+ socket.Shutdown(SocketShutdown.Both);
+ }
+ catch (Exception ex)
+ {
+ _logger.LogDebug(ex, "Socket shutdown on disconnect failed.");
+ }
+ socket.Dispose();
+ _socket = null;
}
- socket.Dispose();
- _socket = null;
}
lock (_listenerGate)
diff --git a/test/StreamFrame.Tests/StateOwnershipBaselineTests.cs b/test/StreamFrame.Tests/StateOwnershipBaselineTests.cs
new file mode 100644
index 0000000..0699684
--- /dev/null
+++ b/test/StreamFrame.Tests/StateOwnershipBaselineTests.cs
@@ -0,0 +1,306 @@
+using System.Collections.Concurrent;
+using System.Net;
+using System.Net.Sockets;
+using System.Reflection;
+using System.Threading.Channels;
+
+namespace StreamFrame.Tests;
+
+///
+/// 状态机与会话所有权行为基线（#68 PR-1）：以确定性或不变量测试固定当前已承诺的语义——
+/// 终态后迟到发布的丢弃、用户回调重入（Connected 回调内 Reconnect/Dispose）、
+/// 连接成功与停机竞速、并发关闭/重连下的终态不变量。为后续会话所有权重构提供行为底线。
+///
+public class StateOwnershipBaselineTests
+{
+ private static readonly TimeSpan Budget = TimeSpan.FromSeconds(10);
+
+ 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 StreamConnection Create(int port, bool active, int retryDelayMs = 25)
+ => new(
+ new LengthPrefixFramer(),
+ StringCodec.Instance,
+ IPAddress.Loopback,
+ port,
+ isActive: active,
+ new StreamConnectionOptions { ConnectRetryDelayMs = retryDelayMs, AcceptRetryDelayMs = retryDelayMs });
+
+ private static object? Field(StreamConnection connection, string name)
+ => typeof(StreamConnection).GetField(name, BindingFlags.Instance | BindingFlags.NonPublic)!.GetValue(connection);
+
+ /// 白盒直发一次内部状态发布（迟到发布的确定性门控测试用）。
+ private static void PublishState(StreamConnection connection, ConnectionState state)
+ => typeof(StreamConnection)
+ .GetMethod("CommunicationStateChanging", BindingFlags.Instance | BindingFlags.NonPublic)!
+ .Invoke(connection, new object?[] { state, null });
+
+ private static async Task ReceiveOne(StreamConnection connection, int budgetMs = 5000)
+ {
+ using var timeout = new CancellationTokenSource(budgetMs);
+ await foreach (var message in connection.GetMessages(timeout.Token))
+ return message;
+ throw new InvalidOperationException("消息流提前结束");
+ }
+
+ private static async Task WaitForAsync(Func condition, string what, int budgetMs = 8000)
+ {
+ var deadline = TestClock.TickCount64 + budgetMs;
+ while (TestClock.TickCount64 < deadline)
+ {
+ if (condition())
+ return;
+ await Task.Delay(20);
+ }
+ throw new TimeoutException($"等待 {what} 超时");
+ }
+
+ /// 终态不变量：停机完成后 State/会话编号/会话资源必须收敛到唯一合法形态。
+ private static void AssertTerminalInvariants(StreamConnection connection, int port)
+ {
+ Assert.True(connection.IsDisposed);
+ Assert.Equal(ConnectionState.Disconnected, connection.State);
+ Assert.Equal(0, connection.CurrentSessionId);
+ Assert.Null(Field(connection, "_server")); // 监听器已释放
+ Assert.Null(Field(connection, "_socket")); // 连接 socket 已释放
+ Assert.Null(Field(connection, "_sessionCts")); // 无孤儿会话
+ Assert.Null(Field(connection, "_receiveTask"));
+ Assert.Null(Field(connection, "_decodeTask"));
+ Assert.Null(Field(connection, "_sendWorkerTask"));
+
+ // 监听端口可立即重新占用（未泄漏仍在监听的 socket）
+ var rebinding = new TcpListener(IPAddress.Loopback, port);
+ rebinding.Start();
+ rebinding.Stop();
+ }
+
+ [Fact]
+ public async Task DisposedState_LatePublications_AreDroppedEntirely()
+ {
+ var port = GetFreePort();
+ await using var server = Create(port, false);
+ await using var client = Create(port, true);
+ server.Start(CancellationToken.None);
+ client.Start(CancellationToken.None);
+ await Task.WhenAll(server.WaitForConnectedAsync(), client.WaitForConnectedAsync())
+ .WaitAsync(Budget);
+ var liveSessionId = server.CurrentSessionId;
+ Assert.NotEqual(0, liveSessionId);
+
+ await server.DisposeAsync();
+
+ // 停机后到达的迟到发布（用户 Reconnect / 迟到的会话故障）与竞速中的连接成功：
+ // 整体丢弃——终态不被翻转、会话编号不被归零后重分配、不创建会话任务。
+ PublishState(server, ConnectionState.Retry);
+ Assert.Equal(ConnectionState.Disconnected, server.State);
+ Assert.Equal(0, server.CurrentSessionId);
+
+ PublishState(server, ConnectionState.Connected);
+ Assert.Equal(ConnectionState.Disconnected, server.State);
+ Assert.Equal(0, server.CurrentSessionId);
+ Assert.Null(Field(server, "_sessionCts"));
+ Assert.Null(Field(server, "_receiveTask"));
+ Assert.Null(Field(server, "_whenConnected")); // 等待器不被迟到的 Connected 唤醒
+
+ AssertTerminalInvariants(server, port);
+ }
+
+ [Fact]
+ public async Task ConnectedCallback_ReconnectReentry_ConvergesToUsableSession()
+ {
+ var port = GetFreePort();
+ await using var server = Create(port, false);
+ await using var firstClient = Create(port, true);
+
+ var states = new ConcurrentQueue();
+ var connectedSessionIds = new ConcurrentQueue();
+ var reentered = 0;
+ server.ConnectionChanged += (_, state) =>
+ {
+ states.Enqueue(state);
+ if (state != ConnectionState.Connected)
+ return;
+
+ // 线性化契约：Connected 事件可见时编号必已分配（分配先于发布）
+ connectedSessionIds.Enqueue(server.CurrentSessionId);
+
+ // 病态重入：仅首个 Connected 回调里同步触发 Reconnect()
+ if (Interlocked.CompareExchange(ref reentered, 1, 0) == 0)
+ server.Reconnect();
+ };
+
+ server.Start(CancellationToken.None);
+ firstClient.Start(CancellationToken.None);
+ // 服务端就绪不走 WaitForConnectedAsync：回调内重连会消费掉 Connected#1 的等待器，
+ // 迟到的等待将挂到 Connected#2（第二个客户端之后才接入）。以客户端就绪 + 回调
+ // 执行标志为准。
+ await firstClient.WaitForConnectedAsync().WaitAsync(Budget);
+ await WaitForAsync(() => Volatile.Read(ref reentered) != 0, "首个 Connected 回调执行重入 Reconnect");
+ var firstSessionId = connectedSessionIds.Single();
+
+ // 重连后的新会话由第二个客户端接入
+ await using var secondClient = Create(port, true);
+ secondClient.Start(CancellationToken.None);
+ await secondClient.WaitForConnectedAsync().WaitAsync(Budget);
+ await WaitForAsync(() => server.State == ConnectionState.Connected && server.CurrentSessionId != firstSessionId,
+ "重入重连后的新会话发布");
+
+ // 收敛后的会话双向可用；编号在 Connected 可见期间恒非零且不被复用
+ await secondClient.SendAsync("after-reentry-reconnect");
+ Assert.Equal("after-reentry-reconnect", await ReceiveOne(server));
+ await server.SendAsync("reply");
+ Assert.Equal("reply", await ReceiveOne(secondClient));
+
+ var observedIds = connectedSessionIds.ToArray();
+ var observedStates = states.ToArray();
+ Assert.All(observedIds, id => Assert.NotEqual(0, id));
+ Assert.Equal(2, observedIds.Length);
+ Assert.NotEqual(observedIds[0], observedIds[1]);
+ Assert.Equal(server.CurrentSessionId, observedIds[1]);
+ Assert.Equal(ConnectionState.Connected, server.State);
+
+ // 观测到的状态序列只含合法转移，且终止于 Connected
+ Assert.Equal(ConnectionState.Connected, observedStates[^1]);
+ for (var i = 1; i < observedStates.Length; i++)
+ {
+ var legal = observedStates[i - 1] switch
+ {
+ ConnectionState.Connecting => observedStates[i] is ConnectionState.Connecting or ConnectionState.Connected or ConnectionState.Retry or ConnectionState.Disconnected,
+ ConnectionState.Connected => observedStates[i] is ConnectionState.Retry or ConnectionState.Connected or ConnectionState.Disconnected,
+ ConnectionState.Retry => observedStates[i] is ConnectionState.Retry or ConnectionState.Connecting or ConnectionState.Disconnected,
+ _ => false, // Disconnected 是终态，之后不得再有可观测转移
+ };
+ Assert.True(legal, $"非法状态转移：{observedStates[i - 1]} → {observedStates[i]}");
+ }
+ }
+
+ [Fact]
+ public async Task ConnectedCallback_DisposeReentry_TerminatesCleanly()
+ {
+ var port = GetFreePort();
+ var server = Create(port, false);
+ await using var client = Create(port, true);
+ var disposedInCallback = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ var disposeCompletedSynchronously = false;
+
+ server.ConnectionChanged += (_, state) =>
+ {
+ if (state == ConnectionState.Connected)
+ {
+ // 若停机挂起，回调无法返回：上方 WaitForConnectedAsync（完成点在回调之后）将超时暴露
+ disposeCompletedSynchronously = server.DisposeAsync().IsCompletedSuccessfully;
+ disposedInCallback.TrySetResult(true);
+ }
+ };
+
+ server.Start(CancellationToken.None);
+ // 服务端等待器先挂起：Connected 发布（回调之后）的完成点在停机取消之后——
+ // 回调内 Dispose 的合法结果就是等待器以取消收尾（#65 契约在重入场景同样成立）
+ var serverReady = server.WaitForConnectedAsync().WaitAsync(Budget);
+ client.Start(CancellationToken.None);
+ await client.WaitForConnectedAsync().WaitAsync(Budget);
+ await disposedInCallback.Task.WaitAsync(Budget);
+ Assert.True(disposeCompletedSynchronously);
+ await Assert.ThrowsAnyAsync(() => serverReady);
+
+ AssertTerminalInvariants(server, port);
+ Assert.Equal(0, server.CurrentSessionId);
+
+ // 停机后的公共契约：消息流自然结束、发送通道关闭
+ using var messageBudget = new CancellationTokenSource(Budget);
+ await using (var enumerator = server.GetMessages(messageBudget.Token).GetAsyncEnumerator())
+ Assert.False(await enumerator.MoveNextAsync()); // 通道已完成：立即 false；若悬挂则由预算令牌超时暴露
+ await Assert.ThrowsAnyAsync(() => server.SendAsync("late"));
+ await client.DisposeAsync();
+ }
+
+ [Theory]
+ [InlineData(0)]
+ [InlineData(1)]
+ [InlineData(2)]
+ public async Task ConnectSuccess_RacingDispose_TerminalInvariantsHold(int jitterMode)
+ {
+ // 连接建立与停机的竞速不变量：无论 accept 成功与 Dispose 的相对次序如何，
+ // 终态收敛且不泄漏（连接 socket / 监听器 / 孤儿会话任务）。
+ // 迭代量刻意保守：这是行为基线测试，不是 soak；过高并发会放大其它测试的
+ // 慢收敛敏感（StopSessionCore 同步等待），干扰并行套件。
+ const int iterations = 10;
+ for (var i = 0; i < iterations; i++)
+ {
+ var port = GetFreePort();
+ await using var server = Create(port, false);
+ server.Start(CancellationToken.None);
+
+ using var client = new TcpClient();
+ var connect = Task.Run(async () =>
+ {
+ try
+ {
+ if (jitterMode == 1)
+ await Task.Delay(1);
+ await client.ConnectAsync(IPAddress.Loopback, port).WaitAsync(Budget);
+ }
+ catch (SocketException)
+ {
+ // 服务端先停机：连接被拒，属预期分支
+ }
+ catch (OperationCanceledException)
+ {
+ }
+ });
+ var dispose = Task.Run(async () =>
+ {
+ if (jitterMode == 2)
+ await Task.Delay(1);
+ await server.DisposeAsync();
+ });
+
+ await Task.WhenAll(connect, dispose).WaitAsync(Budget);
+ client.Dispose();
+ AssertTerminalInvariants(server, port);
+ }
+ }
+
+ [Fact]
+ public async Task ConcurrentDisposeAndReconnect_TerminalInvariantsHold()
+ {
+ // 故障重连进行中并发执行用户 Reconnect 与 Dispose：全部参与者收敛后终态不变量成立。
+ // 迭代量刻意保守（同上：行为基线而非 soak，避免放大并行套件的慢收敛敏感）。
+ const int iterations = 8;
+ for (var i = 0; i < iterations; i++)
+ {
+ var port = GetFreePort();
+ var server = Create(port, false);
+ await using var client = Create(port, true);
+ server.Start(CancellationToken.None);
+ client.Start(CancellationToken.None);
+ await Task.WhenAll(server.WaitForConnectedAsync(), client.WaitForConnectedAsync())
+ .WaitAsync(Budget);
+
+ // 对端拆除 → 服务端进入故障重连（重新监听），与并发 Reconnect/Dispose 竞速
+ await client.DisposeAsync();
+
+ var actors = Enumerable.Range(0, 4).Select(n => Task.Run(() =>
+ {
+ try
+ {
+ server.Reconnect();
+ }
+ catch (ObjectDisposedException)
+ {
+ // 已停机：合法竞速结果
+ }
+ })).Append(Task.Run(() => server.DisposeAsync().AsTask()));
+
+ await Task.WhenAll(actors).WaitAsync(Budget);
+ AssertTerminalInvariants(server, port);
+ }
+ }
+}