diff --git a/CHANGELOG.md b/CHANGELOG.md index e797849..989da62 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,6 +21,10 @@ ### 新增 +- 增加默认关闭的 Trace 时序重放；按 allowlist 筛选后的 Sent 记录时间戳 + 计算间隔，支持显式速度倍率和缩放后单次等待封顶。 +- 增加时序倒退预验证、筛选间隔、缩放/封顶、等待取消和默认零等待测试； + 时序路径仍通过公共发送 API 创建新事务，不复用旧协议标识。 - 增加独立 SecsFrame.Trace 程序集和版本化 SecsFrame-Trace/1 信封，为已解码数据消息提供确定性导出与 严格读取，并保留 UTC 时间、方向和可选 HSMS 诊断标识。 diff --git a/README.md b/README.md index 926e42f..67505aa 100644 --- a/README.md +++ b/README.md @@ -42,8 +42,8 @@ Equipment 应用发送策略的 Collection Event、报警目录查询、单报 严格读取、全部 Item 类型往返和显式资源边界；它是非规范性调试 profile， 不进入线上通讯路径。 SecsFrame.Trace 在该文本层上提供版本化的数据消息 Trace 信封、 -严格的 Item 路径脱敏和显式 allowlist 重放；它不会接管连接事件消费者， -也不会复用原事务标识。 +严格的 Item 路径脱敏、显式 allowlist 重放，以及默认关闭的缩放/封顶时序 +重放；它不会接管连接事件消费者，也不会复用原事务标识。 Item 用法与边界见 [docs/SECS-II-ITEMS.md](docs/SECS-II-ITEMS.md)，消息集成见 [docs/SECS-MESSAGES.md](docs/SECS-MESSAGES.md)，传输适配边界见 diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 4f26f72..a52e807 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -15,6 +15,8 @@ E172 SEDD 与 E173 SMN 是可选元数据/表示层，不进入核心通讯路 核心包不反向依赖它。 SecsFrame.Trace 依赖 SML 表示层和核心公共发送 API，只处理已经 解码的数据消息；它不订阅内部 transport 或创建第二个公共事件消费者。 +Trace 的时序重放必须显式启用，只在允许发送记录之间等待，并继续通过公共 +发送 API 创建新事务；默认重放不引入时间等待。 ## 核心原则 @@ -163,6 +165,7 @@ HSMS 头 Session ID 与 System Bytes。没有 W-Bit 的出站消息在写出后 - `SecsFrame.Gem` 依赖 `SecsFrame`。 - `SecsFrame.Sml` 依赖 `SecsFrame`，不得依赖 GEM 或线上传输实现。 - `SecsFrame.Trace` 依赖 `SecsFrame.Sml` 与 `SecsFrame`，重放必须回到 - 公共发送 API，不得直接构造或写出保留旧事务标识的线上帧。 + 公共发送 API，不得直接构造或写出保留旧事务标识的线上帧；可选时序只 + 控制相邻发送前的等待，不得绕过连接状态、T3 或显式 allowlist。 - 核心包不得依赖 GEM、SML/SMN、依赖注入容器或具体日志实现。 - 状态机计时使用可替换的时间抽象，测试不得依赖真实长时间等待。 diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index af9d459..5a0b6d5 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -97,7 +97,9 @@ E30-0526 材料。下一可执行优先级已转入阶段 3；SML 调试读写 和全部 SecsItem 类型提供确定性文本写出、严格读取、资源边界与 自主协议向量，不宣称标准 SML 合规。已在该稳定文本表示上增加已解码数据 消息的版本化 Trace 导出、结构化脱敏与显式 allowlist 受控重放；重放通过 -公共发送 API 创建新事务，不复用旧会话或 System Bytes。 +公共发送 API 创建新事务，不复用旧会话或 System Bytes。可选时序重放也已 +完成：只使用允许发送记录的相邻时间戳，支持显式加速和缩放后单次等待封顶， +默认重放仍不等待。 ## 阶段 3：元数据与工具 @@ -105,10 +107,12 @@ E30-0526 材料。下一可执行优先级已转入阶段 3；SML 调试读写 - E173 SMN 日志与消息文档。 - E172 SEDD 设备字典导入、校验与发现。 - [x] 已解码数据消息的 Trace 导出、结构化脱敏和受控重放基线。 -- 控制消息、结构化诊断、故障原始样本和重放时序等互操作诊断扩展。 +- [x] 显式启用、可缩放且单次等待封顶的 Trace 重放时序。 +- 控制消息、结构化诊断和故障原始样本等互操作诊断扩展。 -当前 Trace 基线不接管连接事件流，也不把原始标识写回线上。后续控制与故障 -诊断扩展必须先定义原始字节的数据分级、默认不导出策略和可验证脱敏边界。 +当前 Trace 基线不接管连接事件流，也不把原始标识写回线上。时序重放默认 +关闭，并继续经过显式 allowlist 与公共发送 API。后续控制与故障诊断扩展 +必须先定义原始字节的数据分级、默认不导出策略和可验证脱敏边界。 E173 SMN 与 E172 SEDD 仍需先核对授权标准及允许分发的 Schema/表示形式。 ## 发布门槛 diff --git a/docs/TRACE.md b/docs/TRACE.md index f8bc555..1640791 100644 --- a/docs/TRACE.md +++ b/docs/TRACE.md @@ -6,7 +6,8 @@ - 确定性的 SecsFrame-Trace/1 文本导出和严格读取； - 按 Stream、Function 与 Item List 路径执行的结构化替换脱敏； -- 只选择显式允许的本地发送记录，并通过正常公共发送 API 串行重放。 +- 只选择显式允许的本地发送记录，并通过正常公共发送 API 串行重放； +- 可选按缩放且封顶的源时间间隔执行受控时序重放。 当前切片不是原始网络抓包器，不记录 TCP 字节、transport Session generation、 控制帧或异常对象，也不会接管 HsmsConnection 的单消费者事件流。 @@ -99,16 +100,47 @@ var results = await new SecsTraceReplayer().ReplayAsync( - 只考虑 Sent 记录，Received 永远跳过； - 按源顺序逐条等待正常 SendAsync，W-Bit 继续使用现有 T3； - 忽略原时间、Session ID、System Bytes 和回复令牌，由活连接创建新事务； -- 不等待连接进入 Selected，不自动重试，也不按原时间间隔暂停； +- 默认的 ReplayAsync 不等待连接进入 Selected、不自动重试， + 也不按原时间间隔暂停； - 返回每条已发送记录及新事务获得的可选 Secondary。 allowlist 是业务安全边界。库不会依据 Function 奇偶、消息名称或未知设备状态 推断 Primary/Secondary、幂等性或命令权限；应用只能允许已确认可在目标设备 与当前状态执行的消息。 +### 可选时序 + +只有显式调用 ReplayWithTimingAsync 才会启用时序等待： + +~~~csharp +var timing = new SecsTraceReplayTimingOptions( + speedMultiplier: 2.0, + maxDelay: TimeSpan.FromSeconds(5)); + +var results = await new SecsTraceReplayer().ReplayWithTimingAsync( + records, + connection, + record => record.Message.Stream == 1 && + record.Message.Function == 1, + timing, + cancellationToken); +~~~ + +间隔只在经过方向与 allowlist 筛选的 Sent 记录之间计算；被跳过的 +记录不会拉长等待。时间戳必须非递减，库会在第一次发送前完成全部检查。 +第一条记录立即发送，后续间隔先除以 SpeedMultiplier，再按 +MaxDelay 封顶；相同时间戳不等待。上例把源间隔加速两倍，并把 +缩放后的单次等待限制为五秒。 + +每次等待从上一条消息的发送任务完成后开始，因此该模式保留相邻消息的最小 +节奏，不承诺复现原始绝对发送时刻。等待期间取消会阻止下一条发送，已经完成 +的发送不会回滚。时序模式同样不等待 Selected、不自动重试，也不会绕过 +正常 T3 或连接状态检查。 + ## 验证边界 测试覆盖确定性信封、多记录往返、畸形头与长度、SML 嵌入错误、记录和文本 资源限制、路径漂移、规则重叠、敏感值不进入导出文本、预验证零发送，以及 -真实 TCP 上重新分配 Session ID/System Bytes 并获得新 Secondary。它证明 +时序缩放/封顶、筛选间隔、时间倒退预验证、等待取消和默认零等待。真实 TCP +测试还验证重新分配 Session ID/System Bytes 并获得新 Secondary。它证明 当前工具组合边界，不代表设备命令安全性或 SEMI 一致性认证。 diff --git a/src/SecsFrame.Trace/Properties/AssemblyInfo.cs b/src/SecsFrame.Trace/Properties/AssemblyInfo.cs new file mode 100644 index 0000000..f1f5d55 --- /dev/null +++ b/src/SecsFrame.Trace/Properties/AssemblyInfo.cs @@ -0,0 +1,3 @@ +using System.Runtime.CompilerServices; + +[assembly: InternalsVisibleTo("SecsFrame.Tests")] diff --git a/src/SecsFrame.Trace/SecsTraceReplayTimingOptions.cs b/src/SecsFrame.Trace/SecsTraceReplayTimingOptions.cs new file mode 100644 index 0000000..0f79dbd --- /dev/null +++ b/src/SecsFrame.Trace/SecsTraceReplayTimingOptions.cs @@ -0,0 +1,47 @@ +namespace SecsFrame.Trace; + +/// Configures explicit source-interval pacing for trace replay. +public sealed class SecsTraceReplayTimingOptions +{ + /// The default upper bound for one scaled replay delay. + public static readonly TimeSpan DefaultMaxDelay = TimeSpan.FromMinutes(1); + + /// Creates explicit replay timing options. + /// + /// Replay speed relative to source intervals. Two means half-length delays. + /// + /// Upper bound applied after interval scaling. + public SecsTraceReplayTimingOptions( + double speedMultiplier = 1d, + TimeSpan? maxDelay = null) + { + if (double.IsNaN(speedMultiplier) || double.IsInfinity(speedMultiplier) || speedMultiplier <= 0d) + throw new ArgumentOutOfRangeException(nameof(speedMultiplier), speedMultiplier, "The speed multiplier must be finite and positive."); + + var effectiveMaxDelay = maxDelay ?? DefaultMaxDelay; + if (effectiveMaxDelay <= TimeSpan.Zero) + throw new ArgumentOutOfRangeException(nameof(maxDelay), effectiveMaxDelay, "The maximum delay must be positive."); + + SpeedMultiplier = speedMultiplier; + MaxDelay = effectiveMaxDelay; + } + + /// Gets the replay speed relative to source intervals. + public double SpeedMultiplier { get; } + + /// Gets the upper bound for one scaled replay delay. + public TimeSpan MaxDelay { get; } + + internal TimeSpan Scale(TimeSpan sourceInterval) + { + if (sourceInterval < TimeSpan.Zero) + throw new ArgumentOutOfRangeException(nameof(sourceInterval), sourceInterval, "The source interval cannot be negative."); + + var scaledTicks = sourceInterval.Ticks / SpeedMultiplier; + if (scaledTicks >= MaxDelay.Ticks) + return MaxDelay; + + var roundedTicks = (long)Math.Round(scaledTicks, MidpointRounding.AwayFromZero); + return TimeSpan.FromTicks(roundedTicks); + } +} diff --git a/src/SecsFrame.Trace/SecsTraceReplayer.cs b/src/SecsFrame.Trace/SecsTraceReplayer.cs index fa29005..3bf6c20 100644 --- a/src/SecsFrame.Trace/SecsTraceReplayer.cs +++ b/src/SecsFrame.Trace/SecsTraceReplayer.cs @@ -5,15 +5,25 @@ namespace SecsFrame.Trace; /// Replays explicitly allowed sent records through a normal send API. public sealed class SecsTraceReplayer { + private readonly Func _delayAsync; + /// The default maximum number of source records in one replay request. public const int DefaultMaxRecordCount = SecsTraceCodec.DefaultMaxRecordCount; /// Creates a controlled replayer with an explicit source-record limit. public SecsTraceReplayer(int maxRecordCount = DefaultMaxRecordCount) + : this(maxRecordCount, Task.Delay) + { + } + + internal SecsTraceReplayer( + int maxRecordCount, + Func delayAsync) { if (maxRecordCount <= 0) throw new ArgumentOutOfRangeException(nameof(maxRecordCount), maxRecordCount, "The maximum record count must be positive."); MaxRecordCount = maxRecordCount; + _delayAsync = delayAsync ?? throw new ArgumentNullException(nameof(delayAsync)); } /// Gets the maximum number of source records in one replay request. @@ -32,18 +42,62 @@ public Task> ReplayAsync( if (connection is null) throw new ArgumentNullException(nameof(connection)); - return ReplayAsync(records, connection.SendAsync, isAllowed, cancellationToken); + return ReplayCoreAsync(records, connection.SendAsync, isAllowed, timingOptions: null, cancellationToken); + } + + /// + /// Replays allowed sent records through an existing selected connection and + /// waits scaled source intervals between completed sends. + /// + public Task> ReplayWithTimingAsync( + IEnumerable records, + HsmsConnection connection, + Func isAllowed, + SecsTraceReplayTimingOptions timingOptions, + CancellationToken cancellationToken = default) + { + if (connection is null) + throw new ArgumentNullException(nameof(connection)); + if (timingOptions is null) + throw new ArgumentNullException(nameof(timingOptions)); + + return ReplayCoreAsync(records, connection.SendAsync, isAllowed, timingOptions, cancellationToken); } /// /// Replays allowed sent records through a caller-supplied normal send API. /// This overload can target a role endpoint or a test boundary. /// - public async Task> ReplayAsync( + public Task> ReplayAsync( IEnumerable records, Func> sendAsync, Func isAllowed, CancellationToken cancellationToken = default) + => ReplayCoreAsync(records, sendAsync, isAllowed, timingOptions: null, cancellationToken); + + /// + /// Replays allowed sent records through a caller-supplied normal send API + /// and waits scaled source intervals between completed sends. + /// + public Task> ReplayWithTimingAsync( + IEnumerable records, + Func> sendAsync, + Func isAllowed, + SecsTraceReplayTimingOptions timingOptions, + CancellationToken cancellationToken = default) + { + if (timingOptions is null) + throw new ArgumentNullException(nameof(timingOptions)); + + return ReplayCoreAsync(records, sendAsync, isAllowed, timingOptions, cancellationToken); + } + + private async Task> ReplayCoreAsync( + IEnumerable records, + Func> sendAsync, + Func isAllowed, + SecsTraceReplayTimingOptions? timingOptions, + CancellationToken cancellationToken) { if (records is null) throw new ArgumentNullException(nameof(records)); @@ -65,10 +119,23 @@ public async Task> ReplayAsync( selected.Add(record); } + var delays = timingOptions is null + ? null + : CalculateDelays(selected, timingOptions); var results = new List(selected.Count); - foreach (var record in selected) + for (var index = 0; index < selected.Count; index++) { cancellationToken.ThrowIfCancellationRequested(); + if (delays is not null && index != 0 && delays[index - 1] > TimeSpan.Zero) + { + var delay = _delayAsync(delays[index - 1], cancellationToken); + if (delay is null) + throw new InvalidOperationException("The replay delay delegate returned a null task."); + await delay.ConfigureAwait(false); + cancellationToken.ThrowIfCancellationRequested(); + } + + var record = selected[index]; var operation = sendAsync(record.Message, cancellationToken); if (operation is null) throw new InvalidOperationException("The replay send delegate returned a null task."); @@ -78,4 +145,19 @@ public async Task> ReplayAsync( return new ReadOnlyCollection(results); } + + private static TimeSpan[] CalculateDelays( + IReadOnlyList records, + SecsTraceReplayTimingOptions timingOptions) + { + var delays = new TimeSpan[Math.Max(0, records.Count - 1)]; + for (var index = 1; index < records.Count; index++) + { + var sourceInterval = records[index].Timestamp - records[index - 1].Timestamp; + if (sourceInterval < TimeSpan.Zero) + throw new InvalidOperationException("Allowed sent trace records must use nondecreasing timestamps for timed replay."); + delays[index - 1] = timingOptions.Scale(sourceInterval); + } + return delays; + } } diff --git a/test/SecsFrame.Tests/SecsTraceTests.cs b/test/SecsFrame.Tests/SecsTraceTests.cs index 47ba50b..7155d0d 100644 --- a/test/SecsFrame.Tests/SecsTraceTests.cs +++ b/test/SecsFrame.Tests/SecsTraceTests.cs @@ -234,6 +234,146 @@ await Assert.ThrowsAnyAsync(() => replayer.ReplayAsy Assert.Throws(() => new SecsTraceReplayer(0)); } + [Fact] + public async Task Timed_replay_scales_caps_and_filters_source_intervals() + { + var delays = new List(); + var sent = new List(); + var replayer = new SecsTraceReplayer( + SecsTraceReplayer.DefaultMaxRecordCount, + (delay, _) => + { + delays.Add(delay); + return Task.CompletedTask; + }); + var records = new[] + { + SentAt(TimeSpan.Zero, stream: 1, function: 1), + SentAt(TimeSpan.FromSeconds(100), stream: 2, function: 2), + new SecsTraceRecord(Epoch.AddSeconds(200), SecsTraceDirection.Received, new SecsMessage(1, 9)), + SentAt(TimeSpan.FromSeconds(10), stream: 1, function: 3), + SentAt(TimeSpan.FromSeconds(13), stream: 1, function: 5), + }; + var timing = new SecsTraceReplayTimingOptions( + speedMultiplier: 2, + maxDelay: TimeSpan.FromSeconds(4)); + + var results = await replayer.ReplayWithTimingAsync( + records, + (message, _) => + { + sent.Add(message.Function); + return Task.FromResult(null); + }, + record => record.Message.Stream == 1, + timing).ConfigureAwait(true); + + Assert.Equal(new byte[] { 1, 3, 5 }, sent); + Assert.Equal(new[] { TimeSpan.FromSeconds(4), TimeSpan.FromSeconds(1.5) }, delays); + Assert.Equal(3, results.Count); + } + + [Fact] + public async Task Timed_replay_rejects_timestamp_regression_before_delay_or_send() + { + var delays = 0; + var sends = 0; + var replayer = new SecsTraceReplayer( + SecsTraceReplayer.DefaultMaxRecordCount, + (_, _) => + { + delays++; + return Task.CompletedTask; + }); + var records = new[] + { + SentAt(TimeSpan.FromSeconds(2), stream: 1, function: 1), + SentAt(TimeSpan.FromSeconds(1), stream: 1, function: 3), + }; + + await Assert.ThrowsAsync(() => replayer.ReplayWithTimingAsync( + records, + (_, _) => + { + sends++; + return Task.FromResult(null); + }, + _ => true, + new SecsTraceReplayTimingOptions())).ConfigureAwait(true); + + Assert.Equal(0, delays); + Assert.Equal(0, sends); + } + + [Fact] + public async Task Default_replay_ignores_timestamps_and_never_invokes_delay() + { + var sends = 0; + var replayer = new SecsTraceReplayer( + SecsTraceReplayer.DefaultMaxRecordCount, + (_, _) => throw new InvalidOperationException("Default replay must not delay.")); + var records = new[] + { + SentAt(TimeSpan.FromSeconds(2), stream: 1, function: 1), + SentAt(TimeSpan.FromSeconds(1), stream: 1, function: 3), + }; + + var results = await replayer.ReplayAsync( + records, + (_, _) => + { + sends++; + return Task.FromResult(null); + }, + _ => true).ConfigureAwait(true); + + Assert.Equal(2, sends); + Assert.Equal(2, results.Count); + } + + [Fact] + public async Task Canceling_a_timed_delay_prevents_the_next_send() + { + using var cancellation = new CancellationTokenSource(); + var sends = 0; + var replayer = new SecsTraceReplayer( + SecsTraceReplayer.DefaultMaxRecordCount, + (_, token) => + { + cancellation.Cancel(); + return Task.FromCanceled(token); + }); + var records = new[] + { + SentAt(TimeSpan.Zero, stream: 1, function: 1), + SentAt(TimeSpan.FromSeconds(1), stream: 1, function: 3), + }; + + await Assert.ThrowsAnyAsync(() => replayer.ReplayWithTimingAsync( + records, + (_, _) => + { + sends++; + return Task.FromResult(null); + }, + _ => true, + new SecsTraceReplayTimingOptions(), + cancellation.Token)).ConfigureAwait(true); + + Assert.Equal(1, sends); + } + + [Fact] + public void Timing_options_reject_non_finite_speed_and_non_positive_delay() + { + Assert.Throws(() => new SecsTraceReplayTimingOptions(speedMultiplier: 0)); + Assert.Throws(() => new SecsTraceReplayTimingOptions(speedMultiplier: -1)); + Assert.Throws(() => new SecsTraceReplayTimingOptions(speedMultiplier: double.NaN)); + Assert.Throws(() => new SecsTraceReplayTimingOptions(speedMultiplier: double.PositiveInfinity)); + Assert.Throws(() => new SecsTraceReplayTimingOptions(maxDelay: TimeSpan.Zero)); + Assert.Throws(() => new SecsTraceReplayTimingOptions(maxDelay: TimeSpan.FromTicks(-1))); + } + [Fact] public async Task Replay_uses_a_new_real_tcp_transaction_and_ignores_received_records() { @@ -287,6 +427,14 @@ private static HsmsConnectionOptions CreateOptions(int port, HsmsConnectionMode private static DateTimeOffset Epoch => new(1970, 1, 1, 0, 0, 0, TimeSpan.Zero); + private static SecsTraceRecord SentAt( + TimeSpan offset, + byte stream, + byte function) + => SecsTraceRecord.CreateSent( + Epoch.Add(offset), + new SecsMessage(stream, function)); + private static int GetFreePort() { var listener = new System.Net.Sockets.TcpListener(IPAddress.Loopback, 0);