diff --git a/CHANGELOG.md b/CHANGELOG.md index 56a01d2..6bb6b98 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,7 @@ ## [Unreleased] ### 修复 +- **终态连接等待([#65](https://github.com/CSJ608/StreamFrame/issues/65))**:取得连接等待器后复查停机状态,关闭首次注册与 Shutdown 的竞态;Dispose 或生命周期取消后的新等待也以任务取消结束,无需调用方令牌。补充终态调用、1000 次首次注册竞速及调用方取消隔离回归。 - **坏头后完整帧交付([#64](https://github.com/CSJ608/StreamFrame/issues/64))**:定界器返回 false 但已消费无效前缀时继续解析现有缓冲,无进展才等待输入;保留非法长度头四字节丢弃策略,避免完整帧挂起及误触发半帧超时。补充推进契约及连续坏头、粘包、半帧、EOF、自定义定界器和错误元数据回归。 - **监听绑定失败恢复([#63](https://github.com/CSJ608/StreamFrame/issues/63))**:监听 Socket 完成配置、Bind/Listen 后才发布,失败释放并保持字段为空,端口释放后可自动重试接入;发布与停机释放互斥,保留接受循环代次及单客户端监听关闭语义。新增同步信号门控的端口占用恢复、双向消息及取消/Dispose 竞态回归。 - **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 必需检查名不变。 diff --git a/src/StreamFrame/Connection/StreamConnection.cs b/src/StreamFrame/Connection/StreamConnection.cs index d06c4a1..16141ee 100644 --- a/src/StreamFrame/Connection/StreamConnection.cs +++ b/src/StreamFrame/Connection/StreamConnection.cs @@ -1074,10 +1074,18 @@ public void Fault(Exception exception) /// public Task WaitForConnectedAsync(CancellationToken ct = default) { - if (State == ConnectionState.Connected) + if (!IsDisposed && State == ConnectionState.Connected) return Task.CompletedTask; var tcs = GetOrCreateWhenConnected(); + // 必须在注册后复查:Shutdown 可能已清空旧等待器,随后本次才发布新等待器。 + // 若终态发生在复查之后,Shutdown 会取得已发布的等待器并负责取消。 + if (IsDisposed) + { + tcs.TrySetCanceled(); + return tcs.Task; + } + return State == ConnectionState.Connected ? Task.CompletedTask // 获取等待器的间隙恰好连上了 : tcs.Task.WaitAsync(ct); diff --git a/test/StreamFrame.Tests/ConnectionUsabilityTests.cs b/test/StreamFrame.Tests/ConnectionUsabilityTests.cs index c684c92..4426666 100644 --- a/test/StreamFrame.Tests/ConnectionUsabilityTests.cs +++ b/test/StreamFrame.Tests/ConnectionUsabilityTests.cs @@ -40,11 +40,12 @@ public async Task WaitForConnectedAsync_CompletesWhenConnectionEstablished() await using var server = CreateConnection(port, isActive: false); client.Start(CancellationToken.None); + var pending = client.WaitForConnectedAsync(); // 服务端稍后才启动:等待者必须跨过若干次连接失败重试 await Task.Delay(400); server.Start(CancellationToken.None); - await client.WaitForConnectedAsync().WaitAsync(TimeSpan.FromSeconds(5)); + await pending.WaitAsync(TimeSpan.FromSeconds(5)); await server.WaitForConnectedAsync().WaitAsync(TimeSpan.FromSeconds(5)); Assert.Equal(ConnectionState.Connected, client.State); } @@ -93,6 +94,74 @@ public async Task WaitForConnectedAsync_Dispose_CancelsPendingWaiter() await Assert.ThrowsAnyAsync(() => waitTask); } + [Fact] + public async Task WaitForConnectedAsync_AfterDispose_IsCanceled() + { + await using var client = CreateConnection(12345, isActive: true); + await client.DisposeAsync(); + + var wait = client.WaitForConnectedAsync(); + Assert.True(wait.IsCanceled); + await Assert.ThrowsAnyAsync(() => wait); + } + + [Fact] + public async Task WaitForConnectedAsync_FirstRegistrationRacesDispose_NoOrphanedWaiter() + { + for (var iteration = 0; iteration < 1000; iteration++) + { + await using var client = CreateConnection(12345, isActive: true); + using var start = new Barrier(2); + Task? wait = null; + var register = Task.Run(() => + { + Assert.True(start.SignalAndWait(TimeSpan.FromSeconds(5))); + wait = client.WaitForConnectedAsync(); + }); + var shutdown = Task.Run(async () => + { + Assert.True(start.SignalAndWait(TimeSpan.FromSeconds(5))); + await client.DisposeAsync(); + }); + await Task.WhenAll(register, shutdown).WaitAsync(TimeSpan.FromSeconds(10)); + Assert.NotNull(wait); + // net48 的 WaitAsync 兼容包装通过异步延续传播取消,必须等待传播完成。 + await Assert.ThrowsAnyAsync( + () => wait.WaitAsync(TimeSpan.FromSeconds(5))); + Assert.True(wait.IsCanceled); + } + } + + [Fact] + public async Task WaitForConnectedAsync_CallerCancellation_DoesNotCancelSharedWaiter() + { + await using var client = CreateConnection(12345, isActive: true); + using var caller = new CancellationTokenSource(); + var shared = client.WaitForConnectedAsync(); + var canceled = client.WaitForConnectedAsync(caller.Token); + caller.Cancel(); + await Assert.ThrowsAnyAsync(() => canceled); + Assert.False(shared.IsCompleted); + await client.DisposeAsync(); + await Assert.ThrowsAnyAsync( + () => shared.WaitAsync(TimeSpan.FromSeconds(5))); + Assert.True(shared.IsCanceled); + } + + [Fact] + public async Task WaitForConnectedAsync_LifetimeCancellation_CancelsExistingAndFutureWaiters() + { + await using var client = CreateConnection(GetFreePort(), isActive: true); + using var lifetime = new CancellationTokenSource(); + client.Start(lifetime.Token); + var wait = client.WaitForConnectedAsync(); + lifetime.Cancel(); + await Assert.ThrowsAnyAsync( + () => wait.WaitAsync(TimeSpan.FromSeconds(5))); + Assert.True(wait.IsCanceled); + Assert.True(client.WaitForConnectedAsync().IsCanceled); + } + [Fact] public async Task Start_SecondCall_ThrowsInvalidOperation() {