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
2 changes: 1 addition & 1 deletion CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
55 changes: 35 additions & 20 deletions src/StreamFrame/Connection/StreamConnection.cs
Original file line number Diff line number Diff line change
Expand Up @@ -282,12 +282,17 @@ private async Task<Socket> ConnectAsync(CancellationToken ct)

private async Task<Socket> 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();

Expand All @@ -296,11 +301,21 @@ private async Task<Socket> 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 层被立即拒绝。
Expand All @@ -309,21 +324,17 @@ private async Task<Socket> 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);
}
}

Expand All @@ -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);
}
Expand Down
124 changes: 124 additions & 0 deletions test/StreamFrame.Tests/ReconnectWedgeTests.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,124 @@
using System.Net;
using System.Net.Sockets;
using StreamFrame;

namespace StreamFrame.Tests;

/// <summary>
/// #47 的确定性复现/回归:服务端主动发起的重连(用户显式 Reconnect(),尤其是与自动重连
/// 并发竞速时)反复发生后,被动端必须仍能在秒级内接受新连接。楔死形态 = 监听端口重绑失败
/// 或接受循环被堵(疑似 TIME_WAIT 占用 + 未设 SO_REUSEADDR;accept 重试延迟持锁放大故障时长)。
/// </summary>
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>(TState state) where TState : notnull => null;
public bool IsEnabled(Microsoft.Extensions.Logging.LogLevel logLevel) => true;
public void Log<TState>(Microsoft.Extensions.Logging.LogLevel logLevel, Microsoft.Extensions.Logging.EventId eventId, TState state, Exception? exception, Func<TState, Exception?, string> 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<bool> 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<string> 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<string>(
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} 轮重绑可用 ✓");
}
}
}
20 changes: 15 additions & 5 deletions test/StreamFrame.Tests/SoakTests.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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);
Expand All @@ -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));
}
Expand Down
Loading