diff --git a/CHANGELOG.md b/CHANGELOG.md index e367399..96b3e82 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,6 +21,13 @@ ### 新增 +- 增加默认关闭的完整 HSMS 控制消息元数据观测流，覆盖内部 Select、 + Linktest、Deselect、Reject 与 Separate 的收发，而不改变现有业务事件流。 +- 观测只保留方向、actor 观察状态与原十字节头；入站在协议处理前发布， + 出站仅在整帧写出确认后发布，并可直接创建控制 Trace 记录。 +- 增加状态机顺序、未知 SType/Reject、消费者替换和 Active/Passive 真实 + TCP Select/Linktest 测试；原始字节、Body、transport generation 与时间戳 + 仍不进入观测模型。 - 增加独立 SecsFrame-ControlTrace/1 信封，对公共未认领控制 事件提供不含 Body 的十字节头元数据导出与严格读取。 - 控制记录保留方向、观察状态和原始头字段，允许未知非零 SType 往返；增加 diff --git a/README.md b/README.md index 6d0a0ef..56044bf 100644 --- a/README.md +++ b/README.md @@ -44,8 +44,9 @@ Equipment 应用发送策略的 Collection Event、报警目录查询、单报 SecsFrame.Trace 在该文本层上提供版本化的数据消息 Trace 信封、 严格的 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 0c15249..1d5d31b 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -19,8 +19,9 @@ Trace 的时序重放必须显式启用，只在允许发送记录之间等待 发送 API 创建新事务；默认重放不引入时间等待。 结构化诊断使用独立只读快照信封：只复制公共诊断的稳定标量，明确排除原 异常和未解码帧，也不能进入重放路径。 -控制消息使用另一个只读元数据信封，只从调用方已消费的公共未认领控制事件 -复制十字节头字段；它不增加内部控制面观察者，也不代表完整握手抓包。 +控制消息使用另一个只读元数据信封。核心连接可显式启用独立控制观测流， +从会话 actor 复制全部收发控制消息的十字节头字段；Trace 仍由调用方消费 +该流并创建记录，不自行订阅连接。既有公共未认领控制事件工厂继续可用。 ## 核心原则 @@ -128,6 +129,11 @@ Selected 转发数据消息；提前到达的数据、可识别的不支持类 无法由会话层认领的 Reject 作为 ControlMessageReceived 事件向上转发，供数据事务 actor 关联，不会被静默丢弃。 +可选控制观测通道也由同一 actor 写入：非 Data Message 的入站头在协议 +验证和状态转换前记录，出站头只在绑定当前会话的整帧发送成功后记录。 +通道默认不创建，记录不含 Body、原始字节、transport Session generation +或时间戳；它是诊断元数据边界，不参与协议决策。 + ### 会话绑定的数据事务 内部 HsmsDataTransactionManager 独占消费会话事件，并把 @@ -152,7 +158,8 @@ HSMS 头 Session ID 与 System Bytes。没有 W-Bit 的出站消息在写出后 HsmsConnection 是运输、会话和数据事务 actor 之上的薄公共 组合层。它要求显式网络、协议 Session ID 和 T3/T5/T6/T7/T8 配置，公开 动态消息发送、一次性入站回复、控制命令、Selected 等待和单消费者事件 -流。公共层不暴露 transport Session generation、StreamFrame +流，并可显式启用与业务事件独立的单消费者控制元数据观测流。公共层不暴露 +transport Session generation、StreamFrame StreamConnectionOptions 或 StreamFrame 的会话感知接口。 连接内部独占消费事务事件，并分别更新状态等待信号与公共事件 channel， @@ -172,6 +179,6 @@ HSMS 头 Session ID 与 System Bytes。没有 W-Bit 的出站消息在写出后 公共发送 API，不得直接构造或写出保留旧事务标识的线上帧；可选时序只 控制相邻发送前的等待，不得绕过连接状态、T3 或显式 allowlist。诊断 导出不得隐式包含 Exception 或 HsmsFrame；控制 - 元数据导出不得包含 Body 或隐式订阅连接事件流。 + 元数据导出不得包含 Body 或隐式订阅任何连接流。 - 核心包不得依赖 GEM、SML/SMN、依赖注入容器或具体日志实现。 - 状态机计时使用可替换的时间抽象，测试不得依赖真实长时间等待。 diff --git a/docs/HSMS-CONNECTION.md b/docs/HSMS-CONNECTION.md index 7ff309a..828f013 100644 --- a/docs/HSMS-CONNECTION.md +++ b/docs/HSMS-CONNECTION.md @@ -14,7 +14,8 @@ StreamFrame 运输适配、HSMS-SS 会话 actor 和数据事务 actor，调用 - W-Bit Primary 的 T3 关联与 Secondary 返回； - 接收入站 Primary 或未匹配消息，并使用不可伪造令牌回复一次； - Linktest、Deselect 和 Separate 控制命令； -- 状态、数据、未消费控制消息和解码失败的单消费者异步事件流。 +- 状态、数据、未消费控制消息和解码失败的单消费者异步事件流； +- 默认关闭、与业务事件独立的完整控制消息头元数据观测流。 Host/Equipment 是后续业务能力角色，不由 Active/Passive 推断。 需要按运行期 Stream/Function 处理 Primary 时，可在本事件流上组合 @@ -35,6 +36,7 @@ Host/Equipment 是后续业务能力角色，不由 Active/Passive 推断。 | T6 | Select、Linktest、Deselect 控制响应等待 | | T7 | TCP 建立后等待进入 Selected | | T8 | 未完成消息接收进展等待 | +| EnableControlMessageObservation | 是否创建完整控制消息元数据观测流；默认 false | 仓库没有内置或暗示标准默认值。T5 必须能无损转换成 StreamFrame 使用的 整毫秒范围；Passive 监听重试不被解释成 T5。每个取值应依据团队合法获得 @@ -52,7 +54,8 @@ var options = new HsmsConnectionOptions( t5: TimeSpan.FromSeconds(10), t6: TimeSpan.FromSeconds(5), t7: TimeSpan.FromSeconds(10), - t8: TimeSpan.FromSeconds(5)); + t8: TimeSpan.FromSeconds(5), + enableControlMessageObservation: true); await using var connection = new HsmsConnection(options); connection.Start(); @@ -89,6 +92,33 @@ var secondary = await connection.SendAsync( 不包含已被内部 Select、Linktest、Deselect 或数据事务认领的控制消息， 因此快照不是完整控制面抓包。格式与安全边界见 [TRACE.md](TRACE.md)。 +完整控制面诊断必须在构造连接时显式设置 +EnableControlMessageObservation，并由应用持续消费独立流： + +~~~csharp +await foreach (var observation in connection + .GetControlMessageObservationsAsync(cancellationToken)) +{ + var record = SecsTraceControlRecord.Create( + DateTimeOffset.UtcNow, + observation); +} +~~~ + +该流与 GetEventsAsync 各自只有一个活动消费者，取消或释放旧 +枚举器后可重新获取，二者互不争抢。入站控制头在会话 actor 执行协议处理和 +状态转换前发布；出站控制头只在整帧写入成功后发布，因此 Select Request +通常观察为 Selecting，Passive 的成功 Select Response 仍观察为 Connected。 +极快响应可能先于本地发送完成回调进入 actor，跨方向记录顺序和出站状态应 +解释为 actor 实际处理顺序，而不是线上字节的精确时间顺序。 +启用后通道会缓存尚未消费的记录，应用必须持续读取；默认关闭时不创建通道， +调用观测 API 会失败。 + +HsmsControlMessageObservation 只包含本地方向、观察状态和原 +十字节 HsmsMessageHeader。它不含 Body、原始 TCP/帧字节、 +transport Session generation 或时间戳，也不改变未认领 +ControlMessageReceived 业务事件的既有行为。 + 带错误的状态事件和解码失败事件还会提供可选 HsmsDiagnostic，用于按稳定代码、层级、操作与计时器处理， 无需解析异常消息。调用任务抛出的异常可通过 @@ -117,4 +147,5 @@ var secondary = await connection.SendAsync( 真实 TCP 测试使用两个公共连接完成 Select、Linktest、嵌套 Item Primary/Secondary、重复回复拒绝和取消后保持 Selected。它证明当前实现 -各层能够组合工作，不替代 SEMI 一致性认证或第三方设备互操作。 +各层能够组合工作；另有完整控制观测测试核对双方 Select/Linktest 的方向、 +状态和头字段相等。它们不替代 SEMI 一致性认证或第三方设备互操作。 diff --git a/docs/HSMS-SESSION-STATE-MACHINE.md b/docs/HSMS-SESSION-STATE-MACHINE.md index 85b4caa..1f112b1 100644 --- a/docs/HSMS-SESSION-STATE-MACHINE.md +++ b/docs/HSMS-SESSION-STATE-MACHINE.md @@ -20,7 +20,9 @@ - 本地和远端 Separate 都终止当前协议会话； - 只有 Selected 状态的数据消息才向上游事件流转发； - 本地数据发送同样进入 actor 做 Selected 与当前 Session ID 校验，任务 - 在 IHsmsTransport 确认完整写出后完成。 + 在 IHsmsTransport 确认完整写出后完成； +- 可显式启用独立控制消息元数据观测：入站在协议处理前记录，出站在完整 + 写出确认后记录，默认关闭且不影响协议事件。 连接主动/被动模式不表示 Host/Equipment 业务角色。状态机没有 Host/Equipment 分支，后续角色 API 可以独立组合。 @@ -70,6 +72,7 @@ System Bytes。旧会话的帧、发送完成和计时器回调不会推进替 - Selected 前拒绝数据、Selected 后转发数据； - Linktest/Deselect 的写出确认、T6、响应匹配、拒绝和并发互斥； - Reject 黄金向量、四类生成路径、未消费 Reject 转发； +- 完整 Select/Linktest 收发观测顺序和未知 SType 到 Reject 的头字段保留； - 非法控制头、本地及远端 Separate； - Selected 数据发送门控、完整写出确认、会话关闭和替换会话不重放。 diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index d9ee4c2..daf9aa6 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -112,12 +112,14 @@ E30-0526 材料。下一可执行优先级已转入阶段 3；SML 调试读写 - [x] 显式启用、可缩放且单次等待封顶的 Trace 重放时序。 - [x] 不包含异常与原始帧的受限结构化诊断 Trace 导出和严格读取。 - [x] 公共未认领控制消息的受限头元数据 Trace 导出和严格读取。 -- 完整控制面观测和故障原始样本等互操作诊断扩展。 +- [x] 默认关闭、不含 Body/原始字节的完整控制面头元数据观测。 +- 故障原始样本等互操作诊断扩展。 当前 Trace 基线不接管连接事件流，也不把原始标识写回线上。时序重放默认 关闭，并继续经过显式 allowlist 与公共发送 API。诊断快照始终排除 -Error 与 Frame。控制元数据只覆盖应用事件循环已经 -取得的未认领控制帧，不声称观察内部握手。后续完整控制观测与故障原始样本 +Error 与 Frame。完整控制元数据观测必须在连接 +选项中显式启用，覆盖内部握手但只复制方向、actor 状态和十字节头，不含 +Body、原始线上字节、transport generation 或时间戳。后续故障原始样本 扩展必须先定义原始字节的数据分级、默认不导出策略和可验证脱敏边界。 E173 SMN 与 E172 SEDD 仍需先核对授权标准及允许分发的 Schema/表示形式。 diff --git a/docs/TRACE.md b/docs/TRACE.md index 1d6c28c..7850588 100644 --- a/docs/TRACE.md +++ b/docs/TRACE.md @@ -9,7 +9,8 @@ - 只选择显式允许的本地发送记录，并通过正常公共发送 API 串行重放； - 可选按缩放且封顶的源时间间隔执行受控时序重放； - 不包含异常和原始帧的结构化 HSMS 诊断快照导出与严格读取； -- 公共未认领控制事件的十字节头元数据导出与严格读取。 +- 显式启用的完整控制观测或公共未认领控制事件的十字节头元数据导出与 + 严格读取。 当前切片不是原始网络抓包器，不记录 TCP 字节、transport Session generation、 控制帧的线上原始字节或异常对象，也不会接管 HsmsConnection 的 @@ -90,17 +91,18 @@ Session ID、System Bytes、远端状态字节和被拒绝的 SType。枚举只 与 SecsFrame-Trace/1 消息信封相互独立，不能交给重放器。协议 标识和状态字节仍是运维元数据；存储和共享文件时仍需执行访问控制与保留策略。 -## 未认领控制消息元数据 +## 控制消息元数据 -应用在自己的连接事件循环中可以对 -ControlMessageReceived 创建只包含头字段的快照： +推荐在连接选项中显式启用完整控制观测，并由应用持续消费独立流后创建只 +包含头字段的快照： ~~~csharp -if (connectionEvent.Kind == HsmsConnectionEventKind.ControlMessageReceived) +await foreach (var observation in connection + .GetControlMessageObservationsAsync(cancellationToken)) { - var record = SecsTraceControlRecord.CreateReceived( + var record = SecsTraceControlRecord.Create( DateTimeOffset.UtcNow, - connectionEvent); + observation); var text = new SecsTraceControlCodec().Encode(new[] { record }); } ~~~ @@ -117,10 +119,17 @@ Session ID、Header Byte 2、Header Byte 3、PType、SType 和 System Bytes。 字节字段使用固定宽度大写十六进制。SType 以原始字节保存：0x00 因表示 Data Message 而拒绝，未知非零值允许严格往返，便于保留互操作证据。 -控制帧按核心模型不能携带 Body，快照类型也没有 Body 字段。当前公共事件只 -转发未被 Select、Linktest、Deselect 或数据事务认领的控制帧，主要用于未匹配 -Reject；这个工具不会订阅内部状态机，也不是完整控制面抓包。控制信封不能 -进入消息重放器，头字段仍应按运维元数据执行访问控制与保留策略。 +控制帧按核心模型不能携带 Body，快照类型也没有 Body 字段。完整观测流默认 +关闭，启用后覆盖内部 Select、Linktest、Deselect、Reject 与 Separate： +入站记录发生在协议处理前，出站只记录成功写出的帧。Trace 工具不会自行 +订阅该单消费者流，也不提供原始线上字节或 transport Session generation。 +极快响应与本地发送完成回调可能以不同顺序进入 actor，因此记录顺序不是 +PCAP 级线上时序；需要线速证据时不能从本元数据流推断字节先后。 + +既有 SecsTraceControlRecord.CreateReceived 仍可从业务事件流的 +ControlMessageReceived 创建快照，但该事件只转发未被会话或 +数据事务认领的控制帧，主要用于未匹配 Reject。两条创建路径生成同一信封； +控制信封不能进入消息重放器，头字段仍应按运维元数据执行访问控制与保留策略。 ## 结构化脱敏 @@ -213,5 +222,5 @@ var results = await new SecsTraceReplayer().ReplayWithTimingAsync( 测试还验证重新分配 Session ID/System Bytes 并获得新 Secondary。它证明 当前工具组合边界，不代表设备命令安全性或 SEMI 一致性认证。诊断信封另有 黄金向量、可选字段往返、严格枚举/十六进制、资源限制及异常/帧内容不泄漏 -测试。控制信封另有未匹配 Reject 黄金向量、未知 SType 往返、Data Message -拒绝、严格字段与资源限制测试。 +测试。控制信封另有完整 Select/Linktest 双向观测、未匹配 Reject 黄金向量、 +未知 SType 往返、Data Message 拒绝、严格字段与资源限制测试。 diff --git a/src/SecsFrame.Trace/SecsTraceControlRecord.cs b/src/SecsFrame.Trace/SecsTraceControlRecord.cs index 8660198..c6d2877 100644 --- a/src/SecsFrame.Trace/SecsTraceControlRecord.cs +++ b/src/SecsFrame.Trace/SecsTraceControlRecord.cs @@ -90,6 +90,36 @@ public static SecsTraceControlRecord Create( header.SystemBytes); } + /// Creates a metadata record from a full control-message observation. + public static SecsTraceControlRecord Create( + DateTimeOffset timestamp, + HsmsControlMessageObservation observation) + { + if (observation is null) + throw new ArgumentNullException(nameof(observation)); + + var direction = observation.Direction switch + { + HsmsControlMessageDirection.Sent => SecsTraceDirection.Sent, + HsmsControlMessageDirection.Received => SecsTraceDirection.Received, + _ => throw new ArgumentOutOfRangeException( + nameof(observation), + observation.Direction, + "Unknown control-message direction."), + }; + var header = observation.Header; + return new SecsTraceControlRecord( + timestamp, + direction, + observation.State, + header.SessionId, + header.HeaderByte2, + header.HeaderByte3, + header.PresentationType, + (byte)header.MessageType, + header.SystemBytes); + } + /// /// Creates a received metadata record from an unclaimed public control event. /// diff --git a/src/SecsFrame/Hsms/Connection/HsmsConnection.cs b/src/SecsFrame/Hsms/Connection/HsmsConnection.cs index ac4509d..e5872d2 100644 --- a/src/SecsFrame/Hsms/Connection/HsmsConnection.cs +++ b/src/SecsFrame/Hsms/Connection/HsmsConnection.cs @@ -24,6 +24,7 @@ public sealed class HsmsConnection : IAsyncDisposable private Task? _eventPump; private int _state = (int)HsmsSessionState.Disconnected; private int _eventReaderClaimed; + private int _controlMessageObservationReaderClaimed; private int _started; private int _disposed; @@ -87,6 +88,33 @@ public IAsyncEnumerable GetEventsAsync( return ReadEventsAsync(cancellationToken); } + /// + /// Gets the single-consumer stream of sent and received HSMS control-message + /// header metadata. + /// + /// Cancels only this observation-stream read. + /// + /// This stream is independent from and is available + /// only when + /// is enabled. Sent observations are emitted after the entire control frame is + /// written. Received observations are emitted before protocol handling. The + /// stream contains no frame body, transport-session generation, raw bytes, or + /// timestamp. Only one observation stream can be active at a time. + /// + public IAsyncEnumerable + GetControlMessageObservationsAsync( + CancellationToken cancellationToken = default) + { + ThrowIfNotRunning(); + if (!Options.EnableControlMessageObservation) + { + throw new InvalidOperationException( + "HSMS control-message observation is not enabled in the connection options."); + } + + return ReadControlMessageObservationsAsync(cancellationToken); + } + /// Waits until this connection next reaches Selected state. /// Cancels only this wait operation. public Task WaitUntilSelectedAsync( @@ -203,7 +231,8 @@ private static HsmsDataTransactionManager CreateTransactions( new HsmsSessionOptions( options.ConnectionMode, options.T6, - options.T7)); + options.T7, + options.EnableControlMessageObservation)); return new HsmsDataTransactionManager( session, new HsmsDataTransactionOptions(options.T3)); @@ -305,6 +334,35 @@ private async IAsyncEnumerable ReadEventsAsync( } } + private async IAsyncEnumerable + ReadControlMessageObservationsAsync( + [EnumeratorCancellation] CancellationToken cancellationToken) + { + if (Interlocked.Exchange( + ref _controlMessageObservationReaderClaimed, + 1) != 0) + { + throw new InvalidOperationException( + "The HSMS control-message observation stream already has a consumer."); + } + + try + { + await foreach (var observation in _transactions + .GetControlMessageObservationsAsync(cancellationToken) + .ConfigureAwait(false)) + { + yield return observation; + } + } + finally + { + Interlocked.Exchange( + ref _controlMessageObservationReaderClaimed, + 0); + } + } + private static HsmsConnectionEvent MapEvent( HsmsDataTransactionEvent transactionEvent) => transactionEvent.Kind switch diff --git a/src/SecsFrame/Hsms/Connection/HsmsConnectionOptions.cs b/src/SecsFrame/Hsms/Connection/HsmsConnectionOptions.cs index d25d608..16ba5f6 100644 --- a/src/SecsFrame/Hsms/Connection/HsmsConnectionOptions.cs +++ b/src/SecsFrame/Hsms/Connection/HsmsConnectionOptions.cs @@ -18,6 +18,9 @@ public sealed class HsmsConnectionOptions /// The control transaction reply timeout. /// The selection timeout after TCP connection. /// The positive whole-millisecond incomplete-message receive timeout. + /// + /// Enables the separate restricted-metadata control-message observation stream. + /// public HsmsConnectionOptions( IPAddress ipAddress, int port, @@ -27,7 +30,8 @@ public HsmsConnectionOptions( TimeSpan t5, TimeSpan t6, TimeSpan t7, - TimeSpan t8) + TimeSpan t8, + bool enableControlMessageObservation = false) { IpAddress = ipAddress ?? throw new ArgumentNullException(nameof(ipAddress)); if (port is < 1 or > ushort.MaxValue) @@ -69,6 +73,7 @@ public HsmsConnectionOptions( T6 = t6; T7 = t7; T8 = t8; + EnableControlMessageObservation = enableControlMessageObservation; } /// Gets the remote or local bind IP address. @@ -98,6 +103,11 @@ public HsmsConnectionOptions( /// Gets the incomplete-message receive timeout. public TimeSpan T8 { get; } + /// + /// Gets whether the separate control-message metadata stream is enabled. + /// + public bool EnableControlMessageObservation { get; } + private static void ValidatePositive( TimeSpan value, string parameterName, diff --git a/src/SecsFrame/Hsms/Connection/HsmsControlMessageDirection.cs b/src/SecsFrame/Hsms/Connection/HsmsControlMessageDirection.cs new file mode 100644 index 0000000..79c85b7 --- /dev/null +++ b/src/SecsFrame/Hsms/Connection/HsmsControlMessageDirection.cs @@ -0,0 +1,11 @@ +namespace SecsFrame; + +/// Identifies the local direction of an observed HSMS control message. +public enum HsmsControlMessageDirection +{ + /// The control message was written to the peer. + Sent, + + /// The control message was received from the peer. + Received, +} diff --git a/src/SecsFrame/Hsms/Connection/HsmsControlMessageObservation.cs b/src/SecsFrame/Hsms/Connection/HsmsControlMessageObservation.cs new file mode 100644 index 0000000..f08d058 --- /dev/null +++ b/src/SecsFrame/Hsms/Connection/HsmsControlMessageObservation.cs @@ -0,0 +1,54 @@ +namespace SecsFrame; + +/// +/// Describes the restricted metadata observed for one HSMS control message. +/// +/// +/// The observation contains only the ten-byte HSMS header. It does not expose +/// frame bodies, transport-session generations, raw network bytes, or a clock. +/// +public sealed class HsmsControlMessageObservation +{ + internal HsmsControlMessageObservation( + HsmsControlMessageDirection direction, + HsmsSessionState state, + HsmsMessageHeader header) + { + if (direction != HsmsControlMessageDirection.Sent && + direction != HsmsControlMessageDirection.Received) + { + throw new ArgumentOutOfRangeException( + nameof(direction), + direction, + "Unknown control-message direction."); + } + + if (!Enum.IsDefined(typeof(HsmsSessionState), state)) + { + throw new ArgumentOutOfRangeException( + nameof(state), + state, + "Unknown HSMS session state."); + } + + if (header.IsDataMessage) + { + throw new ArgumentException( + "A control-message observation requires a nonzero SType.", + nameof(header)); + } + + Direction = direction; + State = state; + Header = header; + } + + /// Gets the local message direction. + public HsmsControlMessageDirection Direction { get; } + + /// Gets the session state observed with the control message. + public HsmsSessionState State { get; } + + /// Gets the original ten-byte HSMS header fields. + public HsmsMessageHeader Header { get; } +} diff --git a/src/SecsFrame/Hsms/Session/HsmsSessionOptions.cs b/src/SecsFrame/Hsms/Session/HsmsSessionOptions.cs index c71a5f0..6b1dfd4 100644 --- a/src/SecsFrame/Hsms/Session/HsmsSessionOptions.cs +++ b/src/SecsFrame/Hsms/Session/HsmsSessionOptions.cs @@ -5,7 +5,8 @@ internal sealed class HsmsSessionOptions public HsmsSessionOptions( HsmsConnectionMode connectionMode, TimeSpan controlReplyTimeout, - TimeSpan selectionTimeout) + TimeSpan selectionTimeout, + bool enableControlMessageObservation = false) { if (connectionMode != HsmsConnectionMode.Active && connectionMode != HsmsConnectionMode.Passive) @@ -35,6 +36,7 @@ public HsmsSessionOptions( ConnectionMode = connectionMode; ControlReplyTimeout = controlReplyTimeout; SelectionTimeout = selectionTimeout; + EnableControlMessageObservation = enableControlMessageObservation; } public HsmsConnectionMode ConnectionMode { get; } @@ -42,4 +44,6 @@ public HsmsSessionOptions( public TimeSpan ControlReplyTimeout { get; } public TimeSpan SelectionTimeout { get; } + + public bool EnableControlMessageObservation { get; } } diff --git a/src/SecsFrame/Hsms/Session/HsmsSessionStateMachine.cs b/src/SecsFrame/Hsms/Session/HsmsSessionStateMachine.cs index 9a1155d..d561a2c 100644 --- a/src/SecsFrame/Hsms/Session/HsmsSessionStateMachine.cs +++ b/src/SecsFrame/Hsms/Session/HsmsSessionStateMachine.cs @@ -12,6 +12,7 @@ internal sealed class HsmsSessionStateMachine : IAsyncDisposable private readonly IHsmsSystemBytesProvider _systemBytesProvider; private readonly Channel _inputs; private readonly Channel _events; + private readonly Channel? _controlMessageObservations; private readonly HashSet _pendingDataSends = new(); private CancellationTokenSource? _lifetime; private Task? _transportPump; @@ -50,6 +51,17 @@ public HsmsSessionStateMachine( SingleWriter = true, AllowSynchronousContinuations = false, }); + if (options.EnableControlMessageObservation) + { + _controlMessageObservations = + Channel.CreateUnbounded( + new UnboundedChannelOptions + { + SingleReader = true, + SingleWriter = true, + AllowSynchronousContinuations = false, + }); + } } public HsmsSessionState State @@ -87,6 +99,21 @@ public async IAsyncEnumerable GetEventsAsync( } } + public async IAsyncEnumerable + GetControlMessageObservationsAsync( + [EnumeratorCancellation] CancellationToken cancellationToken) + { + var observations = _controlMessageObservations ?? + throw new InvalidOperationException( + "HSMS control-message observation is not enabled."); + var reader = observations.Reader; + while (await reader.WaitToReadAsync(cancellationToken).ConfigureAwait(false)) + { + while (reader.TryRead(out var observation)) + yield return observation; + } + } + public Task SeparateAsync(CancellationToken cancellationToken = default) => RequestCommandAsync(MachineInputKind.SeparateRequested, cancellationToken); @@ -182,6 +209,7 @@ public async ValueTask DisposeAsync() _pendingControlCommand?.Completion.TrySetCanceled(); _separateCompletion?.TrySetCanceled(); _events.Writer.TryComplete(); + _controlMessageObservations?.Writer.TryComplete(); _lifetime?.Dispose(); } @@ -234,6 +262,7 @@ private async Task ProcessInputsAsync() CancelT6(); CancelT7(); _events.Writer.TryComplete(); + _controlMessageObservations?.Writer.TryComplete(); } } @@ -370,6 +399,13 @@ private void ProcessSessionClosed(Exception? error) private void ProcessFrame(HsmsFrame frame) { var header = frame.Header; + if (!header.IsDataMessage) + { + ObserveControlMessage( + HsmsControlMessageDirection.Received, + header); + } + if (header.PresentationType != 0) { SendReject(header, HsmsRejectReason.UnsupportedPresentationType); @@ -644,6 +680,10 @@ private void ProcessSendCompleted(SendOperation operation) if (operation.SessionId != _sessionId) return; + ObserveControlMessage( + HsmsControlMessageDirection.Sent, + CreateControlHeader(operation)); + switch (operation.Purpose) { case SendPurpose.SelectRequest: @@ -861,6 +901,14 @@ private void ProcessDataSendFailed( } private void StartControlSend(SendOperation operation) + { + var frame = new HsmsFrame(CreateControlHeader(operation)); + + _ = SendControlAsync(operation, frame); + } + + private static HsmsMessageHeader CreateControlHeader( + SendOperation operation) { var messageType = operation.Purpose switch { @@ -893,12 +941,15 @@ private void StartControlSend(SendOperation operation) messageType, operation.SystemBytes, operation.HeaderByte3); - var frame = new HsmsFrame( - header); - - _ = SendControlAsync(operation, frame); + return header; } + private void ObserveControlMessage( + HsmsControlMessageDirection direction, + HsmsMessageHeader header) + => _controlMessageObservations?.Writer.TryWrite( + new HsmsControlMessageObservation(direction, State, header)); + private async Task SendControlAsync(SendOperation operation, HsmsFrame frame) { try diff --git a/src/SecsFrame/Hsms/Transactions/HsmsDataTransactionManager.cs b/src/SecsFrame/Hsms/Transactions/HsmsDataTransactionManager.cs index 10a3412..5b68725 100644 --- a/src/SecsFrame/Hsms/Transactions/HsmsDataTransactionManager.cs +++ b/src/SecsFrame/Hsms/Transactions/HsmsDataTransactionManager.cs @@ -82,6 +82,10 @@ public async IAsyncEnumerable GetEventsAsync( } } + public IAsyncEnumerable + GetControlMessageObservationsAsync(CancellationToken cancellationToken) + => _session.GetControlMessageObservationsAsync(cancellationToken); + public Task SendAsync( ushort sessionId, SecsMessage primary, diff --git a/test/SecsFrame.Tests/HsmsConnectionOptionsTests.cs b/test/SecsFrame.Tests/HsmsConnectionOptionsTests.cs index 839f0a3..113ca1b 100644 --- a/test/SecsFrame.Tests/HsmsConnectionOptionsTests.cs +++ b/test/SecsFrame.Tests/HsmsConnectionOptionsTests.cs @@ -28,6 +28,25 @@ public void Explicit_values_are_preserved() Assert.Equal(TimeSpan.FromSeconds(5), options.T6); Assert.Equal(TimeSpan.FromSeconds(10), options.T7); Assert.Equal(TimeSpan.FromSeconds(5), options.T8); + Assert.False(options.EnableControlMessageObservation); + } + + [Fact] + public void Control_message_observation_can_be_explicitly_enabled() + { + var options = new HsmsConnectionOptions( + IPAddress.Loopback, + 5000, + HsmsConnectionMode.Active, + 10, + TimeSpan.FromSeconds(45), + TimeSpan.FromSeconds(10), + TimeSpan.FromSeconds(5), + TimeSpan.FromSeconds(10), + TimeSpan.FromSeconds(5), + enableControlMessageObservation: true); + + Assert.True(options.EnableControlMessageObservation); } [Fact] diff --git a/test/SecsFrame.Tests/HsmsConnectionTests.cs b/test/SecsFrame.Tests/HsmsConnectionTests.cs index caad07e..96a4d80 100644 --- a/test/SecsFrame.Tests/HsmsConnectionTests.cs +++ b/test/SecsFrame.Tests/HsmsConnectionTests.cs @@ -37,6 +37,94 @@ await AssertRoundTripAsync( cancellation.Token).ConfigureAwait(true); } + [Fact] + public async Task Enabled_control_observation_captures_select_and_linktest_over_real_tcp() + { + var port = GetFreePort(); + await using var passive = new HsmsConnection( + CreateOptions( + port, + HsmsConnectionMode.Passive, + enableControlMessageObservation: true)); + await using var active = new HsmsConnection( + CreateOptions( + port, + HsmsConnectionMode.Active, + enableControlMessageObservation: true)); + using var cancellation = new CancellationTokenSource( + TimeSpan.FromSeconds(15)); + + passive.Start(); + active.Start(); + await using var passiveObservations = passive + .GetControlMessageObservationsAsync(cancellation.Token) + .GetAsyncEnumerator(); + await using var activeObservations = active + .GetControlMessageObservationsAsync(cancellation.Token) + .GetAsyncEnumerator(); + await Task.WhenAll( + passive.WaitUntilSelectedAsync(cancellation.Token), + active.WaitUntilSelectedAsync(cancellation.Token)) + .ConfigureAwait(true); + + await AssertSelectObservationsAsync( + passiveObservations, + activeObservations).ConfigureAwait(true); + + await active.LinktestAsync(cancellation.Token).ConfigureAwait(true); + await AssertLinktestObservationsAsync( + passiveObservations, + activeObservations).ConfigureAwait(true); + } + + private static async Task AssertSelectObservationsAsync( + IAsyncEnumerator passiveObservations, + IAsyncEnumerator activeObservations) + { + var active = await ReadControlExchangeAsync( + activeObservations, + HsmsControlMessageDirection.Sent, + HsmsMessageType.SelectRequest, + HsmsMessageType.SelectResponse).ConfigureAwait(true); + var passive = await ReadControlExchangeAsync( + passiveObservations, + HsmsControlMessageDirection.Received, + HsmsMessageType.SelectRequest, + HsmsMessageType.SelectResponse).ConfigureAwait(true); + + Assert.True( + active.Request.State is + HsmsSessionState.Selecting or HsmsSessionState.Selected); + Assert.Equal(HsmsSessionState.Connected, passive.Request.State); + Assert.Equal(active.Request.Header, passive.Request.Header); + Assert.Equal(HsmsSessionState.Connected, passive.Response.State); + Assert.Equal(HsmsSessionState.Selecting, active.Response.State); + Assert.Equal(passive.Response.Header, active.Response.Header); + } + + private static async Task AssertLinktestObservationsAsync( + IAsyncEnumerator passiveObservations, + IAsyncEnumerator activeObservations) + { + var active = await ReadControlExchangeAsync( + activeObservations, + HsmsControlMessageDirection.Sent, + HsmsMessageType.LinktestRequest, + HsmsMessageType.LinktestResponse).ConfigureAwait(true); + var passive = await ReadControlExchangeAsync( + passiveObservations, + HsmsControlMessageDirection.Received, + HsmsMessageType.LinktestRequest, + HsmsMessageType.LinktestResponse).ConfigureAwait(true); + + Assert.Equal(HsmsSessionState.Selected, active.Request.State); + Assert.Equal(HsmsSessionState.Selected, passive.Request.State); + Assert.Equal(active.Request.Header, passive.Request.Header); + Assert.Equal(HsmsSessionState.Selected, passive.Response.State); + Assert.Equal(HsmsSessionState.Selected, active.Response.State); + Assert.Equal(passive.Response.Header, active.Response.Header); + } + [Fact] public async Task Canceling_open_transaction_does_not_close_public_connection() { @@ -149,6 +237,52 @@ await Assert.ThrowsAnyAsync( () => firstRead).ConfigureAwait(true); } + [Fact] + public async Task Control_observation_is_disabled_by_default() + { + await using var connection = new HsmsConnection( + CreateOptions(GetFreePort(), HsmsConnectionMode.Active)); + connection.Start(); + + Assert.Throws( + () => connection.GetControlMessageObservationsAsync()); + } + + [Fact] + public async Task Control_observation_enforces_one_reader_and_allows_replacement() + { + await using var connection = new HsmsConnection( + CreateOptions( + GetFreePort(), + HsmsConnectionMode.Active, + enableControlMessageObservation: true)); + using var cancellation = new CancellationTokenSource(); + connection.Start(); + await using var first = connection + .GetControlMessageObservationsAsync(cancellation.Token) + .GetAsyncEnumerator(); + var firstRead = first.MoveNextAsync().AsTask(); + var concurrent = connection + .GetControlMessageObservationsAsync() + .GetAsyncEnumerator(); + await using var concurrentScope = concurrent.ConfigureAwait(true); + + await Assert.ThrowsAsync( + () => concurrent.MoveNextAsync().AsTask()).ConfigureAwait(true); + cancellation.Cancel(); + await Assert.ThrowsAnyAsync( + () => firstRead).ConfigureAwait(true); + + using var replacementCancellation = new CancellationTokenSource( + TimeSpan.FromMilliseconds(100)); + var replacement = connection + .GetControlMessageObservationsAsync(replacementCancellation.Token) + .GetAsyncEnumerator(); + await using var replacementScope = replacement.ConfigureAwait(true); + await Assert.ThrowsAnyAsync( + () => replacement.MoveNextAsync().AsTask()).ConfigureAwait(true); + } + [Fact] public async Task Connection_rejects_commands_before_start_and_duplicate_start() { @@ -175,7 +309,8 @@ public async Task Connection_rejects_commands_before_start_and_duplicate_start() private static HsmsConnectionOptions CreateOptions( int port, - HsmsConnectionMode connectionMode) + HsmsConnectionMode connectionMode, + bool enableControlMessageObservation = false) => new( IPAddress.Loopback, port, @@ -185,7 +320,8 @@ private static HsmsConnectionOptions CreateOptions( t5: TimeSpan.FromMilliseconds(10), t6: TimeSpan.FromSeconds(5), t7: TimeSpan.FromSeconds(10), - t8: TimeSpan.FromSeconds(5)); + t8: TimeSpan.FromSeconds(5), + enableControlMessageObservation: enableControlMessageObservation); private static async Task AssertSelectedEventAsync( IAsyncEnumerator passiveEvents, @@ -280,4 +416,29 @@ private static async Task NextMatchingAsync( Assert.Fail("The expected public HSMS connection event was not received."); return null!; } + + private static async Task<( + HsmsControlMessageObservation Request, + HsmsControlMessageObservation Response)> ReadControlExchangeAsync( + IAsyncEnumerator observations, + HsmsControlMessageDirection requestDirection, + HsmsMessageType requestType, + HsmsMessageType responseType) + { + Assert.True(await observations.MoveNextAsync().ConfigureAwait(true)); + var first = observations.Current; + Assert.True(await observations.MoveNextAsync().ConfigureAwait(true)); + var second = observations.Current; + var values = new[] { first, second }; + var request = Assert.Single( + values, + item => item.Direction == requestDirection && + item.Header.MessageType == requestType); + var response = Assert.Single( + values, + item => item.Direction != requestDirection && + item.Header.MessageType == responseType); + + return (request, response); + } } diff --git a/test/SecsFrame.Tests/HsmsSessionStateMachineTests.ControlObservation.cs b/test/SecsFrame.Tests/HsmsSessionStateMachineTests.ControlObservation.cs new file mode 100644 index 0000000..b6727e6 --- /dev/null +++ b/test/SecsFrame.Tests/HsmsSessionStateMachineTests.ControlObservation.cs @@ -0,0 +1,131 @@ +namespace SecsFrame.Tests; + +public sealed partial class HsmsSessionStateMachineTests +{ + [Fact] + public async Task Enabled_observation_captures_select_and_linktest_around_state_transitions() + { + var transport = new FakeHsmsTransport(); + await using var machine = CreateMachine( + transport, + HsmsConnectionMode.Active, + new ManualTimerFactory(), + new FixedSystemBytesProvider(0x01020304), + enableControlMessageObservation: true); + using var cancellation = new CancellationTokenSource(TimeSpan.FromSeconds(5)); + await using var observations = machine + .GetControlMessageObservationsAsync(cancellation.Token) + .GetAsyncEnumerator(); + machine.Start(cancellation.Token); + + await AssertActiveSelectObservationsAsync( + transport, + machine, + observations).ConfigureAwait(true); + + var linktest = machine.LinktestAsync(cancellation.Token); + await WaitUntilAsync(() => transport.SendCount == 2).ConfigureAwait(true); + transport.CompleteSend(1); + var linktestSent = await NextControlObservationAsync( + observations, + item => item.Direction == HsmsControlMessageDirection.Sent) + .ConfigureAwait(true); + + Assert.Equal(HsmsSessionState.Selected, linktestSent.State); + Assert.Equal(HsmsMessageType.LinktestRequest, linktestSent.Header.MessageType); + + transport.Receive( + new HsmsFrame( + HsmsMessageHeader.CreateControl( + HsmsMessageType.LinktestResponse, + linktestSent.Header.SystemBytes))); + var linktestReceived = await NextControlObservationAsync( + observations, + item => item.Direction == HsmsControlMessageDirection.Received) + .ConfigureAwait(true); + + Assert.Equal(HsmsSessionState.Selected, linktestReceived.State); + Assert.Equal(HsmsMessageType.LinktestResponse, linktestReceived.Header.MessageType); + await linktest.ConfigureAwait(true); + } + + private static async Task AssertActiveSelectObservationsAsync( + FakeHsmsTransport transport, + HsmsSessionStateMachine machine, + IAsyncEnumerator observations) + { + transport.Open(new HsmsTransportSessionId(1)); + await WaitUntilAsync(() => transport.SendCount == 1).ConfigureAwait(true); + transport.CompleteSend(0); + var selectSent = await NextControlObservationAsync( + observations, + item => item.Direction == HsmsControlMessageDirection.Sent) + .ConfigureAwait(true); + + Assert.Equal(HsmsSessionState.Selecting, selectSent.State); + Assert.Equal(HsmsMessageType.SelectRequest, selectSent.Header.MessageType); + Assert.Equal(0x01020304U, selectSent.Header.SystemBytes); + + transport.Receive( + new HsmsFrame( + HsmsMessageHeader.CreateControl( + HsmsMessageType.SelectResponse, + 0x01020304))); + var selectReceived = await NextControlObservationAsync( + observations, + item => item.Direction == HsmsControlMessageDirection.Received) + .ConfigureAwait(true); + + Assert.Equal(HsmsSessionState.Selecting, selectReceived.State); + Assert.Equal(HsmsMessageType.SelectResponse, selectReceived.Header.MessageType); + await WaitUntilAsync(() => machine.State == HsmsSessionState.Selected) + .ConfigureAwait(true); + } + + [Fact] + public async Task Enabled_observation_preserves_unknown_control_header_and_generated_reject() + { + var transport = new FakeHsmsTransport(); + await using var machine = CreateMachine( + transport, + HsmsConnectionMode.Passive, + new ManualTimerFactory(), + enableControlMessageObservation: true); + using var cancellation = new CancellationTokenSource(TimeSpan.FromSeconds(5)); + await using var observations = machine + .GetControlMessageObservationsAsync(cancellation.Token) + .GetAsyncEnumerator(); + machine.Start(cancellation.Token); + await SelectPassiveAsync(transport, machine).ConfigureAwait(true); + var unknown = new HsmsMessageHeader + { + SessionId = ushort.MaxValue, + HeaderByte2 = 0, + HeaderByte3 = 0, + PresentationType = 0, + MessageType = (HsmsMessageType)0x7F, + SystemBytes = 0x10203040, + }; + + transport.Receive(new HsmsFrame(unknown)); + var received = await NextControlObservationAsync( + observations, + item => (byte)item.Header.MessageType == 0x7F) + .ConfigureAwait(true); + await WaitUntilAsync(() => transport.SendCount == 2).ConfigureAwait(true); + transport.CompleteSend(1); + var sent = await NextControlObservationAsync( + observations, + item => item.Direction == HsmsControlMessageDirection.Sent && + item.Header.MessageType == HsmsMessageType.RejectRequest) + .ConfigureAwait(true); + + Assert.Equal(HsmsControlMessageDirection.Received, received.Direction); + Assert.Equal(HsmsSessionState.Selected, received.State); + Assert.Equal(unknown, received.Header); + Assert.Equal(HsmsSessionState.Selected, sent.State); + Assert.Equal((byte)0x7F, sent.Header.HeaderByte2); + Assert.Equal((byte)HsmsRejectReason.UnsupportedSessionType, sent.Header.HeaderByte3); + Assert.Equal(unknown.SystemBytes, sent.Header.SystemBytes); + } +} diff --git a/test/SecsFrame.Tests/HsmsSessionStateMachineTests.cs b/test/SecsFrame.Tests/HsmsSessionStateMachineTests.cs index 6836d30..fb9ffc7 100644 --- a/test/SecsFrame.Tests/HsmsSessionStateMachineTests.cs +++ b/test/SecsFrame.Tests/HsmsSessionStateMachineTests.cs @@ -553,10 +553,15 @@ private static HsmsSessionStateMachine CreateMachine( FakeHsmsTransport transport, HsmsConnectionMode mode, ManualTimerFactory timers, - IHsmsSystemBytesProvider? systemBytesProvider = null) + IHsmsSystemBytesProvider? systemBytesProvider = null, + bool enableControlMessageObservation = false) => new( transport, - new HsmsSessionOptions(mode, T6, T7), + new HsmsSessionOptions( + mode, + T6, + T7, + enableControlMessageObservation), timers, systemBytesProvider); @@ -617,6 +622,21 @@ private static async Task NextMatchingAsync( return default; } + private static async Task + NextControlObservationAsync( + IAsyncEnumerator observations, + Func predicate) + { + while (await observations.MoveNextAsync().ConfigureAwait(true)) + { + if (predicate(observations.Current)) + return observations.Current; + } + + Assert.Fail("The expected HSMS control-message observation was not received."); + return null!; + } + private static async Task WaitUntilAsync( Func predicate, TimeSpan? timeout = null) diff --git a/test/SecsFrame.Tests/SecsTraceControlTests.cs b/test/SecsFrame.Tests/SecsTraceControlTests.cs index 0bebee3..5deb009 100644 --- a/test/SecsFrame.Tests/SecsTraceControlTests.cs +++ b/test/SecsFrame.Tests/SecsTraceControlTests.cs @@ -47,6 +47,31 @@ public void Codec_preserves_unknown_nonzero_stype_and_sent_direction() AssertRecordEqual(record, decoded); } + [Fact] + public void Observation_factory_maps_direction_state_and_original_header() + { + var header = HsmsMessageHeader.CreateControl( + HsmsMessageType.LinktestRequest, + 0x10203040); + var observation = new HsmsControlMessageObservation( + HsmsControlMessageDirection.Sent, + HsmsSessionState.Selected, + header); + + var record = SecsTraceControlRecord.Create(Epoch, observation); + + Assert.Equal(SecsTraceDirection.Sent, record.Direction); + Assert.Equal(HsmsSessionState.Selected, record.State); + Assert.Equal(header.SessionId, record.ProtocolSessionId); + Assert.Equal(header.HeaderByte2, record.HeaderByte2); + Assert.Equal(header.HeaderByte3, record.HeaderByte3); + Assert.Equal(header.PresentationType, record.PresentationType); + Assert.Equal((byte)header.MessageType, record.MessageType); + Assert.Equal(header.SystemBytes, record.SystemBytes); + Assert.Throws( + () => SecsTraceControlRecord.Create(Epoch, null!)); + } + [Fact] public void Factories_reject_data_frames_and_non_control_events() {