diff --git a/CHANGELOG.md b/CHANGELOG.md index 98bd91d..7c90065 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,7 +15,7 @@ - demo 新增场景 5：会话感知收发（编号生命周期、整帧写完才完成、旧会话发送失效不重放、新会话照常收发）。 ### 修复 -- 暂无 +- **被动端监听韧性加固（#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 8e68e02..e0b58a9 100644 --- a/src/StreamFrame/Connection/StreamConnection.cs +++ b/src/StreamFrame/Connection/StreamConnection.cs @@ -282,12 +282,17 @@ private async Task ConnectAsync(CancellationToken ct) private async Task AcceptAsync(CancellationToken ct) { - await _acceptLock.WaitAsync(ct).ConfigureAwait(false); - try + while (true) { - while (true) + ct.ThrowIfCancellationRequested(); + + // 逐次尝试获取锁、重试延迟在锁外等待：绑定/接受持续失败期间不堵死其它 + // 接受循环（#47 防御性修复——旧实现整个重试循环持锁，故障期被放大为全停） + await _acceptLock.WaitAsync(ct).ConfigureAwait(false); + Socket? accepted = null; + int? retryDelay = null; + try { - ct.ThrowIfCancellationRequested(); if (_server == null) InitServer(); @@ -296,11 +301,21 @@ private async Task AcceptAsync(CancellationToken ct) #if NETSTANDARD2_0 var listener = _server!; // FromAsync 无法取消；停机时 listener 被释放，挂起的 accept 以异常收尾 - var socket = await Task.Factory.FromAsync(listener.BeginAccept, listener.EndAccept, null).ConfigureAwait(false); + accepted = await Task.Factory.FromAsync(listener.BeginAccept, listener.EndAccept, null).ConfigureAwait(false); #else - var socket = await _server!.AcceptAsync(ct).ConfigureAwait(false); + accepted = await _server!.AcceptAsync(ct).ConfigureAwait(false); #endif - ConfigureSocket(socket); + } + catch (Exception ex) when (!ct.IsCancellationRequested && !IsDisposed) + { + retryDelay = _retryScheduler.NextDelayMs(); + _logger.LogWarning(ex, "Accept on {Local} failed (attempt {Attempt}); retry in {Delay}ms.", + $"{IpAddress}:{Port}", _retryScheduler.Attempt, retryDelay.Value); + } + + if (accepted is not null) + { + ConfigureSocket(accepted); // 单客户端模式：accept 到第一个客户端后关闭监听 socket， // 后续连接在 TCP 层被立即拒绝。 @@ -309,21 +324,17 @@ private async Task AcceptAsync(CancellationToken ct) _server.Dispose(); _server = null; } - - return socket; - } - catch (Exception ex) when (!ct.IsCancellationRequested && !IsDisposed) - { - var delay = _retryScheduler.NextDelayMs(); - _logger.LogWarning(ex, "Accept on {Local} failed (attempt {Attempt}); retry in {Delay}ms.", - $"{IpAddress}:{Port}", _retryScheduler.Attempt, delay); - await Task.Delay(delay, ct).ConfigureAwait(false); } } - } - finally - { - _acceptLock.Release(); + finally + { + _acceptLock.Release(); + } + + if (accepted is not null) + return accepted; + + await Task.Delay(retryDelay ?? _retryScheduler.NextDelayMs(), ct).ConfigureAwait(false); } } @@ -337,6 +348,10 @@ private void InitServer() _server = CreateTcpSocket(); _server.Blocking = false; + // 允许重绑覆盖遗留的 TIME_WAIT：服务端主动关闭（用户 Reconnect/停机）后立即重新 + // 监听不受 2MSL 限制（Linux 上没有该选项会遇到 EADDRINUSE；Windows 实测宽松， + // 一并设置保持跨平台行为一致——#47 防御性修复） + _server.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, true); _server.Bind(new IPEndPoint(NormalizeListenAddress(IpAddress), Port)); _server.Listen(0); } diff --git a/test/StreamFrame.Tests/ReconnectWedgeTests.cs b/test/StreamFrame.Tests/ReconnectWedgeTests.cs new file mode 100644 index 0000000..0a5bb1b --- /dev/null +++ b/test/StreamFrame.Tests/ReconnectWedgeTests.cs @@ -0,0 +1,124 @@ +using System.Net; +using System.Net.Sockets; +using StreamFrame; + +namespace StreamFrame.Tests; + +/// +/// #47 的确定性复现/回归：服务端主动发起的重连（用户显式 Reconnect()，尤其是与自动重连 +/// 并发竞速时）反复发生后，被动端必须仍能在秒级内接受新连接。楔死形态 = 监听端口重绑失败 +/// 或接受循环被堵（疑似 TIME_WAIT 占用 + 未设 SO_REUSEADDR；accept 重试延迟持锁放大故障时长）。 +/// +public class ReconnectWedgeTests +{ + private readonly Xunit.Abstractions.ITestOutputHelper _output; + + public ReconnectWedgeTests(Xunit.Abstractions.ITestOutputHelper output) => _output = output; + + private sealed class DumpLogger : Microsoft.Extensions.Logging.ILogger + { + private readonly Xunit.Abstractions.ITestOutputHelper _sink; + public DumpLogger(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) + => _sink.WriteLine($"[LOG {logLevel}] {formatter(state, exception)}{(exception is null ? string.Empty : " (" + exception.GetType().Name + ")")}"); + } + + 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 async Task TryConnectAsync(int port) + { + try + { + using var probe = new TcpClient(); + await probe.ConnectAsync(IPAddress.Loopback, port); + return true; + } + catch (SocketException) + { + return false; + } + } + + private static async Task ConnectWithRetryAsync(TcpClient client, int port, int timeoutMs = 10_000) + { + // Start() 是异步的：监听器完成 bind 前连接会被拒（CI 的 ubuntu 上线程池调度更慢， + // 必须容忍启动窗口），重试直到超时 + var deadline = TestClock.TickCount64 + timeoutMs; + while (true) + { + try + { + await client.ConnectAsync(IPAddress.Loopback, port); + return; + } + catch (SocketException) when (TestClock.TickCount64 < deadline) + { + await Task.Delay(100); + } + } + } + + private static async Task WaitConnectedAsync(StreamConnection server, int timeoutMs = 5000) + { + var deadline = TestClock.TickCount64 + timeoutMs; + while (TestClock.TickCount64 < deadline && server.State != ConnectionState.Connected) + await Task.Delay(20); + Assert.Equal(ConnectionState.Connected, server.State); + } + + [Fact] + public async Task RapidServerInitiatedReconnects_PortStaysAcceptable() + { + const int rounds = 8; + var port = GetFreePort(); + + await using var server = new StreamConnection( + new LengthPrefixFramer(), StringCodec.Instance, IPAddress.Loopback, port, + isActive: false, new StreamConnectionOptions { AcceptRetryDelayMs = 200 }, + logger: new DumpLogger(_output)); + server.Start(CancellationToken.None); + + for (var round = 0; round < rounds; round++) + { + // 对端接入形成会话 + using (var client = new TcpClient()) + { + await ConnectWithRetryAsync(client, port); + await WaitConnectedAsync(server); + } + + // 等自动重连完成、再次形成会话 + using (var client = new TcpClient()) + { + await ConnectWithRetryAsync(client, port); + await WaitConnectedAsync(server, 8000); + + // 两路并发触发（#47 的竞速形态）：杀对端（自动重连）与用户显式 Reconnect() 竞争 + var kill = Task.Run(client.Dispose); + server.Reconnect(); + await kill; + } + + var rebindDeadline = TestClock.TickCount64 + 10_000; // 修复目标：秒级可用（10s 为宽松上限） + while (TestClock.TickCount64 < rebindDeadline) + { + if (await TryConnectAsync(port)) + break; + await Task.Delay(100); + } + + Assert.True(TestClock.TickCount64 < rebindDeadline, + $"第 {round} 轮：服务端主动重连后 {10_000}ms 内无法接受新连接（监听楔死，#47）。"); + _output.WriteLine($"[{TestClock.TickCount64 % 100000}] 第 {round} 轮重绑可用 ✓"); + } + } +} diff --git a/test/StreamFrame.Tests/SoakTests.cs b/test/StreamFrame.Tests/SoakTests.cs index 144995c..8b88a2e 100644 --- a/test/StreamFrame.Tests/SoakTests.cs +++ b/test/StreamFrame.Tests/SoakTests.cs @@ -29,7 +29,7 @@ private static int GetFreePort() private static double SoakSeconds() { var raw = Environment.GetEnvironmentVariable("STREAMFRAME_SOAK_SECONDS"); - return double.TryParse(raw, out var value) && value > 0 ? value : 0; + return double.TryParse(raw, System.Globalization.NumberStyles.Float, System.Globalization.CultureInfo.InvariantCulture, out var value) && value > 0 ? value : 0; } private static async Task ConnectWithRetryAsync(TcpClient client, int port) @@ -260,12 +260,9 @@ public async Task Soak_Chaos_RandomFaults_NoHangAndIntegrity() await WaitForStateAsync(server, s => s != ConnectionState.Connected); } } - else + else if (action < 97) { // 杀对端（RST 路径：Linger(true,0) 硬复位） - // 注：不混入用户显式 Reconnect()——用户重连与自动重连并发竞速会触发 - // 双 StartAsync 接受循环的存量脆弱点（监听状态机楔死，v2.3.0 评审 - // "附带发现"同源），属独立的状态过渡原子化设计任务，不在本套件覆盖 if (client is not null) { client.LingerState = new LingerOption(true, 0); @@ -275,6 +272,19 @@ public async Task Soak_Chaos_RandomFaults_NoHangAndIntegrity() await WaitForStateAsync(server, s => s != ConnectionState.Connected); } } + else + { + // 用户显式重连（#47 观察到的楔死触发器，与本会话并发故障竞速）： + // 不等当前会话的自动重连，立即发起服务端主动拆除 + server.Reconnect(); + if (client is not null) + { + await WaitForStateAsync(server, s => s != ConnectionState.Connected); + client.Dispose(); + client = null; + stream = null; + } + } await Task.Delay(random.Next(20, 120)); }