diff --git a/CHANGELOG.md b/CHANGELOG.md index 8590bf6..e23bcdb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,15 @@ ## [Unreleased] +### 修复 + +- 升级官方 StreamFrame 2.3.1，修复迟到旧会话故障污染活会话状态、旧会话 + 绑定消息跨 Socket 错发，以及会话拆除时发送失败类型不稳定的问题。 +- 串行化适配器的当前会话关闭与 transport 释放，避免已入队的迟到 + Separate 在销毁竞态中调用已经释放的 StreamFrame 生命周期。 +- 增加关闭/释放并发状态测试和真实 TCP 会话替换回归测试，验证排队发送 + 以会话失效结束且不会在新会话重放。 + ### 新增 - 初始化 SecsFrame 仓库、工程规范和 CI。 diff --git a/Directory.Packages.props b/Directory.Packages.props index f48ff7f..e5c7c44 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -9,7 +9,7 @@ - + diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 993c6ab..da1849b 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -53,17 +53,20 @@ HSMS 会话切换后，旧会话的控制请求、数据事务和等待中的应 IHsmsTransport 使用单一事件流按 Session ID 报告会话打开、 帧到达和会话关闭。StreamFrameHsmsTransport 已在内部实现： -- 直接采用 StreamFrame 2.3.0 的单调 TCP Session ID； +- 直接采用 StreamFrame 2.3.1 的单调 TCP Session ID； - 通过会话感知接收流保留帧的原始 Session ID，迟到消费不会被标成新会话； - 会话绑定发送在 StreamFrame 原生 FIFO 中校验 Session ID，旧会话消息 不会重放，整帧写入本机 Socket 后才完成； +- 当前会话关闭/重连与 transport 释放串行化，迟到 actor 输入不会访问已 + 释放的 StreamFrame 生命周期； - Active 连接把显式 T5 映射为固定 StreamFrame 重试间隔，关闭指数退避； - 显式 T8 映射为 StreamFrame 原生未完成帧超时，并把对应 FrameError 转换为带 transport Session ID 的 HSMS 关闭原因。 这些能力仍隔离在 IHsmsTransport 后，不进入公共 API。此前 -跟踪的 StreamFrame #38/#39 已在 2.3.0 完成迁移；SecsFrame 只保留 -协议关闭原因和异常边界转换。详细失效模式与验证证据见 +跟踪的 StreamFrame #38/#39 已在 2.3.0 完成迁移，2.3.1 进一步封闭迟到 +旧故障污染活会话和排队消息跨 Socket 错发的竞态；SecsFrame 只保留协议 +关闭原因和异常边界转换。详细失效模式与验证证据见 [STREAMFRAME-ADAPTER.md](STREAMFRAME-ADAPTER.md)。 ### 单线程会话状态机 diff --git a/docs/COMPATIBILITY.md b/docs/COMPATIBILITY.md index 1eb60e8..f643a16 100644 --- a/docs/COMPATIBILITY.md +++ b/docs/COMPATIBILITY.md @@ -20,7 +20,8 @@ ## 上游能力集成 StreamFrame #38（未完成帧超时）与 #39（会话感知发送确认及消息上下文） -已经随官方 StreamFrame 2.3.0 发布。SecsFrame 固定该 NuGet 版本，并由 +已经随官方 StreamFrame 2.3.0 发布。SecsFrame 固定官方 2.3.1 NuGet 包， +包含迟到旧会话故障隔离、排队发送 Session ID 复核和稳定失效异常修复，并由 内部 StreamFrameHsmsTransport 使用原生会话 API；此前基于 原始字节回调、发送信封和自建 Session ID 的临时实现已经删除。 diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 13c4ff7..bb5173c 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -10,7 +10,8 @@ - [x] 完整 SECS-II Item 类型、嵌套 List 和严格二进制编解码。 - [x] 动态 SecsMessage、可选单根 Item 与 HSMS Data Message Payload 编解码。 -- [x] 迁移到 StreamFrame 2.3.0 原生会话绑定、写出确认和未完成帧超时。 +- [x] 迁移到 StreamFrame 2.3.1 原生会话绑定、写出确认、会话竞态修复和 + 未完成帧超时。 - [x] Active/Passive 共用状态机的 Select/Select Response、Separate、 T6/T7、System Bytes 关联和会话失效处理。 - [x] Linktest、Reject、Deselect 控制消息与对应 T6 分支。 @@ -26,7 +27,8 @@ - [x] Host、Equipment 双角色端点及两种 Active/Passive 拓扑的双向回环。 阶段 1 的工程能力已经形成完整基础链路，StreamFrame #38/#39 的正式包 -迁移与等价测试已经完成。后续切片继续扩展 GEM 通用行为。任何标准默认值 +迁移、2.3.1 会话隔离修复与等价测试已经完成。后续切片继续扩展 GEM 通用 +行为。任何标准默认值 和完整合规结论都依赖团队合法获得的 E37/E37.1 及相关 GEM 标准副本与 一致性测试。 diff --git a/docs/STREAMFRAME-ADAPTER.md b/docs/STREAMFRAME-ADAPTER.md index d16b029..50f29e2 100644 --- a/docs/STREAMFRAME-ADAPTER.md +++ b/docs/STREAMFRAME-ADAPTER.md @@ -2,14 +2,20 @@ ## 上游基线 -SecsFrame 固定依赖官方 [StreamFrame 2.3.0](https://www.nuget.org/packages/StreamFrame/2.3.0)。 -该版本已经发布并解决此前跟踪的两项能力缺口： +SecsFrame 固定依赖官方 [StreamFrame 2.3.1](https://www.nuget.org/packages/StreamFrame/2.3.1)。 +2.3.0 已经解决此前跟踪的两项能力缺口： - [#38](https://github.com/CSJ608/StreamFrame/issues/38) / [PR #41](https://github.com/CSJ608/StreamFrame/pull/41)： 未完成帧超时与 IncompleteFrameTimeout 诊断； - [#39](https://github.com/CSJ608/StreamFrame/issues/39) / [PR #42](https://github.com/CSJ608/StreamFrame/pull/42)： 会话编号、会话绑定发送、写出确认和带会话归属的接收消息。 +2.3.1 在相同公共 API 上修复两个会话隔离缺陷：过期会话的迟到故障不会 +再发布 Retry 或清零活会话编号；发送 worker 会再次校验排队条目的 Session +ID，残留的旧会话绑定消息以 SessionExpiredException 结束且 +不会写入新 Socket。停机、Socket 释放及会话故障竞态也统一收敛到该异常 +边界。详见 [2.3.1 发布说明](https://github.com/CSJ608/StreamFrame/releases/tag/v2.3.1)。 + SecsFrame 不再维护基于原始字节回调的 T8 监视器、发送确认计数器、会话 信封 codec 或自建 TCP Session ID。公共 API 和内部 IHsmsTransport 契约保持不变，上层 HSMS 状态机不直接依赖 @@ -37,6 +43,9 @@ StreamFrame 的 SessionExpiredException 在内部边界转换为 HsmsTransportSessionExpiredException，并保留原异常。协议 主动关闭或 T8 关闭时，适配器按 TCP Session ID 保留首个关闭原因，使 SessionClosed 与尚未完成的发送观察到同一个协议异常。 +当前会话的 Reconnect 与 transport 释放还通过独立生命周期 +边界串行化：释放先阻止新的关闭请求，再等待已经开始的重连结束，避免 actor +清空迟到 Separate 输入时访问已释放的 StreamFrame 生命周期。 ## T5 与 T8 @@ -60,11 +69,11 @@ StreamFrame 只在已有未完成帧时运行该超时：空闲连接和完整 ## 验证边界 适配器单元测试覆盖原生 Session ID、迟到消息、发送完成、并发入队、会话 -失效、显式关闭原因和 T8 诊断映射。真实 TCP 回环测试覆盖分片帧收发与 -实际 T8 到期；完整套件继续覆盖 T3/T6 从发送完成点启动以及旧会话不 -重放。 +失效、显式关闭原因、关闭/释放竞态和 T8 诊断映射。真实 TCP 回环测试覆盖 +分片帧收发、实际 T8 到期，以及会话替换时排队发送的精确失效与新 Socket +内容隔离；完整套件继续覆盖 T3/T6 从发送完成点启动以及旧会话不重放。 -这些测试证明当前工程契约与 StreamFrame 2.3.0 的互操作行为，不构成 +这些测试证明当前工程契约与 StreamFrame 2.3.1 的互操作行为，不构成 SEMI 合规声明。T5/T8 的标准默认值、精确启停边界和异常恢复仍须依据团队 合法获得的 SEMI E37/E37.1 版本及一致性测试核对。仓库不得提交标准正文、 表格或 Schema。 diff --git a/src/SecsFrame/Transport/StreamFrameHsmsTransport.cs b/src/SecsFrame/Transport/StreamFrameHsmsTransport.cs index e419ac7..953d239 100644 --- a/src/SecsFrame/Transport/StreamFrameHsmsTransport.cs +++ b/src/SecsFrame/Transport/StreamFrameHsmsTransport.cs @@ -11,8 +11,10 @@ internal sealed class StreamFrameHsmsTransport : IHsmsTransport private readonly ISessionAwareStreamConnection _connection; private readonly Channel _events; #if NET9_0_OR_GREATER + private readonly Lock _lifecycleGate = new(); private readonly Lock _sessionGate = new(); #else + private readonly object _lifecycleGate = new(); private readonly object _sessionGate = new(); #endif private readonly Dictionary _sessions = new(); @@ -130,6 +132,11 @@ public async ValueTask DisposeAsync() if (Interlocked.Exchange(ref _disposed, 1) != 0) return; + lock (_lifecycleGate) + { + // Wait for an already-started Reconnect before disposing its lifetime state. + } + await _connection.DisposeAsync().ConfigureAwait(false); if (_messagePump is not null) @@ -153,33 +160,37 @@ public bool TryCloseSession( Exception? error = null) { ThrowIfDisposed(); - - Exception? previousReason; - lock (_sessionGate) + lock (_lifecycleGate) { - if (_currentSessionId != sessionId.Value || - _connection.CurrentSessionId != sessionId.Value || - !_sessions.TryGetValue(sessionId.Value, out var context) || - context.IsClosed) + ThrowIfDisposed(); + + Exception? previousReason; + lock (_sessionGate) { - return false; - } + if (_currentSessionId != sessionId.Value || + _connection.CurrentSessionId != sessionId.Value || + !_sessions.TryGetValue(sessionId.Value, out var context) || + context.IsClosed) + { + return false; + } - previousReason = context.CloseReason; - if (error is not null && context.CloseReason is null) - context.CloseReason = error; - } + previousReason = context.CloseReason; + if (error is not null && context.CloseReason is null) + context.CloseReason = error; + } - try - { - _connection.Reconnect(); - return true; - } - catch - { - if (error is not null) - RestoreCloseReason(sessionId, error, previousReason); - throw; + try + { + _connection.Reconnect(); + return true; + } + catch + { + if (error is not null) + RestoreCloseReason(sessionId, error, previousReason); + throw; + } } } diff --git a/test/SecsFrame.Tests/StreamFrameHsmsTransportTests.cs b/test/SecsFrame.Tests/StreamFrameHsmsTransportTests.cs index f980bc9..cc5acf2 100644 --- a/test/SecsFrame.Tests/StreamFrameHsmsTransportTests.cs +++ b/test/SecsFrame.Tests/StreamFrameHsmsTransportTests.cs @@ -1,3 +1,4 @@ +using System.Buffers; using System.Net; using System.Runtime.CompilerServices; using System.Threading.Channels; @@ -255,6 +256,46 @@ public async Task Explicit_close_for_expired_session_does_not_reconnect_current_ await events.DisposeAsync().ConfigureAwait(true); } + [Fact] + public async Task Dispose_waits_for_inflight_session_close() + { + var connection = new FakeStreamConnection { BlockReconnect = true }; + var transport = CreateTransport(connection); + using var cancellation = new CancellationTokenSource(TimeSpan.FromSeconds(5)); + var events = transport.GetEventsAsync(cancellation.Token).GetAsyncEnumerator(); + transport.Start(cancellation.Token); + connection.RaiseState(ConnectionState.Connected); + var opened = await NextAsync(events).ConfigureAwait(true); + + var close = Task.Factory.StartNew( + () => transport.TryCloseSession(opened.SessionId), + CancellationToken.None, + TaskCreationOptions.LongRunning, + TaskScheduler.Default); + var started = await Task.WhenAny( + connection.ReconnectStarted, + Task.Delay(TimeSpan.FromSeconds(2), cancellation.Token)).ConfigureAwait(true); + if (!ReferenceEquals(connection.ReconnectStarted, started)) + { + connection.ReleaseReconnect(); + Assert.Same(connection.ReconnectStarted, started); + } + + var dispose = Task.Factory.StartNew( + () => transport.DisposeAsync().AsTask(), + CancellationToken.None, + TaskCreationOptions.LongRunning, + TaskScheduler.Default).Unwrap(); + await Task.Delay(TimeSpan.FromMilliseconds(50), cancellation.Token).ConfigureAwait(true); + var disposeWaitedForClose = !dispose.IsCompleted; + connection.ReleaseReconnect(); + + Assert.True(await close.ConfigureAwait(true)); + await dispose.ConfigureAwait(true); + Assert.True(disposeWaitedForClose); + await events.DisposeAsync().ConfigureAwait(true); + } + [Fact] public async Task Real_streamframe_connection_sends_and_receives_fragmented_frame() { @@ -356,10 +397,106 @@ await client.GetStream().WriteAsync( await events.DisposeAsync().ConfigureAwait(true); } + [Fact] + public async Task Real_streamframe_replacement_expires_queued_send_without_replay() + { + var port = GetFreePort(); + var codec = new ControlledEncodeCodec(); + await using var transport = CreateControlledTransport(port, codec); + using var cancellation = new CancellationTokenSource(TimeSpan.FromSeconds(20)); + var events = transport.GetEventsAsync(cancellation.Token).GetAsyncEnumerator(); + transport.Start(cancellation.Token); + + using var firstClient = new System.Net.Sockets.TcpClient(); + await firstClient.ConnectAsync(IPAddress.Loopback, port).ConfigureAwait(true); + var firstOpened = await NextAsync(events).ConfigureAwait(true); + await SendAndReadAsync( + transport, + firstOpened.SessionId, + firstClient, + 0x01, + cancellation.Token).ConfigureAwait(true); + codec.BlockNextEncode(); + var inFlight = transport.SendAsync( + firstOpened.SessionId, + CreateFrame(new byte[] { 0x11 }), + cancellation.Token).AsTask(); + var encodeStarted = await Task.WhenAny( + codec.BlockedEncodeStarted, + Task.Delay(TimeSpan.FromSeconds(5), cancellation.Token)).ConfigureAwait(true); + Assert.Same(codec.BlockedEncodeStarted, encodeStarted); + + var queued = transport.SendAsync( + firstOpened.SessionId, + CreateFrame(new byte[] { 0x12 }), + cancellation.Token).AsTask(); + + Assert.True(transport.TryCloseSession(firstOpened.SessionId)); + var firstClosed = await NextAsync(events).ConfigureAwait(true); + var queuedException = await Assert.ThrowsAsync( + async () => await queued.ConfigureAwait(true)).ConfigureAwait(true); + Assert.IsType(queuedException.InnerException); + Assert.NotNull(await Record.ExceptionAsync( + async () => await inFlight.ConfigureAwait(true)).ConfigureAwait(true)); + Assert.Equal(HsmsTransportEventKind.SessionClosed, firstClosed.Kind); + Assert.Equal(firstOpened.SessionId, firstClosed.SessionId); + + firstClient.Dispose(); + using var secondClient = new System.Net.Sockets.TcpClient(); + await secondClient.ConnectAsync(IPAddress.Loopback, port).ConfigureAwait(true); + var secondOpened = await NextAsync(events).ConfigureAwait(true); + Assert.True(secondOpened.SessionId.Value > firstOpened.SessionId.Value); + + var currentBytes = await SendAndReadAsync( + transport, + secondOpened.SessionId, + secondClient, + 0x22, + cancellation.Token).ConfigureAwait(true); + + Assert.Equal(0x22, currentBytes[currentBytes.Length - 1]); + await events.DisposeAsync().ConfigureAwait(true); + } + private static StreamFrameHsmsTransport CreateTransport( FakeStreamConnection connection) => new(connection); + private static StreamFrameHsmsTransport CreateControlledTransport( + int port, + ICodec codec) + => new( + new StreamConnection( + new HsmsFramer(), + codec, + IPAddress.Loopback, + port, + isActive: false, + new StreamConnectionOptions + { + AcceptRetryDelayMs = 10, + ConnectRetryDelayMs = 10, + })); + + private static async Task SendAndReadAsync( + StreamFrameHsmsTransport transport, + HsmsTransportSessionId sessionId, + System.Net.Sockets.TcpClient client, + byte body, + CancellationToken cancellationToken) + { + var send = transport.SendAsync( + sessionId, + CreateFrame(new[] { body }), + cancellationToken).AsTask(); + var bytes = await ReadExactlyAsync( + client.GetStream(), + HsmsFramer.LengthPrefixSize + HsmsMessageHeader.EncodedSize + 1, + cancellationToken).ConfigureAwait(true); + await send.ConfigureAwait(true); + return bytes; + } + private static int GetFreePort() { var listener = new System.Net.Sockets.TcpListener(IPAddress.Loopback, 0); @@ -420,6 +557,39 @@ private static async Task NextAsync( return events.Current; } + private sealed class ControlledEncodeCodec : ICodec + { + private readonly HsmsFrameCodec _inner = new(); + private readonly TaskCompletionSource _blockedEncodeStarted = new( + TaskCreationOptions.RunContinuationsAsynchronously); + private int _blockNextEncode; + + public Task BlockedEncodeStarted => _blockedEncodeStarted.Task; + + public void BlockNextEncode() + => Interlocked.Exchange(ref _blockNextEncode, 1); + + public HsmsFrame Decode( + in ReadOnlySequence frame, + CancellationToken ct = default) + => _inner.Decode(in frame, ct); + + public void Encode( + HsmsFrame message, + IBufferWriter writer, + CancellationToken ct = default) + { + if (Interlocked.Exchange(ref _blockNextEncode, 0) != 0) + { + _blockedEncodeStarted.TrySetResult(true); + ct.WaitHandle.WaitOne(); + ct.ThrowIfCancellationRequested(); + } + + _inner.Encode(message, writer, ct); + } + } + private sealed class FakeStreamConnection : ISessionAwareStreamConnection { private readonly Channel> _messages = @@ -430,7 +600,12 @@ private sealed class FakeStreamConnection : ISessionAwareStreamConnection _sent = new(); + private readonly TaskCompletionSource _continueReconnect = new( + TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _reconnectStarted = new( + TaskCreationOptions.RunContinuationsAsynchronously); private long _sessionCounter; + private int _disposed; public ConnectionState State { get; private set; } = ConnectionState.Connecting; @@ -448,6 +623,10 @@ private sealed class FakeStreamConnection : ISessionAwareStreamConnection _reconnectStarted.Task; + public int SentCount { get @@ -472,9 +651,23 @@ public void Start(CancellationToken ct) public void Reconnect() { ReconnectCount++; + if (BlockReconnect) + { + _reconnectStarted.TrySetResult(true); + _continueReconnect.Task.GetAwaiter().GetResult(); + if (Volatile.Read(ref _disposed) != 0) + { + throw new InvalidOperationException( + "The connection was disposed during Reconnect."); + } + } + RaiseState(ConnectionState.Retry); } + public void ReleaseReconnect() + => _continueReconnect.TrySetResult(true); + public Task WaitForConnectedAsync(CancellationToken ct = default) => Task.CompletedTask; @@ -532,6 +725,7 @@ public async IAsyncEnumerable> GetSessionMessages( public ValueTask DisposeAsync() { + Interlocked.Exchange(ref _disposed, 1); RaiseState(ConnectionState.Disconnected); _messages.Writer.TryComplete(); return default;