diff --git a/CHANGELOG.md b/CHANGELOG.md index 93b08ac..1098a1e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,6 +21,9 @@ ### 新增 +- 增加真实 TCP Active/Passive 重复会话抖动故障注入测试；固定三轮断开与 + 重选并复用相同 System Bytes，验证待处理事务确定中断、旧回复与强制迟到 + T3 回调不污染替换会话，且新事务可成功完成。 - 增加默认关闭的独立 T8 transport fault 观测流，保留 StreamFrame TCP Session ID、actor 状态与最多 8 KiB 的未完成帧前缀快照。 - T8 观测队列容量显式可配且满时丢弃最旧项，不阻塞协议 actor；默认路径 diff --git a/docs/DEVELOPMENT-STATUS.md b/docs/DEVELOPMENT-STATUS.md index 2bc8abd..3a218b7 100644 --- a/docs/DEVELOPMENT-STATUS.md +++ b/docs/DEVELOPMENT-STATUS.md @@ -29,9 +29,13 @@ GitHub Issue/Actions： 诊断/控制元数据 Trace、解码失败样本和 T8 前缀快照均已落地。 - T8 故障观测默认关闭、队列有界，最多保存 StreamFrame 提供的 8 KiB 前缀；它不是完整 TCP 片段，也不能进入消息重放路径。 -- SecsFrame 当前没有开放 Issue。PR #40 合并后的 `ci` 与 `codeql` 均通过。 +- 真实 TCP Active/Passive 事务测试固定重复三次断开和重选，复用相同 + System Bytes 并强制投递旧 T3 回调，验证旧回复、旧计时器和待处理事务 + 不会污染替换会话。 +- SecsFrame 当前没有开放 Issue。当前 `main` 提交 `8fc4599`（PR #41）的 + `ci` 与 `codeql` 均通过。 - 最近一次完整本地验证中，核心测试在 `net48`、`net8.0`、`net10.0` 各 - 通过 273 项；官方 secs4net 互操作测试在 `net8.0`、`net10.0` 各通过 + 通过 274 项；官方 secs4net 互操作测试在 `net8.0`、`net10.0` 各通过 2 项。 所有新提交仍须重新运行： @@ -45,8 +49,10 @@ dotnet test SecsFrame.slnx -c Release --no-build ### StreamFrame 错误归属 -[StreamFrame #56](https://github.com/CSJ608/StreamFrame/issues/56) 已得到维护者 -接受，正在实现。评审建议在 StreamFrame `2.6.0` 提供： +[StreamFrame #56](https://github.com/CSJ608/StreamFrame/issues/56) 已由 +[PR #57](https://github.com/CSJ608/StreamFrame/pull/57) 实现并关闭，但截至 +2026-08-29，最新 GitHub Release 与 nuget.org 官方包仍为 `2.5.0`。上游 +计划在 StreamFrame `2.6.0` 提供： - `FrameError` 的原始 transport Session ID； - 错误发生时的实际观测字节数； @@ -79,11 +85,8 @@ dotnet test SecsFrame.slnx -c Release --no-build ## 等待期间的下一优先级 -不依赖 StreamFrame #56 或新增标准语义的下一项，是补充 `1.0` 发布门槛中的 -确定性故障注入证据。 - -建议的最小垂直切片是“重复会话抖动下的事务恢复”：在真实 TCP -Active/Passive 端点的同一生命周期中，固定次数重复执行连接、Select、 +不依赖 StreamFrame 新包或新增标准语义的确定性故障注入基线已经补齐： +真实 TCP Active/Passive 端点在同一生命周期中固定重复三次连接、Select、 Primary/Secondary、断开和重选，并验证： - 每一代会话只完成本代事务，旧回复和旧计时器不能污染新会话； @@ -92,7 +95,9 @@ Primary/Secondary、断开和重选，并验证： - 测试时间有严格上限，可稳定进入现有普通 CI。 这个切片只增加故障注入测试和必要的测试支撑，不改变协议默认值或公共 API。 -长时间 soak、随机网络故障和独立定时工作流应在该确定性基线稳定后另行评估。 +下一可执行项是评估不进入普通 CI 的独立定时工作流：先固定运行上限、随机 +种子、故障重现信息和产物保留边界，再增加长时间 soak 与随机网络故障。 +该工作流不得引入新的协议默认值，也不能替代确定性回归和完整普通 CI。 ## 状态更新规则 diff --git a/docs/HSMS-DATA-TRANSACTIONS.md b/docs/HSMS-DATA-TRANSACTIONS.md index 4b7a21e..ad498ab 100644 --- a/docs/HSMS-DATA-TRANSACTIONS.md +++ b/docs/HSMS-DATA-TRANSACTIONS.md @@ -80,7 +80,10 @@ transport session 的信封不能在新连接上回复。 嵌套 Item 往返、并发独立计时器、重复 System Bytes 避让、复合键匹配、 Reject、Deselect、断线、取消、迟到消息、畸形 Secondary 和一次性回复。 Active/Passive 双端真实 TCP 回环另行覆盖 Select 后的 Primary/Secondary -完整往返。 +完整往返，以及同一端点生命周期内固定三次断开、重选和事务恢复。抖动测试 +在新旧会话复用相同 System Bytes，显式投递已释放的旧 T3 回调并尝试旧 +会话回复，验证待处理事务确定中断且替换会话仍可完成本代事务。同步只使用 +状态事件、计时器武装信号和事务任务，并由统一取消令牌严格封顶。 这些测试是自主构造的工程向量，不是 SEMI 一致性测试。 diff --git a/test/SecsFrame.Tests/HsmsSessionStateMachineTests.DataTransactionTcp.cs b/test/SecsFrame.Tests/HsmsSessionStateMachineTests.DataTransactionTcp.cs index dd4710b..0a173c8 100644 --- a/test/SecsFrame.Tests/HsmsSessionStateMachineTests.DataTransactionTcp.cs +++ b/test/SecsFrame.Tests/HsmsSessionStateMachineTests.DataTransactionTcp.cs @@ -1,10 +1,13 @@ using System.Net; +using System.Threading.Channels; using StreamFrame; namespace SecsFrame.Tests; public sealed partial class HsmsSessionStateMachineTests { + private const uint ChurnSystemBytes = 0x01020304; + [Fact] public async Task Active_and_passive_transactions_round_trip_over_real_tcp() { @@ -54,9 +57,152 @@ await passive.ReplyAsync( Assert.Equal(SecsItem.Boolean(true), secondary.Message.RootItem); } + [Fact] + public async Task Repeated_tcp_session_churn_isolates_transactions_and_stale_timers() + { + const int ChurnCount = 3; + var t3Timers = new SignaledT3TimerFactory(); + var (passiveManager, activeManager) = CreateTcpTransactions( + t3Timers, + new FixedSystemBytesProvider(ChurnSystemBytes)); + await using var passive = passiveManager; + await using var active = activeManager; + using var cancellation = new CancellationTokenSource(TimeSpan.FromSeconds(20)); + await using var passiveEvents = + passive.GetEventsAsync(cancellation.Token).GetAsyncEnumerator(); + await using var activeEvents = + active.GetEventsAsync(cancellation.Token).GetAsyncEnumerator(); + + passive.Start(cancellation.Token); + active.Start(cancellation.Token); + var state = new TcpSessionState( + await NextSelectedTransactionEventAsync(passiveEvents) + .ConfigureAwait(true), + await NextSelectedTransactionEventAsync(activeEvents) + .ConfigureAwait(true)); + + for (var cycle = 0; cycle < ChurnCount; cycle++) + { + var interrupted = await InterruptTcpTransactionAsync( + passive, + active, + passiveEvents, + activeEvents, + t3Timers, + state, + cycle, + cancellation.Token) + .ConfigureAwait(true); + await AssertReplacementTcpTransactionAsync( + passive, + active, + passiveEvents, + t3Timers, + interrupted, + cycle, + cancellation.Token) + .ConfigureAwait(true); + state = interrupted.NextState; + } + + Assert.Equal(HsmsSessionState.Selected, passive.State); + Assert.Equal(HsmsSessionState.Selected, active.State); + } + + private static async Task InterruptTcpTransactionAsync( + HsmsDataTransactionManager passive, + HsmsDataTransactionManager active, + IAsyncEnumerator passiveEvents, + IAsyncEnumerator activeEvents, + SignaledT3TimerFactory t3Timers, + TcpSessionState state, + int cycle, + CancellationToken cancellationToken) + { + var send = active.SendAsync( + 10, + new SecsMessage(6, 11, true, SecsItem.U1((byte)cycle)), + cancellationToken); + var primaryEvent = await NextDataMessageEventAsync(passiveEvents) + .ConfigureAwait(true); + var primary = primaryEvent.DataMessage!; + var timer = await t3Timers.WaitForArmedTimerAsync(cancellationToken) + .ConfigureAwait(true); + + Assert.Equal(state.Passive.SessionId, primaryEvent.SessionId); + Assert.Equal(ChurnSystemBytes, primary.DataMessage.SystemBytes); + Assert.True(timer.IsArmed); + + await active.SeparateAsync(cancellationToken).ConfigureAwait(true); + var error = await Assert.ThrowsAsync( + () => send).ConfigureAwait(true); + Assert.Equal(HsmsSessionState.Disconnected, error.State); + Assert.True(timer.IsDisposed); + + var nextState = new TcpSessionState( + await NextSelectedTransactionEventAsync(passiveEvents) + .ConfigureAwait(true), + await NextSelectedTransactionEventAsync(activeEvents) + .ConfigureAwait(true)); + Assert.True(nextState.Passive.SessionId.Value > state.Passive.SessionId.Value); + Assert.True(nextState.Active.SessionId.Value > state.Active.SessionId.Value); + + return new InterruptedTcpTransaction(primary, timer, nextState); + } + + private static async Task AssertReplacementTcpTransactionAsync( + HsmsDataTransactionManager passive, + HsmsDataTransactionManager active, + IAsyncEnumerator passiveEvents, + SignaledT3TimerFactory t3Timers, + InterruptedTcpTransaction interrupted, + int cycle, + CancellationToken cancellationToken) + { + var staleReply = await Assert.ThrowsAsync< + HsmsDataTransactionInterruptedException>( + () => passive.ReplyAsync( + interrupted.Primary, + new SecsMessage(6, 12), + cancellationToken)) + .ConfigureAwait(true); + Assert.Equal(HsmsSessionState.Selected, staleReply.State); + + var send = active.SendAsync( + 10, + new SecsMessage(6, 11, true, SecsItem.U1((byte)(cycle + 1))), + cancellationToken); + var primaryEvent = await NextDataMessageEventAsync(passiveEvents) + .ConfigureAwait(true); + var primary = primaryEvent.DataMessage!; + var timer = await t3Timers.WaitForArmedTimerAsync(cancellationToken) + .ConfigureAwait(true); + Assert.Equal(interrupted.NextState.Passive.SessionId, primaryEvent.SessionId); + Assert.Equal(ChurnSystemBytes, primary.DataMessage.SystemBytes); + + interrupted.Timer.ForceFire(); + await passive.ReplyAsync( + primary, + new SecsMessage( + 6, + 12, + rootItem: SecsItem.U1((byte)(cycle + 1))), + cancellationToken).ConfigureAwait(true); + var secondary = await send.ConfigureAwait(true); + + Assert.NotNull(secondary); + Assert.Equal(ChurnSystemBytes, secondary.SystemBytes); + Assert.Equal( + SecsItem.U1((byte)(cycle + 1)), + secondary.Message.RootItem); + Assert.True(timer.IsDisposed); + } + private static ( HsmsDataTransactionManager Passive, - HsmsDataTransactionManager Active) CreateTcpTransactions() + HsmsDataTransactionManager Active) CreateTcpTransactions( + IHsmsTransportTimerFactory? activeT3TimerFactory = null, + IHsmsSystemBytesProvider? activeSystemBytesProvider = null) { var port = GetFreePort(); var options = new StreamConnectionOptions @@ -88,7 +234,9 @@ private static ( new HsmsDataTransactionOptions(T3)), new HsmsDataTransactionManager( activeSession, - new HsmsDataTransactionOptions(T3))); + new HsmsDataTransactionOptions(T3), + timerFactory: activeT3TimerFactory, + systemBytesProvider: activeSystemBytesProvider)); } private static Task NextSelectedTransactionEventAsync( @@ -98,4 +246,91 @@ private static Task NextSelectedTransactionEventAsync( item => item.Kind == HsmsDataTransactionEventKind.StateChanged && item.State == HsmsSessionState.Selected); + + private static Task NextDataMessageEventAsync( + IAsyncEnumerator events) + => NextMatchingTransactionEventAsync( + events, + item => item.Kind == + HsmsDataTransactionEventKind.DataMessageReceived); + + private readonly record struct TcpSessionState( + HsmsDataTransactionEvent Passive, + HsmsDataTransactionEvent Active); + + private readonly record struct InterruptedTcpTransaction( + HsmsIncomingDataMessage Primary, + SignaledT3Timer Timer, + TcpSessionState NextState); + + private sealed class SignaledT3TimerFactory : IHsmsTransportTimerFactory + { + private readonly Channel _armedTimers = + Channel.CreateUnbounded(new UnboundedChannelOptions + { + SingleReader = true, + SingleWriter = true, + AllowSynchronousContinuations = false, + }); + + public IHsmsTransportTimer Create(Action callback) + => new SignaledT3Timer( + callback, + timer => _armedTimers.Writer.TryWrite(timer)); + + public ValueTask WaitForArmedTimerAsync( + CancellationToken cancellationToken) + => _armedTimers.Reader.ReadAsync(cancellationToken); + } + + private sealed class SignaledT3Timer : IHsmsTransportTimer + { + private readonly Action _callback; + private readonly Action _onArmed; +#if NET9_0_OR_GREATER + private readonly Lock _sync = new(); +#else + private readonly object _sync = new(); +#endif + private TimeSpan _dueTime = Timeout.InfiniteTimeSpan; + + public SignaledT3Timer( + Action callback, + Action onArmed) + { + _callback = callback; + _onArmed = onArmed; + } + + public bool IsArmed + { + get + { + lock (_sync) + return _dueTime != Timeout.InfiniteTimeSpan; + } + } + + public bool IsDisposed { get; private set; } + + public void Change(TimeSpan dueTime) + { + lock (_sync) + _dueTime = dueTime; + + _onArmed(this); + } + + public void ForceFire() + => _callback(); + + public void Dispose() + { + lock (_sync) + { + _dueTime = Timeout.InfiniteTimeSpan; + IsDisposed = true; + } + } + } }