diff --git a/.github/workflows/soak.yml b/.github/workflows/soak.yml
new file mode 100644
index 0000000..37c8b45
--- /dev/null
+++ b/.github/workflows/soak.yml
@@ -0,0 +1,49 @@
+name: soak
+
+# 长期观察：每夜以 600 秒跑浸泡/混沌套件（ubuntu + windows 双平台）。
+# 覆盖重连竞速（Soak_ReconnectRacing_LongRun，含状态机转移合法性与会话编号线性化的
+# 全程校验）、随机故障混沌与长流完整性。失败会通过 GitHub 通知提醒维护者。
+# 手动触发：workflow_dispatch 可指定时长（分钟数）。
+
+on:
+ schedule:
+ - cron: '0 19 * * *' # UTC 19:00 = 北京时间 03:00
+ workflow_dispatch:
+ inputs:
+ seconds:
+ description: '浸泡时长（秒）'
+ required: false
+ default: '600'
+
+permissions:
+ contents: read
+
+jobs:
+ soak:
+ name: soak (${{ matrix.os }})
+ runs-on: ${{ matrix.os }}
+ timeout-minutes: 60
+ strategy:
+ fail-fast: false
+ matrix:
+ os: [ubuntu-latest, windows-latest]
+
+ steps:
+ - uses: actions/checkout@v4
+
+ - name: Setup .NET
+ uses: actions/setup-dotnet@v4
+ with:
+ dotnet-version: |
+ 8.0.x
+ 10.0.x
+
+ - name: Build (Release)
+ run: dotnet build StreamFrame.slnx -c Release
+
+ - name: Soak tests
+ env:
+ STREAMFRAME_SOAK_SECONDS: ${{ inputs.seconds || '600' }}
+ run: >-
+ dotnet test test/StreamFrame.Tests/StreamFrame.Tests.csproj -c Release --no-build
+ --filter "FullyQualifiedName~Soak"
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 1fde7cb..b9d6025 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -24,6 +24,7 @@
- **大报文优化与归因**：发送编码缓冲自适应初始尺寸（按连接记忆上一帧高水位、封顶 1MB——稳态尺寸相近的协议不再从 1KB 起爬几何增长梯子）。新增 64KB 三 codec 归因基准（byte[] 透传 / string+span / string+分配式），结论：框架纯成本距裸 TCP 仅 ≈20–30%，其余为 codec 写法税（span 重载可消除中间数组）与 string 消息固有税（UTF-16 物化 ≈2× 报文）。README 双语新增"大报文指南"，bench/README 附归因表。
### 修复
+- **接受循环代次门控（#47 阶段二，竞速泄漏监听器 bug）**：用户重连与自动重连竞速产生多个并发 StartAsync 接受循环时，被取代的"僵尸"循环完成 accept 后可能看到 `_server` 已被新循环替换而跳过单客户端的监听器关闭——**泄漏一个仍在监听但永无人 accept 的 socket**，客户端 SYN 进入无人消费的 backlog 后 `ConnectAsync` 永久挂起（远比"拒绝"严重）。新增 `_acceptLoopId` 代次：每次 StartCore 递增，被取代的循环在取得 `_acceptLock` 后静默退出，监听器的创建/关闭/accept 只归属当代循环。由新增的重连竞速混沌（`Soak_ReconnectRacing_LongRun`，60% 动作占比并发竞速）复现并锁定。
- **被动端监听韧性加固（#47，防御性）**：排查确认原始"用户 `Reconnect()` 竞速楔死监听"**无法复现**（顺序/并发两组确定性复现 + 混沌套件恢复该动作后多轮全绿，此前观察到的楔死系当时测试自身缺陷），但修复排查中识别的两个真实弱点——① 接受循环的重试延迟改为锁外等待（旧实现整个重试循环持有 `_acceptLock`，绑定/接受持续失败期间会堵死其它接受循环）；② 监听 socket 设置 `SO_REUSEADDR`（服务端主动关闭后立即重绑不受 2MSL/TIME_WAIT 限制；Windows 实测宽松，Linux 上无此选项会遇 `EADDRINUSE`，一并设置保持跨平台一致）。新增确定性回归测试（8 轮服务端主动重连，端口须秒级可重新接受），混沌套件恢复用户重连动作。
## [2.3.1] - 2026-08-28
diff --git a/src/StreamFrame/Connection/StreamConnection.cs b/src/StreamFrame/Connection/StreamConnection.cs
index 5fd7e4e..f2986af 100644
--- a/src/StreamFrame/Connection/StreamConnection.cs
+++ b/src/StreamFrame/Connection/StreamConnection.cs
@@ -79,6 +79,15 @@ public string? RemoteIpAddress
private readonly object _sessionGate = new();
#endif
private Pipe? _pipe;
+
+ ///
+ /// 接受循环代次（#47）：每次 StartCore 递增。用户重连与自动重连竞速会产生多个并发的
+ /// StartAsync 接受循环——旧的"僵尸"循环完成 accept 后可能看到 _server 已被新循环替换，
+ /// 跳过单客户端的监听器关闭，泄漏一个"仍在监听但永无人 accept"的 socket（客户端 SYN
+ /// 进入无人消费的 backlog，ConnectAsync 永久挂起）。代次门控让被取代的循环在取得
+ /// _acceptLock 后静默退出：监听器的创建/关闭/accept 只归属当代循环。
+ ///
+ private int _acceptLoopId;
private CancellationTokenSource? _sessionCts;
private Task? _receiveTask;
private Task? _decodeTask;
@@ -209,9 +218,9 @@ public void Start(CancellationToken ct)
/// 内部启动入口（重连流程复用），不做一次性启动校验。
private void StartCore(CancellationToken ct)
- => _ = StartAsync(ct);
+ => _ = StartAsync(ct, Interlocked.Increment(ref _acceptLoopId));
- private async Task StartAsync(CancellationToken ct)
+ private async Task StartAsync(CancellationToken ct, int loopId)
{
var connected = false;
try
@@ -221,7 +230,13 @@ private async Task StartAsync(CancellationToken ct)
CommunicationStateChanging(ConnectionState.Connecting);
try
{
- _socket = IsActive ? await ConnectAsync(ct).ConfigureAwait(false) : await AcceptAsync(ct).ConfigureAwait(false);
+ 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)
+ return; // 排队等到锁时已被取代（AcceptAsync 内部的代次检查）
+
_retryScheduler.Reset();
CommunicationStateChanging(ConnectionState.Connected);
connected = true;
@@ -280,7 +295,7 @@ private async Task ConnectAsync(CancellationToken ct)
return socket;
}
- private async Task AcceptAsync(CancellationToken ct)
+ private async Task AcceptAsync(CancellationToken ct, int loopId)
{
while (true)
{
@@ -293,6 +308,12 @@ private async Task AcceptAsync(CancellationToken ct)
int? retryDelay = null;
try
{
+ // 代次门控（#47 阶段二）：排队等到锁时若已被更新的循环取代，静默退出——
+ // 监听器的创建/关闭/accept 只归属当代循环，杜绝僵尸循环泄漏监听器
+ // （泄漏的监听器无人 accept，客户端 SYN 进入 backlog 后 ConnectAsync 永久挂起）
+ if (Volatile.Read(ref _acceptLoopId) != loopId)
+ return null;
+
if (_server == null)
InitServer();
@@ -318,7 +339,8 @@ private async Task AcceptAsync(CancellationToken ct)
ConfigureSocket(accepted);
// 单客户端模式：accept 到第一个客户端后关闭监听 socket，
- // 后续连接在 TCP 层被立即拒绝。
+ // 后续连接在 TCP 层被立即拒绝。代次门控保证此处 _server 属于当代循环，
+ // 不会误关/漏关其它循环的监听器。
if (_options.AcceptFirstClientOnly && _server != null)
{
_server.Dispose();
diff --git a/test/StreamFrame.Tests/ReconnectWedgeTests.cs b/test/StreamFrame.Tests/ReconnectWedgeTests.cs
index 0a5bb1b..cbd59a6 100644
--- a/test/StreamFrame.Tests/ReconnectWedgeTests.cs
+++ b/test/StreamFrame.Tests/ReconnectWedgeTests.cs
@@ -67,7 +67,7 @@ private static async Task ConnectWithRetryAsync(TcpClient client, int port, int
}
}
- private static async Task WaitConnectedAsync(StreamConnection server, int timeoutMs = 5000)
+ private static async Task WaitConnectedAsync(StreamConnection server, int timeoutMs = 10_000)
{
var deadline = TestClock.TickCount64 + timeoutMs;
while (TestClock.TickCount64 < deadline && server.State != ConnectionState.Connected)
@@ -93,14 +93,14 @@ public async Task RapidServerInitiatedReconnects_PortStaysAcceptable()
using (var client = new TcpClient())
{
await ConnectWithRetryAsync(client, port);
- await WaitConnectedAsync(server);
+ await WaitConnectedAsync(server, timeoutMs: 10_000);
}
// 等自动重连完成、再次形成会话
using (var client = new TcpClient())
{
await ConnectWithRetryAsync(client, port);
- await WaitConnectedAsync(server, 8000);
+ await WaitConnectedAsync(server, 15_000);
// 两路并发触发（#47 的竞速形态）：杀对端（自动重连）与用户显式 Reconnect() 竞争
var kill = Task.Run(client.Dispose);
@@ -108,7 +108,7 @@ public async Task RapidServerInitiatedReconnects_PortStaysAcceptable()
await kill;
}
- var rebindDeadline = TestClock.TickCount64 + 10_000; // 修复目标：秒级可用（10s 为宽松上限）
+ var rebindDeadline = TestClock.TickCount64 + 20_000; // 修复目标：秒级可用（20s 上限：并行负载下偶发慢收敛，曾以 1/15 概率闪失）
while (TestClock.TickCount64 < rebindDeadline)
{
if (await TryConnectAsync(port))
diff --git a/test/StreamFrame.Tests/SoakTests.cs b/test/StreamFrame.Tests/SoakTests.cs
index 8b88a2e..db1ff33 100644
--- a/test/StreamFrame.Tests/SoakTests.cs
+++ b/test/StreamFrame.Tests/SoakTests.cs
@@ -11,12 +11,103 @@ namespace StreamFrame.Tests;
/// 浸泡/混沌测试（默认不运行）：真实 TCP 回环上的长时间运行与随机故障注入，
/// 验证消息完整性（顺序、无重复、无跨会话错发）与无悬挂。
///
-/// 启用方式（本地手动触发，CI 与默认 dotnet test 均不运行）：
+/// 启用方式（本地手动触发，常规 CI 与默认 dotnet test 均不运行）：
/// STREAMFRAME_SOAK_SECONDS=120 dotnet test -f net8.0 --filter FullyQualifiedName~Soak
/// 未设置环境变量时本套件立即返回（视为通过），不影响提交前的全量验证。
+/// 长期观察：.github/workflows/soak.yml 每夜以 600s 跑本套件（ubuntu + windows），
+/// 覆盖用户重连竞速场景（Soak_ReconnectRacing_LongRun，含状态机转移合法性与
+/// 会话编号不变式的全程校验）。
///
public class SoakTests
{
+ private readonly Xunit.Abstractions.ITestOutputHelper _output;
+
+ public SoakTests(Xunit.Abstractions.ITestOutputHelper output) => _output = output;
+
+ /// 把库内 Warning（accept 失败等）透传到测试输出——长期观察失败时的第一手诊断。
+ private sealed class SoakDumpLogger : Microsoft.Extensions.Logging.ILogger
+ {
+ private readonly Xunit.Abstractions.ITestOutputHelper _sink;
+ public SoakDumpLogger(Xunit.Abstractions.ITestOutputHelper sink) => _sink = sink;
+ public IDisposable? BeginScope(TState state) where TState : notnull => null;
+ public bool IsEnabled(Microsoft.Extensions.Logging.LogLevel logLevel) => true;
+ public void Log(Microsoft.Extensions.Logging.LogLevel logLevel, Microsoft.Extensions.Logging.EventId eventId, TState state, Exception? exception, Func formatter)
+ {
+ var line = $"[LOG {logLevel}] {formatter(state, exception)}{(exception is null ? string.Empty : " (" + exception.GetType().Name + ")")}";
+ _sink.WriteLine(line);
+ }
+ }
+
+ ///
+ /// 状态机转移的全程记录与校验器（长期观察的核心断言面）：
+ /// 合法边检查、Retry 后必须收敛回 Connected、非 Connected 态会话编号必须为 0、
+ /// Connected 态编号严格单调不复用；另统计 Connected→Connected 直连迁移
+ /// （双 StartAsync 接受循环竞速的观测指标，合法但值得长期盯）。
+ ///
+ private sealed class StateTransitionRecorder
+ {
+ private readonly List<(ConnectionState State, long SessionId)> _transitions = new(); // 兼作监视锁（net48 需 object 锁）
+ private long _lastConnectedId;
+
+ public int DirectConnectedMigrations;
+
+ public void Record(ConnectionState state, long sessionId)
+ {
+ lock (_transitions)
+ {
+ if (_transitions.Count > 0)
+ {
+ var previous = _transitions[^1].State;
+ var legal = (previous, state) switch
+ {
+ (ConnectionState.Connecting, ConnectionState.Connecting) => true, // 连接/接受重试
+ (ConnectionState.Connecting, ConnectionState.Connected) => true,
+ (ConnectionState.Connecting, ConnectionState.Retry) => true, // 用户在 Connecting 中触发重连
+ (ConnectionState.Connected, ConnectionState.Connected) => true, // 双 StartAsync 直连迁移（竞速）
+ (ConnectionState.Connected, ConnectionState.Retry) => true,
+ (ConnectionState.Connected, ConnectionState.Disconnected) => true,
+ (ConnectionState.Retry, ConnectionState.Connecting) => true,
+ (ConnectionState.Retry, ConnectionState.Retry) => true, // 重连期间再次显式重连
+ (ConnectionState.Retry, ConnectionState.Disconnected) => true,
+ (ConnectionState.Connecting, ConnectionState.Disconnected) => true,
+ _ => false,
+ };
+ if (!legal)
+ throw new InvalidOperationException($"非法状态转移：{previous} → {state}");
+
+ if (previous == ConnectionState.Connected && state == ConnectionState.Connected)
+ DirectConnectedMigrations++;
+ }
+
+ _transitions.Add((state, sessionId));
+ }
+ }
+
+ /// 终局校验：编号线性化不变式 + 收敛性（终态 Disconnected）。
+ public void Validate()
+ {
+ lock (_transitions)
+ {
+ foreach (var (state, sessionId) in _transitions)
+ {
+ if (state == ConnectionState.Connected)
+ {
+ Assert.NotEqual(0L, sessionId);
+ Assert.True(sessionId > _lastConnectedId, $"会话编号必须严格递增：{_lastConnectedId} → {sessionId}");
+ _lastConnectedId = sessionId;
+ }
+ else
+ {
+ Assert.Equal(0L, sessionId); // 离开 Connected 可见时编号必须已归零（P1-1 保证）
+ }
+ }
+
+ // 终态必须是 Disconnected（DisposeAsync 之后）
+ Assert.Equal(ConnectionState.Disconnected, _transitions[^1].State);
+ }
+ }
+ }
+
private static int GetFreePort()
{
var listener = new TcpListener(IPAddress.Loopback, 0);
@@ -386,4 +477,203 @@ public async Task Soak_Chaos_RandomFaults_NoHangAndIntegrity()
var boundFrames = peerFrames.Where(f => f.StartsWith("b", StringComparison.Ordinal)).ToArray();
Assert.Equal(boundFrames.Length, boundFrames.Distinct().Count());
}
+
+ ///
+ /// 重连竞速的长期观察（默认不运行）：用户显式 Reconnect() 与自动重连（对端死亡）高占比
+ /// 并发竞速——双 StartAsync 接受循环、Connected→Connected 直连迁移、会话编号线性化等
+ /// #47 排查中判定"不可复现但结构脆弱"的路径，靠长时间高压力 + 全程不变式校验盯着。
+ /// 动作分布：~40% 竞速重连（杀对端与 Reconnect 并发）、~20% 纯用户重连、~30% 普通发送
+ /// （验证跨会话续发的 FIFO 完整性）、~10% 短等待；对端仅在无存活连接时重接入
+ /// （单客户端模式：服务端 Connected 时监听已关闭，并发第二条连接被拒是正确行为）。
+ ///
+ [Fact]
+ public async Task Soak_ReconnectRacing_LongRun()
+ {
+ var seconds = SoakSeconds();
+ if (seconds <= 0)
+ return; // 未启用浸泡模式：跳过
+
+ var output = _output;
+ var seed = Environment.TickCount;
+ var random = new Random(seed);
+ var port = GetFreePort();
+
+ var recorder = new StateTransitionRecorder();
+ await using var server = new StreamConnection(
+ new LengthPrefixFramer(), StringCodec.Instance, IPAddress.Loopback, port,
+ isActive: false, new StreamConnectionOptions { SendQueueCapacity = 256, AcceptRetryDelayMs = 200 },
+ logger: new SoakDumpLogger(output));
+ server.ConnectionChanged += (_, state) => recorder.Record(state, server.CurrentSessionId);
+ server.Start(CancellationToken.None);
+
+ var peerFrames = new ConcurrentQueue();
+ var pendingPlain = new ConcurrentQueue();
+ var plainSeq = 0;
+ TcpClient? client = null;
+
+ var deadline = TestClock.TickCount64 + (int)(seconds * 1000);
+ try
+ {
+ while (TestClock.TickCount64 < deadline)
+ {
+ var action = random.Next(100);
+ if (client is null)
+ {
+ // 对端（重）接入（仅在无存活对端时——单客户端模式：服务端 Connected
+ // 时监听已关闭，并发第二条连接被拒是正确行为）
+ client = new TcpClient();
+ await ConnectWithRetryAsync(client, port);
+ var currentStream = client.GetStream();
+ var readCts = new CancellationTokenSource(TimeSpan.FromSeconds(seconds + 60));
+ _ = Task.Run(async () =>
+ {
+ var header = new byte[4];
+ try
+ {
+ while (true)
+ {
+ var read = 0;
+ while (read < 4)
+ {
+ var n = await currentStream.ReadAsync(header, read, 4 - read, readCts.Token);
+ if (n == 0)
+ return;
+ read += n;
+ }
+
+ var length = BinaryPrimitives.ReadInt32BigEndian(header);
+ var payload = new byte[length];
+ read = 0;
+ while (read < length)
+ {
+ var n = await currentStream.ReadAsync(payload, read, length - read, readCts.Token);
+ if (n == 0)
+ return;
+ read += n;
+ }
+
+ peerFrames.Enqueue(Encoding.UTF8.GetString(payload));
+ }
+ }
+ catch (Exception ex) when (ex is ObjectDisposedException or IOException or OperationCanceledException)
+ {
+ }
+ });
+ }
+ else if (action < 50)
+ {
+ // 竞速重连：杀对端（触发自动重连）与用户显式 Reconnect() 同时打——
+ // 两路 Retry 过渡竞争，可能产生双 StartAsync 接受循环
+ var victim = client;
+ var kill = Task.Run(() => victim.Dispose());
+ server.Reconnect();
+ await kill;
+ await WaitForStateAsync(server, st => st != ConnectionState.Connected, timeoutMs: 15_000);
+ client = null;
+ }
+ else if (action < 70)
+ {
+ // 纯用户重连（对端保持存活：服务端主动关闭路径）
+ server.Reconnect();
+ await WaitForStateAsync(server, st => st != ConnectionState.Connected, timeoutMs: 15_000);
+ client?.Dispose();
+ client = null;
+ }
+ else if (action < 95)
+ {
+ // 普通发送：跨会话续发的 FIFO 完整性探针
+ for (var i = 0; i < random.Next(1, 4); i++)
+ {
+ var label = $"p{plainSeq++}";
+ pendingPlain.Enqueue(label);
+ await server.SendAsync(label);
+ }
+ }
+ else
+ {
+ await Task.Delay(random.Next(20, 100));
+ }
+
+ await Task.Delay(random.Next(10, 60));
+ }
+ }
+ finally
+ {
+ client?.Dispose();
+ var exitDeadline = TestClock.TickCount64 + 15_000;
+ while (TestClock.TickCount64 < exitDeadline && server.State == ConnectionState.Connected)
+ await Task.Delay(20);
+ }
+
+ // 终局：干净对端接入，等普通消息全部续发送达
+ using (var finalClient = new TcpClient())
+ {
+ await ConnectWithRetryAsync(finalClient, port);
+ await WaitForStateAsync(server, st => st == ConnectionState.Connected, timeoutMs: 30_000);
+ var finalStream = finalClient.GetStream();
+ var finalCts = new CancellationTokenSource(TimeSpan.FromSeconds(30));
+ var finalReader = Task.Run(async () =>
+ {
+ var header = new byte[4];
+ try
+ {
+ while (true)
+ {
+ var read = 0;
+ while (read < 4)
+ {
+ var n = await finalStream.ReadAsync(header, read, 4 - read, finalCts.Token);
+ if (n == 0)
+ return;
+ read += n;
+ }
+
+ var length = BinaryPrimitives.ReadInt32BigEndian(header);
+ var payload = new byte[length];
+ read = 0;
+ while (read < length)
+ {
+ var n = await finalStream.ReadAsync(payload, read, length - read, finalCts.Token);
+ if (n == 0)
+ return;
+ read += n;
+ }
+
+ peerFrames.Enqueue(Encoding.UTF8.GetString(payload));
+ }
+ }
+ catch (Exception ex) when (ex is ObjectDisposedException or IOException or OperationCanceledException)
+ {
+ }
+ });
+
+ var finalDeadline = TestClock.TickCount64 + 30_000;
+ while (TestClock.TickCount64 < finalDeadline)
+ {
+ var plainSeen = 0;
+ foreach (var f in peerFrames)
+ if (f.StartsWith("p", StringComparison.Ordinal))
+ plainSeen++;
+ if (plainSeen >= pendingPlain.Count)
+ break;
+ await Task.Delay(100);
+ }
+
+ finalCts.Cancel();
+ await finalReader;
+ }
+
+ await server.DisposeAsync();
+
+ // 全程不变式：合法转移、编号线性化、终态收敛
+ recorder.Validate();
+
+ // FIFO 完整性：普通消息无丢失无重复
+ var plainFrames = peerFrames.Where(f => f.StartsWith("p", StringComparison.Ordinal)).ToArray();
+ Assert.Equal(pendingPlain.Count, plainFrames.Length);
+ Assert.Equal(pendingPlain.ToArray(), plainFrames);
+
+ output.WriteLine($"[soak-racing] seed={seed}, 普通消息 {plainSeq} 条全部送达；" +
+ $"Connected→Connected 直连迁移 {recorder.DirectConnectedMigrations} 次（竞速观测指标）");
+ }
}