From 3f89cebb948239177525cce7c76821298638530b Mon Sep 17 00:00:00 2001 From: Marci Date: Sat, 29 Aug 2026 13:01:18 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E6=8E=A5=E5=8F=97=E5=BE=AA=E7=8E=AF?= =?UTF-8?q?=E4=BB=A3=E6=AC=A1=E9=97=A8=E6=8E=A7=E2=80=94=E2=80=94=E4=BF=AE?= =?UTF-8?q?=E5=A4=8D=E9=87=8D=E8=BF=9E=E7=AB=9E=E9=80=9F=E6=B3=84=E6=BC=8F?= =?UTF-8?q?=E7=9B=91=E5=90=AC=E5=99=A8=E5=AF=BC=E8=87=B4=E7=9A=84=E5=AE=A2?= =?UTF-8?q?=E6=88=B7=E7=AB=AF=E6=B0=B8=E4=B9=85=E6=8C=82=E8=B5=B7=EF=BC=88?= =?UTF-8?q?#47=20=E9=98=B6=E6=AE=B5=E4=BA=8C=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 竞速混沌（Soak_ReconnectRacing_LongRun，60% 动作并发竞速）复现：用户重连与 自动重连竞速产生多个并发 StartAsync 接受循环时，被取代的僵尸循环完成 accept 后看到 _server 已被新循环替换，跳过单客户端的监听器关闭——泄漏一个仍在监听 但永无人 accept 的 socket，客户端 SYN 进入无人消费的 backlog 后 ConnectAsync 永久挂起（形态为'卡死 220 秒'或终局连接拒绝，均实测捕获）。 修复：_acceptLoopId 代次（StartCore 递增）；被取代的循环在取得 _acceptLock 后 静默退出，监听器的创建/关闭/accept 只归属当代循环。 随附长期观察基建： - StateTransitionRecorder：全程记录并校验状态机合法转移、非 Connected 态 编号归零（P1-1 线性化不变式）、Connected 态编号严格递增、终态收敛；统计 Connected→Connected 直连迁移（竞速观测指标） - Soak_ReconnectRacing_LongRun：重连动作高占比（40% 竞速 + 20% 纯用户重连） + 普通消息 FIFO 完整性探针 + 状态机全程校验，env 门控 - soak.yml：每夜 03:00（北京时间）600s 双平台跑浸泡套件，失败自动通知 - 楔形回归测试预算放宽（并行负载下 ~1/15 概率闪失的时序余量） --- .github/workflows/soak.yml | 49 +++ CHANGELOG.md | 1 + .../Connection/StreamConnection.cs | 32 +- test/StreamFrame.Tests/ReconnectWedgeTests.cs | 8 +- test/StreamFrame.Tests/SoakTests.cs | 292 +++++++++++++++++- 5 files changed, 372 insertions(+), 10 deletions(-) create mode 100644 .github/workflows/soak.yml 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} 次（竞速观测指标）"); + } }