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
3 changes: 3 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,9 @@

### 新增

- 增加真实 TCP Active/Passive 重复会话抖动故障注入测试;固定三轮断开与
重选并复用相同 System Bytes,验证待处理事务确定中断、旧回复与强制迟到
T3 回调不污染替换会话,且新事务可成功完成。
- 增加默认关闭的独立 T8 transport fault 观测流,保留 StreamFrame TCP
Session ID、actor 状态与最多 8 KiB 的未完成帧前缀快照。
- T8 观测队列容量显式可配且满时丢弃最旧项,不阻塞协议 actor;默认路径
Expand Down
25 changes: 15 additions & 10 deletions docs/DEVELOPMENT-STATUS.md
Original file line number Diff line number Diff line change
Expand Up @@ -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 项。

所有新提交仍须重新运行:
Expand All @@ -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;
- 错误发生时的实际观测字节数;
Expand Down Expand Up @@ -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、断开和重选,并验证:

- 每一代会话只完成本代事务,旧回复和旧计时器不能污染新会话;
Expand All @@ -92,7 +95,9 @@ Primary/Secondary、断开和重选,并验证:
- 测试时间有严格上限,可稳定进入现有普通 CI。

这个切片只增加故障注入测试和必要的测试支撑,不改变协议默认值或公共 API。
长时间 soak、随机网络故障和独立定时工作流应在该确定性基线稳定后另行评估。
下一可执行项是评估不进入普通 CI 的独立定时工作流:先固定运行上限、随机
种子、故障重现信息和产物保留边界,再增加长时间 soak 与随机网络故障。
该工作流不得引入新的协议默认值,也不能替代确定性回归和完整普通 CI。

## 状态更新规则

Expand Down
5 changes: 4 additions & 1 deletion docs/HSMS-DATA-TRANSACTIONS.md
Original file line number Diff line number Diff line change
Expand Up @@ -80,7 +80,10 @@ transport session 的信封不能在新连接上回复。
嵌套 Item 往返、并发独立计时器、重复 System Bytes 避让、复合键匹配、
Reject、Deselect、断线、取消、迟到消息、畸形 Secondary 和一次性回复。
Active/Passive 双端真实 TCP 回环另行覆盖 Select 后的 Primary/Secondary
完整往返。
完整往返,以及同一端点生命周期内固定三次断开、重选和事务恢复。抖动测试
在新旧会话复用相同 System Bytes,显式投递已释放的旧 T3 回调并尝试旧
会话回复,验证待处理事务确定中断且替换会话仍可完成本代事务。同步只使用
状态事件、计时器武装信号和事务任务,并由统一取消令牌严格封顶。

这些测试是自主构造的工程向量,不是 SEMI 一致性测试。

Expand Down
Original file line number Diff line number Diff line change
@@ -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()
{
Expand Down Expand Up @@ -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<InterruptedTcpTransaction> InterruptTcpTransactionAsync(
HsmsDataTransactionManager passive,
HsmsDataTransactionManager active,
IAsyncEnumerator<HsmsDataTransactionEvent> passiveEvents,
IAsyncEnumerator<HsmsDataTransactionEvent> 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<HsmsDataTransactionInterruptedException>(
() => 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<HsmsDataTransactionEvent> 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
Expand Down Expand Up @@ -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<HsmsDataTransactionEvent> NextSelectedTransactionEventAsync(
Expand All @@ -98,4 +246,91 @@ private static Task<HsmsDataTransactionEvent> NextSelectedTransactionEventAsync(
item => item.Kind ==
HsmsDataTransactionEventKind.StateChanged &&
item.State == HsmsSessionState.Selected);

private static Task<HsmsDataTransactionEvent> NextDataMessageEventAsync(
IAsyncEnumerator<HsmsDataTransactionEvent> 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<SignaledT3Timer> _armedTimers =
Channel.CreateUnbounded<SignaledT3Timer>(new UnboundedChannelOptions
{
SingleReader = true,
SingleWriter = true,
AllowSynchronousContinuations = false,
});

public IHsmsTransportTimer Create(Action callback)
=> new SignaledT3Timer(
callback,
timer => _armedTimers.Writer.TryWrite(timer));

public ValueTask<SignaledT3Timer> WaitForArmedTimerAsync(
CancellationToken cancellationToken)
=> _armedTimers.Reader.ReadAsync(cancellationToken);
}

private sealed class SignaledT3Timer : IHsmsTransportTimer
{
private readonly Action _callback;
private readonly Action<SignaledT3Timer> _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<SignaledT3Timer> 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;
}
}
}
}
Loading