diff --git a/CHANGELOG.md b/CHANGELOG.md index b949dfd..089487b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,7 @@ ## [Unreleased] ### 修复 +- **被取代接受循环的状态杂音与迟到接入语义澄清（[#68](https://github.com/CSJ608/StreamFrame/issues/68) PR-3，#47 世代语义边界定稿）**：用户重连与自动重连竞速产生的被取代"僵尸"接受循环——① 循环头代次复查前置：此前从重试退避中醒来的被取代循环会先发布 Connecting 再检查代次退出，状态序列混入替换者之外的杂音；② 明确并测试锁定迟到 accept 的收编语义：监听器是连接级共享资源，客户端向服务器端口的接入本身合法——被取代循环迟到完成的 accept **照常发布 Connected 并建立会话**（纪元机制与会话隔离收敛，取代者循环经代次检查静默退出）。曾评估"迟到连接当场释放"方案：会重置真实客户端、违背接入契约，已否决（实现一度引入后回归测试即复现客户端被吞）。新增确定性回归：accept 等待期间 Reconnect 取代循环，迟到客户端照常收编、状态序列恰为 Connecting→Retry→Connecting→Connected。 - **终态竞速守门与发布段原子化（[#68](https://github.com/CSJ608/StreamFrame/issues/68)）**：并发 Dispose 与 Reconnect/连接成功竞速时，已通过停机检查但尚未写入 State 的发布被抢占，停机写入 Disconnected 后迟到的 Retry/Connected 会把终态翻转（State 停在 Retry/Connected）；Retry 派发还会对停机回收的 `_lifetimeCts` 空引用。状态发布收敛为发布段（`_publishGate` 锁内原子：停机/过期纪元判定 + 编号纪元簿记 + State 写入）与通知段（锁外用户回调/等待器/派发，回调可安全重入）；停机后到达的非终态发布整体丢弃；传输建立与停机竞速时新 socket 在守门处当场释放（此前泄漏到终结器）。新增迟到发布确定性门控、连接成功与停机竞速、并发关闭/重连终态不变量回归。 - **回调重入会话撕裂（[#68](https://github.com/CSJ608/StreamFrame/issues/68)）**：Connected 回调内同步 `Reconnect()` 时，迟到发布帧的 `StartSession` 以过期编号/纪元"兜底重建"僵尸会话，其拆除会关掉新会话刚建立的 socket（编号复用 + 会话失活，实测可复现）。改为纪元现行性门控：过期帧放弃创建，连接由取代者循环完成；`_socket` 被新连接覆盖时旧值当场释放（直连迁移路径不再泄漏）。新增回调重入（Reconnect/DisposeAsync）收敛性、合法状态转移序列与编号线性化回归。 - **基准工作量对齐（[#70](https://github.com/CSJ608/StreamFrame/issues/70)）**：单向不再回显；直接 TCP 与框架共享负载、定界、Codec、物化校验及完成点，覆盖双 Framer、64B/1KB/64KB 和字节/两种字符串编码，往返逐条等待回显。增加全矩阵连续批次验证；原始结果与环境随仓库保存，撤回不同工作量相减所得的框架税结论，修正指标单位和筛选命令，不修改框架实现。 diff --git a/docs/DESIGN.md b/docs/DESIGN.md index 00c044f..3482d3c 100644 --- a/docs/DESIGN.md +++ b/docs/DESIGN.md @@ -181,7 +181,7 @@ socket --ReceiveAsync--> pipe.Writer --[Pipe]--> FrameDecoder.RunAsync - 停机后的任何调用：`Reconnect()`/`SendInSessionAsync` 等抛 `ObjectDisposedException` 或以会话失效收尾；迟到的内部状态发布被发布段丢弃。 - 迟到的旧会话任务/故障：入口过期纪元检查丢弃，不误杀新会话（1.2.0/2.3.1 既有保证，重入路径由纪元门控补齐）。 -**已知限制（后续 PR 方向）**：① 停机与重连仍**同步**等待会话任务退出（锁外、有界 ≤2s）——公共 API 的同步/异步行为按 #68 约束保持不变；若未来允许 `DisposeAsync` 异步化，可把该等待移入后台收尾。② 接受循环被取代的"僵尸"帧在已取得 `_acceptLock` 且 accept 已完成时不再复查代次（ tolerated path：其发布经纪元机制收敛，socket 覆盖时释放）——统一状态转换入口（每次发布前重验代次/停机）是剩余的结构性收敛项。 +**已知限制/明确的竞速语义（#68 定稿）**：① 停机与重连仍**同步**等待会话任务退出（锁外、有界 ≤2s）——公共 API 的同步/异步行为按 #68 约束保持不变；若未来允许 `DisposeAsync` 异步化，可把该等待移入后台收尾。② 被取代循环建成的连接（被动端的迟到 accept、主动端的迟到 connect）**照常收编**：监听器是连接级共享资源，客户端向服务器端口发起的接入本身合法——被取代循环迟到完成的 accept 发布 Connected 并建立会话，由纪元机制与会话隔离收敛（取代者循环取锁后经代次检查静默退出）。曾评估"迟到连接当场释放"方案：会重置真实客户端、违背接入契约，已否决。代次门控的边界因此是**监听器生命周期（创建/关闭/接受串行化）与 Connecting 发布**（循环头复查前置——被取代循环不再发布 Connecting），而非客户端接入。 ### netstandard2.0 兼容层（2.1.0） diff --git a/src/StreamFrame/Connection/StreamConnection.cs b/src/StreamFrame/Connection/StreamConnection.cs index e7a2a5b..bc02944 100644 --- a/src/StreamFrame/Connection/StreamConnection.cs +++ b/src/StreamFrame/Connection/StreamConnection.cs @@ -237,12 +237,14 @@ private async Task StartAsync(CancellationToken ct, int loopId) { while (!connected && !ct.IsCancellationRequested && !IsDisposed) { + // 代次复查前置（#68 统一发布入口）：被取代的循环连 Connecting 也不发布， + // 状态序列不混入替换者之外的杂音 + if (!IsActive && Volatile.Read(ref _acceptLoopId) != loopId) + return; // 已被更新的接受循环取代：静默退出，不再竞争 accept + CommunicationStateChanging(ConnectionState.Connecting); try { - if (!IsActive && Volatile.Read(ref _acceptLoopId) != loopId) - return; // 已被更新的接受循环取代：静默退出，不再竞争 accept - var socket = IsActive ? await ConnectAsync(ct).ConfigureAwait(false) : await AcceptAsync(ct, loopId).ConfigureAwait(false); if (socket is null) return; // 排队等到锁时已被取代（AcceptAsync 内部的代次检查） @@ -251,6 +253,8 @@ private async Task StartAsync(CancellationToken ct, int loopId) // 发布前的窗口内完成——终态已广播、会话已拆除。此时发布 Connected 会翻转 // 终态并重建会话；把 socket 留进 _socket 则无人负责关闭（Shutdown 已走过 // 关闭点），泄漏到终结器。判定与赋值同锁于 Shutdown 末尾的关闭块：窗口闭合。 + // 注意：被取代循环的迟到 accept 不设代次门——监听器是连接级共享资源， + // 客户端向服务器端口的接入本身合法；收编后由纪元机制收敛（见 DESIGN）。 lock (_sessionGate) { if (ct.IsCancellationRequested || IsDisposed) diff --git a/test/StreamFrame.Tests/ReconnectWedgeTests.cs b/test/StreamFrame.Tests/ReconnectWedgeTests.cs index 994543c..ffc57fa 100644 --- a/test/StreamFrame.Tests/ReconnectWedgeTests.cs +++ b/test/StreamFrame.Tests/ReconnectWedgeTests.cs @@ -89,11 +89,11 @@ public async Task RapidServerInitiatedReconnects_PortStaysAcceptable() for (var round = 0; round < rounds; round++) { - // 对端接入形成会话 + // 对端接入形成会话（预算与下方一致：探测永久不可用，并行负载下偶发慢收敛不应误报） using (var client = new TcpClient()) { await ConnectWithRetryAsync(client, port); - await WaitConnectedAsync(server, timeoutMs: 10_000); + await WaitConnectedAsync(server, timeoutMs: 30_000); } // 等自动重连完成、再次形成会话 diff --git a/test/StreamFrame.Tests/StateOwnershipBaselineTests.cs b/test/StreamFrame.Tests/StateOwnershipBaselineTests.cs index 6d0d841..5932fed 100644 --- a/test/StreamFrame.Tests/StateOwnershipBaselineTests.cs +++ b/test/StreamFrame.Tests/StateOwnershipBaselineTests.cs @@ -217,6 +217,39 @@ public async Task ConnectedCallback_DisposeReentry_TerminatesCleanly() await client.DisposeAsync(); } + [Fact] + public async Task ReconnectWhileAccepting_LateClientStillAdmittedBySupersededLoop() + { + var port = GetFreePort(); + await using var server = Create(port, false); + + var states = new ConcurrentQueue(); + server.ConnectionChanged += (_, state) => states.Enqueue(state); + + server.Start(CancellationToken.None); + await WaitForAsync(() => server.State == ConnectionState.Connecting, "接受循环进入监听等待"); + // Start 同步推进到 accept 等待：循环 A 持有 _acceptLock、监听器已发布 + Assert.IsType(Field(server, "_server")!); + + // 用户 Reconnect 在 accept 等待期间取代循环 A（Retry → StartCore 递增代次、派发循环 B； + // B 排队等锁）。Retry/Connecting 的发布在 Reconnect 返回前同步完成——被取代循环 + // 不再发布 Connecting（代次复查前置），状态序列不混入杂音。 + server.Reconnect(); + Assert.Equal( + new[] { ConnectionState.Connecting, ConnectionState.Retry, ConnectionState.Connecting }, + states.ToArray()); + + // 监听器是连接级共享资源：客户端向服务器端口的接入合法——迟到 accept 照常收编 + // （发布 Connected、建立会话），由纪元机制收敛；取代者循环 B 取锁后经代次检查静默退出 + using var client = new TcpClient(); + await client.ConnectAsync(IPAddress.Loopback, port).WaitAsync(Budget); + await WaitForAsync(() => server.State == ConnectionState.Connected, "迟到客户端被收编并发布 Connected"); + Assert.NotEqual(0, server.CurrentSessionId); + Assert.Equal( + new[] { ConnectionState.Connecting, ConnectionState.Retry, ConnectionState.Connecting, ConnectionState.Connected }, + states.ToArray()); + } + [Theory] [InlineData(0)] [InlineData(1)]