From 3f944eff56716fb00650c94a1ba07602f71cf95a Mon Sep 17 00:00:00 2001 From: Marci Date: Sat, 29 Aug 2026 20:52:14 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=A2=9E=E5=8A=A0=20T8=20=E6=95=85?= =?UTF-8?q?=E9=9A=9C=E5=89=8D=E7=BC=80=E8=A7=82=E6=B5=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- CHANGELOG.md | 8 + README.md | 6 +- docs/ARCHITECTURE.md | 13 +- docs/HSMS-CONNECTION.md | 27 +- docs/HSMS-DIAGNOSTICS.md | 7 + docs/ROADMAP.md | 12 +- docs/STREAMFRAME-ADAPTER.md | 20 +- docs/TRACE.md | 69 ++- .../SecsTraceByteRedactionRange.cs | 2 +- .../SecsTraceFaultSampleDataClassification.cs | 8 +- .../SecsTraceTransportFaultCaptureOptions.cs | 106 +++++ .../SecsTraceTransportFaultCodec.cs | 398 ++++++++++++++++++ .../SecsTraceTransportFaultRecord.cs | 230 ++++++++++ .../Hsms/Connection/HsmsConnection.cs | 65 ++- .../Hsms/Connection/HsmsConnectionOptions.cs | 29 +- .../Hsms/Connection/HsmsTransportFaultKind.cs | 8 + .../HsmsTransportFaultObservation.cs | 52 +++ .../Hsms/Session/HsmsSessionOptions.cs | 19 +- .../Hsms/Session/HsmsSessionStateMachine.cs | 43 ++ .../HsmsDataTransactionManager.cs | 4 + src/SecsFrame/Transport/HsmsTransportEvent.cs | 39 +- .../Transport/HsmsTransportEventKind.cs | 1 + .../Transport/HsmsTransportOptions.cs | 8 +- .../Transport/StreamFrameHsmsTransport.cs | 26 +- .../HsmsConnectionOptionsTests.cs | 36 ++ test/SecsFrame.Tests/HsmsConnectionTests.cs | 90 +++- ...eMachineTests.TransportFaultObservation.cs | 112 +++++ .../HsmsSessionStateMachineTests.cs | 18 +- .../SecsTraceTransportFaultTests.cs | 277 ++++++++++++ .../StreamFrameHsmsTransportTests.cs | 47 ++- 30 files changed, 1735 insertions(+), 45 deletions(-) create mode 100644 src/SecsFrame.Trace/SecsTraceTransportFaultCaptureOptions.cs create mode 100644 src/SecsFrame.Trace/SecsTraceTransportFaultCodec.cs create mode 100644 src/SecsFrame.Trace/SecsTraceTransportFaultRecord.cs create mode 100644 src/SecsFrame/Hsms/Connection/HsmsTransportFaultKind.cs create mode 100644 src/SecsFrame/Hsms/Connection/HsmsTransportFaultObservation.cs create mode 100644 test/SecsFrame.Tests/HsmsSessionStateMachineTests.TransportFaultObservation.cs create mode 100644 test/SecsFrame.Tests/SecsTraceTransportFaultTests.cs diff --git a/CHANGELOG.md b/CHANGELOG.md index 722577d..93b08ac 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,6 +21,14 @@ ### 新增 +- 增加默认关闭的独立 T8 transport fault 观测流，保留 StreamFrame TCP + Session ID、actor 状态与最多 8 KiB 的未完成帧前缀快照。 +- T8 观测队列容量显式可配且满时丢弃最旧项，不阻塞协议 actor；默认路径 + 不创建该通道，旧会话观测不会归入替换会话。 +- 增加独立 SecsFrame-TransportFaultTrace/1 信封，沿用 + MetadataOnly、RedactedPayload 与 RawPayload 分级，payload codec 默认关闭。 +- 增加真实 TCP T8、Session 过滤、队列容量、黄金向量、脱敏复核、防御性 + 复制和资源限制测试；快照不声明完整 TCP 片段且不能进入重放路径。 - 增加独立 SecsFrame-FaultSampleTrace/1 信封，为公共 DataMessageDecodeFailed 事件提供已成帧 HSMS data body 样本。 - 捕获必须显式选择 MetadataOnly、RedactedPayload 或 RawPayload；默认 codec diff --git a/README.md b/README.md index b3433be..5359d98 100644 --- a/README.md +++ b/README.md @@ -47,8 +47,10 @@ Equipment 应用发送策略的 Collection Event、报警目录查询、单报 显式启用的完整控制消息观测流或既有未认领控制事件的十字节头元数据。 解码失败样本另用独立信封，并以元数据为默认导出边界；Body 捕获与 codec 导入导出都必须显式开启，声明的脱敏字节范围会固定清零并严格复核。 -控制观测默认关闭，并与业务事件流独立；该包不会自行接管任何连接消费者， -也不会复用原事务标识。 +T8 互操作诊断另有默认关闭、容量有界的独立观测流；它保留 transport +Session ID 和最多 8 KiB 的未完成帧前缀快照。Trace 默认仍只导出元数据， +快照捕获和 payload codec 必须分别显式开启。控制与 T8 观测都和业务事件流 +独立；该包不会自行接管任何连接消费者，也不会复用原事务标识。 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 cc63be2..3ff6e99 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -25,7 +25,9 @@ Trace 的时序重放必须显式启用，只在允许发送记录之间等待 该流并创建记录，不自行订阅连接。既有公共未认领控制事件工厂继续可用。 已成帧但 SECS-II Body 解码失败的样本使用独立信封：捕获和 codec 分别要求 显式数据边界，默认只允许头与原 Body 长度；payload-bearing 记录不能进入 -重放。T8 和成帧失败片段尚未进入该模型。 +重放。T8 使用另一个独立信封：核心必须先显式启用容量有界的观测流，Trace +再选择元数据、脱敏或原始前缀快照；快照最多 8 KiB 且不声明完整 TCP 片段。 +其他成帧错误仍未进入该模型。 ## 核心原则 @@ -77,7 +79,8 @@ HSMS 会话切换后，旧会话的控制请求、数据事务和等待中的应 释放的 StreamFrame 生命周期； - Active 连接把显式 T5 映射为固定 StreamFrame 重试间隔，关闭指数退避； - 显式 T8 映射为 StreamFrame 原生未完成帧超时，并把对应 FrameError - 转换为带 transport Session ID 的 HSMS 关闭原因。 + 转换为带 transport Session ID 的 HSMS 关闭原因；显式启用时还在同一 + transport 事件序列中复制有界前缀快照。 这些能力仍隔离在 IHsmsTransport 后，不进入公共 API。此前 跟踪的 StreamFrame #38/#39 已在 2.3.0 完成迁移，2.3.1 进一步封闭迟到 @@ -138,6 +141,12 @@ Selected 转发数据消息；提前到达的数据、可识别的不支持类 通道默认不创建，记录不含 Body、原始字节、transport Session generation 或时间戳；它是诊断元数据边界，不参与协议决策。 +可选 T8 观测通道同样由 actor 写入，但使用容量有界、满时丢弃最旧项的 +独立队列，避免未消费的原始快照形成无界内存或阻塞协议处理。状态机只接受 +当前 transport Session ID 的 fault；观测保留最多 8 KiB 前缀，不参与关闭 +决策。StreamFrame 2.5.0 的其他 FrameError 不带 session/epoch，核心不会 +用读取“当前 ID”的方式把它们冒充为严格会话关联事件。 + ### 会话绑定的数据事务 内部 HsmsDataTransactionManager 独占消费会话事件，并把 diff --git a/docs/HSMS-CONNECTION.md b/docs/HSMS-CONNECTION.md index 73d810c..9fb2e3e 100644 --- a/docs/HSMS-CONNECTION.md +++ b/docs/HSMS-CONNECTION.md @@ -15,7 +15,8 @@ StreamFrame 运输适配、HSMS-SS 会话 actor 和数据事务 actor，调用 - 接收入站 Primary 或未匹配消息，并使用不可伪造令牌回复一次； - Linktest、Deselect 和 Separate 控制命令； - 状态、数据、未消费控制消息和解码失败的单消费者异步事件流； -- 默认关闭、与业务事件独立的完整控制消息头元数据观测流。 +- 默认关闭、与业务事件独立的完整控制消息头元数据观测流； +- 默认关闭、容量有界的 T8 transport Session 前缀快照观测流。 Host/Equipment 是后续业务能力角色，不由 Active/Passive 推断。 需要按运行期 Stream/Function 处理 Primary 时，可在本事件流上组合 @@ -37,6 +38,8 @@ Host/Equipment 是后续业务能力角色，不由 Active/Passive 推断。 | T7 | TCP 建立后等待进入 Selected | | T8 | 未完成消息接收进展等待 | | EnableControlMessageObservation | 是否创建完整控制消息元数据观测流；默认 false | +| EnableTransportFaultObservation | 是否创建 T8 前缀快照观测流；默认 false | +| TransportFaultObservationCapacity | T8 队列容量；满时丢弃最旧项，默认 16 | 仓库没有内置或暗示标准默认值。T5 必须能无损转换成 StreamFrame 使用的 整毫秒范围；Passive 监听重试不被解释成 T5。每个取值应依据团队合法获得 @@ -123,6 +126,28 @@ await foreach (var observation in connection transport Session generation 或时间戳，也不改变未认领 ControlMessageReceived 业务事件的既有行为。 +T8 原始诊断必须另外显式设置 +EnableTransportFaultObservation。该流不占用业务事件消费者： + +~~~csharp +await foreach (var fault in connection + .GetTransportFaultObservationsAsync(cancellationToken)) +{ + var record = SecsTraceTransportFaultRecord.Create( + DateTimeOffset.UtcNow, + fault, + SecsTraceTransportFaultCaptureOptions.MetadataOnly()); +} +~~~ + +HsmsTransportFaultObservation 保留 +IncompleteFrameTimeout、StreamFrame TCP Session ID、actor +观察状态和防御性复制的前缀快照。快照最多 8 KiB，可能包含四字节 HSMS +长度前缀；StreamFrame 2.5.0 不报告原缓冲总长度或完整性，因此不得把它 +解释成完整 TCP 片段。队列按配置容量保留最新项，满时丢弃最旧项而不阻塞 +协议 actor；应用启用后应持续消费。该观测在对应会话关闭事件处理前写入， +但两个独立异步流之间不承诺消费者可见的全局顺序。 + 带错误的状态事件和解码失败事件还会提供可选 HsmsDiagnostic，用于按稳定代码、层级、操作与计时器处理， 无需解析异常消息。调用任务抛出的异常可通过 diff --git a/docs/HSMS-DIAGNOSTICS.md b/docs/HSMS-DIAGNOSTICS.md index e71c0e8..46176ac 100644 --- a/docs/HSMS-DIAGNOSTICS.md +++ b/docs/HSMS-DIAGNOSTICS.md @@ -86,6 +86,13 @@ System Bytes 和远端状态字节仍属于运维元数据，导出文件仍应 带 Body 的导入或导出都要再次显式开启。该信封不包含异常、TCP 长度前缀、 分片、transport Session generation 或 T8 未完成片段。 +T8 另有默认关闭的 HsmsTransportFaultObservation 独立流和 +SecsFrame-TransportFaultTrace/1 信封。观测只保留当前 T8 所属 +transport Session ID、actor 状态和 StreamFrame 已复制的最多 8 KiB 前缀； +队列有界且满时丢弃最旧项。Trace 捕获必须再选择仅元数据、声明范围清零或 +原始快照，默认 codec 仍拒绝带快照的记录。它不包含异常、原缓冲总长度、 +完整性标记或分片时序，也不能用于重放。 + ## 标准边界 诊断名称描述 SecsFrame 已实现并测试的工程行为，不新增 SEMI 默认值，也 diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 467bffe..4631fd8 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -114,7 +114,8 @@ E30-0526 材料。下一可执行优先级已转入阶段 3；SML 调试读写 - [x] 公共未认领控制消息的受限头元数据 Trace 导出和严格读取。 - [x] 默认关闭、不含 Body/原始字节的完整控制面头元数据观测。 - [x] 显式分级、默认不导出 Body 的数据消息解码失败样本。 -- T8/成帧失败原始片段等运输层互操作诊断扩展。 +- [x] 默认关闭、有界且显式分级的 T8 前缀快照诊断。 +- 其他成帧失败原始片段等运输层互操作诊断扩展。 当前 Trace 基线不接管连接事件流，也不把原始标识写回线上。时序重放默认 关闭，并继续经过显式 allowlist 与公共发送 API。诊断快照始终排除 @@ -124,8 +125,13 @@ Body、原始线上字节、transport generation 或时间戳。后续故障原 已先覆盖公共事件中忠实保留的完整 data frame：捕获分为 MetadataOnly、 RedactedPayload 与 RawPayload，payload codec 默认关闭，声明脱敏范围必须 全为零且可严格复核。它不包含 TCP 长度前缀或分片，也不能进入重放。 -T8/成帧失败片段仍需先把 StreamFrame 原始错误数据与 transport session -安全关联，并保持默认不捕获、容量限制和跨会话隔离。 +T8 现通过默认关闭的独立流与当前 transport Session ID 关联；队列容量有界、 +满时丢弃最旧项，StreamFrame 提供的最多 8 KiB 前缀再按 MetadataOnly、 +RedactedPayload 或 RawPayload 捕获，payload codec 默认关闭。该前缀可能含 +四字节长度头，但上游不报告原缓冲总长度或完整性，不能解释为完整 TCP 片段， +也不能进入重放。其他 DecodeFailed、IncompleteFrameOverflow 与 +DiscardedByResync 仍需要 StreamFrame 在 FrameError 中提供原始 session/epoch +归属后再扩展，不能用回调时读取当前 Session ID 替代严格关联。 E173 SMN 与 E172 SEDD 仍需先核对授权标准及允许分发的 Schema/表示形式。 ## 发布门槛 diff --git a/docs/STREAMFRAME-ADAPTER.md b/docs/STREAMFRAME-ADAPTER.md index 15efdbf..f946332 100644 --- a/docs/STREAMFRAME-ADAPTER.md +++ b/docs/STREAMFRAME-ADAPTER.md @@ -36,8 +36,8 @@ ID 一样在 Connected 发布点分配，闭合状态已经对外可见但会话 [PR #52](https://github.com/CSJ608/StreamFrame/pull/52)。 SecsFrame 不再维护基于原始字节回调的 T8 监视器、发送确认计数器、会话 -信封 codec 或自建 TCP Session ID。公共 API 和内部 -IHsmsTransport 契约保持不变，上层 HSMS 状态机不直接依赖 +信封 codec 或自建 TCP Session ID。内部 IHsmsTransport 现可在 +显式启用时转发 T8 的会话关联前缀快照；上层 HSMS 状态机仍不直接依赖 StreamFrame 类型。 ## 会话边界 @@ -83,13 +83,23 @@ StreamFrame 只在已有未完成帧时运行该超时：空闲连接和完整 触发；接收进展重置计时；到期先发布 FrameErrorKind.IncompleteFrameTimeout，再拆除对应会话。 适配器把该诊断转换为携带 transport Session ID 的 -HsmsT8TimeoutException，不再次请求重连。 +HsmsT8TimeoutException，不再次请求重连。公共连接另提供默认 +关闭的独立 T8 观测流：适配器在错误回调同步执行期间验证当前 Session ID， +先复制 FrameErrorEventArgs.Bytes，再由会话 actor 过滤旧 Session +并附加观察状态。默认路径不发布该 transport fault 事件。 + +StreamFrame 2.5.0 的 T8 Bytes 是已复制、最多 8 KiB 的缓冲前缀，可能包含 +四字节 HSMS 长度头；API 不提供原缓冲总长度或完整性，SecsFrame 因而只称 +其为前缀快照。观测通道容量显式可配，满时丢弃最旧项，不能反向阻塞协议 +actor。DecodeFailed、IncompleteFrameOverflow 和 DiscardedByResync 的公共 +FrameError 不携带 session/epoch，本切片不把它们映射成会话关联观测。 ## 验证边界 适配器单元测试覆盖原生 Session ID、迟到消息、发送完成、并发入队、会话 -失效、显式关闭原因、关闭/释放竞态和 T8 诊断映射。真实 TCP 回环测试覆盖 -分片帧收发、实际 T8 到期，以及会话替换时排队发送的精确失效与新 Socket +失效、显式关闭原因、关闭/释放竞态、T8 诊断映射、快照复制、旧会话过滤和 +有界队列。真实 TCP 回环测试覆盖分片帧收发、实际 T8 到期及前缀快照，以及 +会话替换时排队发送的精确失效与新 Socket 内容隔离；Passive 重连竞态测试反复并发对端关闭与本地显式关闭，验证监听 恢复、Session ID 单调递增和原会话精确关闭。完整套件继续覆盖 T3/T6 从 发送完成点启动以及旧会话不重放。 diff --git a/docs/TRACE.md b/docs/TRACE.md index a8994fd..b5af2bc 100644 --- a/docs/TRACE.md +++ b/docs/TRACE.md @@ -12,10 +12,11 @@ - 显式启用的完整控制观测或公共未认领控制事件的十字节头元数据导出与 严格读取。 - 已成帧但 SECS-II Body 解码失败的显式分级故障样本。 +- 显式启用、会话关联且容量有界的 T8 前缀快照分级导出。 -当前切片不是原始网络抓包器，不记录 TCP 字节、transport Session generation、 -控制帧的线上原始字节或异常对象，也不会接管 HsmsConnection 的 -单消费者事件流。 +当前切片不是原始网络抓包器；除显式启用的 T8 有界前缀外，不记录 TCP +字节、transport Session generation、控制帧的线上原始字节或异常对象， +也不会接管 HsmsConnection 的单消费者事件流。 应用应在自己的事件循环中从 HsmsIncomingDataMessage 创建接收记录， 并在调用发送 API 前显式创建发送记录。 @@ -184,6 +185,64 @@ payload 捕获默认限制为 64 KiB；超限立即失败，不截断样本。 未完成片段，也没有重放 API。清零范围可验证，但哪些范围包含敏感数据仍是 调用方的数据分类责任；未经独立审查不得把 Redacted 等同于已安全公开。 +## T8 前缀快照 + +连接必须先显式启用独立且容量有界的 T8 观测流；Trace 不自动订阅： + +~~~csharp +var options = new HsmsConnectionOptions( + ipAddress, + port, + HsmsConnectionMode.Passive, + sessionId, + t3, + t5, + t6, + t7, + t8, + enableTransportFaultObservation: true, + transportFaultObservationCapacity: 16); + +await foreach (var fault in connection + .GetTransportFaultObservationsAsync(cancellationToken)) +{ + var record = SecsTraceTransportFaultRecord.Create( + DateTimeOffset.UtcNow, + fault, + SecsTraceTransportFaultCaptureOptions.MetadataOnly()); + + var text = new SecsTraceTransportFaultCodec().Encode( + new[] { record }); +} +~~~ + +观测保留 IncompleteFrameTimeout、transport Session ID、actor +状态和 StreamFrame 已复制的前缀。通道满时丢弃最旧项而不阻塞协议 actor； +默认关闭时不创建通道。快照最多 8 KiB，可能包含四字节 HSMS 长度前缀， +但 StreamFrame 2.5.0 不提供原缓冲总长度或完整性标记，因此不得宣称它是 +完整 TCP 片段。 + +SecsTraceTransportFaultCaptureOptions 同样要求明确选择 +MetadataOnly、RedactedPayload 或 RawPayload。payload 模式不允许截断， +超过调用方收紧后的限制立即失败；Redacted 范围相对快照起点，必须有序、 +非重叠、不越界并在记录中全部为零。记录和公共观测都防御性复制字节。 + +独立信封使用九个单空格分隔字段： + +~~~text +SecsFrame-TransportFaultTrace/1 +TransportFault 2026-08-29T05:00:00.0000000Z IncompleteFrameTimeout Connected 17 RedactedPayload 10 4:6 0000000A000000000000 +~~~ + +字段依次为 UTC 时间、fault kind、HSMS 状态、transport Session ID、数据 +分类、StreamFrame 实际提供的快照长度、脱敏范围和大写十六进制快照。默认 +codec 只允许 MetadataOnly；带快照的导入和导出必须再次显式启用。信封不含 +异常、原缓冲总长度、分片时序或完整性推断，也没有重放 API。 + +当前只接受 T8。StreamFrame 2.5.0 的其他 FrameError 不携带原始 session/epoch， +SecsFrame 不会以回调时读取当前 Session ID 的方式把它们包装成严格会话关联 +样本。 + ## 结构化脱敏 脱敏发生在不可变 Item 树上，而不是对 SML 文本做字符串替换： @@ -278,4 +337,6 @@ var results = await new SecsTraceReplayer().ReplayWithTimingAsync( 测试。控制信封另有完整 Select/Linktest 双向观测、未匹配 Reject 黄金向量、 未知 SType 往返、Data Message 拒绝、严格字段与资源限制测试。故障样本 信封另有三种分类、payload 默认拒绝、范围清零复核、防御性复制、畸形字段 -和 Body/记录/文本资源限制测试。 +和 Body/记录/文本资源限制测试。T8 信封另有真实 TCP 前缀、旧 Session +过滤、有界队列、三种分类、默认拒绝、脱敏复核、畸形字段和快照/记录/文本 +资源限制测试。 diff --git a/src/SecsFrame.Trace/SecsTraceByteRedactionRange.cs b/src/SecsFrame.Trace/SecsTraceByteRedactionRange.cs index 2b374e3..68fbc2e 100644 --- a/src/SecsFrame.Trace/SecsTraceByteRedactionRange.cs +++ b/src/SecsFrame.Trace/SecsTraceByteRedactionRange.cs @@ -2,7 +2,7 @@ namespace SecsFrame.Trace; -/// Identifies a zero-based range within an HSMS data-message body. +/// Identifies a zero-based range within a fault-sample byte payload. [StructLayout(LayoutKind.Auto)] public readonly record struct SecsTraceByteRedactionRange { diff --git a/src/SecsFrame.Trace/SecsTraceFaultSampleDataClassification.cs b/src/SecsFrame.Trace/SecsTraceFaultSampleDataClassification.cs index b602956..060b360 100644 --- a/src/SecsFrame.Trace/SecsTraceFaultSampleDataClassification.cs +++ b/src/SecsFrame.Trace/SecsTraceFaultSampleDataClassification.cs @@ -1,14 +1,14 @@ namespace SecsFrame.Trace; -/// Identifies how much payload data a fault-sample record contains. +/// Identifies how much byte content a fault-sample record contains. public enum SecsTraceFaultSampleDataClassification { - /// Only the HSMS header and original body length are retained. + /// Only stable metadata and the observed byte length are retained. MetadataOnly, - /// The body is retained after declared byte ranges are zeroed. + /// The bytes are retained after declared ranges are zeroed. RedactedPayload, - /// The original body bytes are retained without redaction. + /// The original observed bytes are retained without redaction. RawPayload, } diff --git a/src/SecsFrame.Trace/SecsTraceTransportFaultCaptureOptions.cs b/src/SecsFrame.Trace/SecsTraceTransportFaultCaptureOptions.cs new file mode 100644 index 0000000..69b16f7 --- /dev/null +++ b/src/SecsFrame.Trace/SecsTraceTransportFaultCaptureOptions.cs @@ -0,0 +1,106 @@ +using System.Collections.ObjectModel; + +namespace SecsFrame.Trace; + +/// Defines the explicit data boundary for one T8 prefix-snapshot capture. +public sealed class SecsTraceTransportFaultCaptureOptions +{ + /// The maximum StreamFrame T8 prefix snapshot: 8 KiB. + public const int DefaultMaxSnapshotBytes = + HsmsTransportFaultObservation.MaxSnapshotBytes; + + private SecsTraceTransportFaultCaptureOptions( + SecsTraceFaultSampleDataClassification dataClassification, + IEnumerable? redactionRanges, + int maxSnapshotBytes) + { + if (maxSnapshotBytes is <= 0 or > DefaultMaxSnapshotBytes) + { + throw new ArgumentOutOfRangeException( + nameof(maxSnapshotBytes), + maxSnapshotBytes, + $"The maximum snapshot size must be between 1 and {DefaultMaxSnapshotBytes} bytes."); + } + + var ranges = PrepareRanges(redactionRanges); + if (dataClassification == + SecsTraceFaultSampleDataClassification.RedactedPayload) + { + if (ranges.Count == 0) + { + throw new ArgumentException( + "Redacted snapshot capture requires at least one byte range.", + nameof(redactionRanges)); + } + } + else if (ranges.Count != 0) + { + throw new ArgumentException( + "Only redacted snapshot capture can declare byte ranges.", + nameof(redactionRanges)); + } + + DataClassification = dataClassification; + RedactionRanges = ranges; + MaxSnapshotBytes = maxSnapshotBytes; + } + + /// Gets how much snapshot data can be copied. + public SecsTraceFaultSampleDataClassification DataClassification { get; } + + /// Gets the ordered, nonoverlapping ranges zeroed before capture. + public IReadOnlyList RedactionRanges { get; } + + /// Gets the maximum snapshot size copied by payload-bearing modes. + public int MaxSnapshotBytes { get; } + + /// Creates an option that captures no snapshot bytes. + public static SecsTraceTransportFaultCaptureOptions MetadataOnly() + => new( + SecsTraceFaultSampleDataClassification.MetadataOnly, + null, + DefaultMaxSnapshotBytes); + + /// Creates an option that copies and zeroes declared snapshot ranges. + public static SecsTraceTransportFaultCaptureOptions RedactedPayload( + IEnumerable redactionRanges, + int maxSnapshotBytes = DefaultMaxSnapshotBytes) + { + if (redactionRanges is null) + throw new ArgumentNullException(nameof(redactionRanges)); + + return new( + SecsTraceFaultSampleDataClassification.RedactedPayload, + redactionRanges, + maxSnapshotBytes); + } + + /// Creates an option that explicitly copies the unredacted snapshot. + public static SecsTraceTransportFaultCaptureOptions RawPayload( + int maxSnapshotBytes = DefaultMaxSnapshotBytes) + => new( + SecsTraceFaultSampleDataClassification.RawPayload, + null, + maxSnapshotBytes); + + private static IReadOnlyList PrepareRanges( + IEnumerable? ranges) + { + if (ranges is null) + return Array.Empty(); + + var prepared = ranges.OrderBy(static item => item.Offset).ToArray(); + for (var index = 1; index < prepared.Length; index++) + { + var previous = prepared[index - 1]; + if (previous.Offset + previous.Length > prepared[index].Offset) + { + throw new ArgumentException( + "Byte redaction ranges cannot overlap.", + nameof(ranges)); + } + } + + return new ReadOnlyCollection(prepared); + } +} diff --git a/src/SecsFrame.Trace/SecsTraceTransportFaultCodec.cs b/src/SecsFrame.Trace/SecsTraceTransportFaultCodec.cs new file mode 100644 index 0000000..d235419 --- /dev/null +++ b/src/SecsFrame.Trace/SecsTraceTransportFaultCodec.cs @@ -0,0 +1,398 @@ +using System.Collections.ObjectModel; +using System.Globalization; +using System.Text; + +namespace SecsFrame.Trace; + +/// Encodes and decodes explicitly classified T8 prefix snapshots. +public sealed class SecsTraceTransportFaultCodec +{ + /// The exact transport-fault trace format identifier. + public const string FormatIdentifier = "SecsFrame-TransportFaultTrace/1"; + + /// The default maximum number of records. + public const int DefaultMaxRecordCount = SecsTraceCodec.DefaultMaxRecordCount; + + /// The default maximum trace text length. + public const int DefaultMaxTextLength = SecsTraceCodec.DefaultMaxTextLength; + + /// The default maximum snapshot accepted in one payload record. + public const int DefaultMaxSnapshotBytes = + SecsTraceTransportFaultCaptureOptions.DefaultMaxSnapshotBytes; + + /// Creates a transport-fault codec with explicit data and resource limits. + public SecsTraceTransportFaultCodec( + bool allowPayloadRecords = false, + int maxSnapshotBytes = DefaultMaxSnapshotBytes, + int maxRecordCount = DefaultMaxRecordCount, + int maxTextLength = DefaultMaxTextLength) + { + if (maxSnapshotBytes is <= 0 or > DefaultMaxSnapshotBytes) + { + throw new ArgumentOutOfRangeException( + nameof(maxSnapshotBytes), + maxSnapshotBytes, + $"The maximum snapshot size must be between 1 and {DefaultMaxSnapshotBytes} bytes."); + } + if (maxRecordCount <= 0) + throw new ArgumentOutOfRangeException(nameof(maxRecordCount), maxRecordCount, "The maximum record count must be positive."); + if (maxTextLength <= 0) + throw new ArgumentOutOfRangeException(nameof(maxTextLength), maxTextLength, "The maximum text length must be positive."); + + AllowPayloadRecords = allowPayloadRecords; + MaxSnapshotBytes = maxSnapshotBytes; + MaxRecordCount = maxRecordCount; + MaxTextLength = maxTextLength; + } + + /// Gets whether payload-bearing records can be imported or exported. + public bool AllowPayloadRecords { get; } + + /// Gets the maximum snapshot bytes in one payload-bearing record. + public int MaxSnapshotBytes { get; } + + /// Gets the maximum number of records. + public int MaxRecordCount { get; } + + /// Gets the maximum accepted or produced trace length. + public int MaxTextLength { get; } + + /// Encodes records in enumeration order using LF line endings. + public string Encode(IEnumerable records) + { + if (records is null) + throw new ArgumentNullException(nameof(records)); + + var text = new StringBuilder(FormatIdentifier); + text.Append('\n'); + EnsureTextLength(text); + var recordCount = 0; + foreach (var record in records) + { + if (record is null) + throw new ArgumentException("The transport-fault trace sequence contains a null record.", nameof(records)); + if (++recordCount > MaxRecordCount) + throw new InvalidOperationException($"The transport-fault trace record count exceeds the configured maximum {MaxRecordCount}."); + + EnsureRecordAllowed(record); + AppendRecord(text, record); + EnsureTextLength(text); + } + + return text.ToString(); + } + + /// Strictly decodes one complete transport-fault trace file. + public IReadOnlyList Decode(string text) + { + if (text is null) + throw new ArgumentNullException(nameof(text)); + if (text.Length > MaxTextLength) + throw new ArgumentOutOfRangeException(nameof(text), text.Length, $"The transport-fault trace text length cannot exceed {MaxTextLength} characters."); + + return new Parser(text, this).Parse(); + } + + private void EnsureRecordAllowed(SecsTraceTransportFaultRecord record) + { + if (record.DataClassification != + SecsTraceFaultSampleDataClassification.MetadataOnly && + !AllowPayloadRecords) + { + throw new InvalidOperationException( + "Payload-bearing transport-fault samples require AllowPayloadRecords to be enabled explicitly."); + } + + if (record.Snapshot.Length > MaxSnapshotBytes) + { + throw new InvalidOperationException( + $"The transport-fault snapshot length {record.Snapshot.Length} exceeds the configured maximum {MaxSnapshotBytes}."); + } + } + + private static void AppendRecord( + StringBuilder text, + SecsTraceTransportFaultRecord record) + { + text.Append("TransportFault "); + text.Append(record.Timestamp.UtcDateTime.ToString("O", CultureInfo.InvariantCulture)); + text.Append(' '); + text.Append(record.Kind); + text.Append(' '); + text.Append(record.State); + text.Append(' '); + text.Append(record.TransportSessionId.ToString(CultureInfo.InvariantCulture)); + text.Append(' '); + text.Append(record.DataClassification); + text.Append(' '); + text.Append(record.ObservedSnapshotLength.ToString(CultureInfo.InvariantCulture)); + text.Append(' '); + AppendRanges(text, record.RedactionRanges); + text.Append(' '); + AppendSnapshot(text, record.Snapshot); + text.Append('\n'); + } + + private static void AppendRanges( + StringBuilder text, + IReadOnlyList ranges) + { + if (ranges.Count == 0) + { + text.Append('-'); + return; + } + + for (var index = 0; index < ranges.Count; index++) + { + if (index != 0) + text.Append(','); + text.Append(ranges[index].Offset.ToString(CultureInfo.InvariantCulture)); + text.Append(':'); + text.Append(ranges[index].Length.ToString(CultureInfo.InvariantCulture)); + } + } + + private static void AppendSnapshot( + StringBuilder text, + ReadOnlySpan snapshot) + { + if (snapshot.IsEmpty) + { + text.Append('-'); + return; + } + + foreach (var value in snapshot) + text.Append(value.ToString("X2", CultureInfo.InvariantCulture)); + } + + private void EnsureTextLength(StringBuilder text) + { + if (text.Length > MaxTextLength) + throw new InvalidOperationException($"The transport-fault trace text length exceeds the configured maximum {MaxTextLength}."); + } + + private sealed class Parser + { + private readonly string _text; + private readonly SecsTraceTransportFaultCodec _options; + private int _index; + private int _recordIndex; + + public Parser(string text, SecsTraceTransportFaultCodec options) + { + _text = text; + _options = options; + } + + public IReadOnlyList Parse() + { + var headerOffset = _index; + var formatIdentifier = ReadLine(); + if (!string.Equals(formatIdentifier, FormatIdentifier, StringComparison.Ordinal)) + throw Error("The transport-fault trace format identifier is missing or unsupported", -1, headerOffset); + + var records = new List(); + while (!IsEnd) + { + if (_recordIndex >= _options.MaxRecordCount) + throw Error($"The transport-fault trace record count exceeds the configured maximum {_options.MaxRecordCount}", _recordIndex, _index); + records.Add(ParseRecord()); + _recordIndex++; + } + + return new ReadOnlyCollection(records); + } + + private SecsTraceTransportFaultRecord ParseRecord() + { + var recordOffset = _index; + var fields = ReadLine().Split(new[] { ' ' }, StringSplitOptions.None); + if (fields.Length != 9 || + !string.Equals(fields[0], "TransportFault", StringComparison.Ordinal)) + { + throw Error("A transport-fault record must contain exactly nine single-space-separated fields", _recordIndex, recordOffset); + } + + var classification = ParseEnum(fields[5], "data classification", recordOffset); + if (classification != SecsTraceFaultSampleDataClassification.MetadataOnly && + !_options.AllowPayloadRecords) + { + throw Error("Payload-bearing transport-fault samples require AllowPayloadRecords to be enabled explicitly", _recordIndex, recordOffset); + } + + var snapshot = ParseSnapshot(fields[8], recordOffset); + var ranges = ParseRanges(fields[7], recordOffset); + try + { + return new SecsTraceTransportFaultRecord( + ParseTimestamp(fields[1], recordOffset), + ParseEnum(fields[2], "transport-fault kind", recordOffset), + ParseEnum(fields[3], "session state", recordOffset), + ParseInt64(fields[4], "transport Session ID", recordOffset), + classification, + ParseInt32(fields[6], "observed snapshot length", recordOffset), + snapshot, + ranges); + } + catch (ArgumentException ex) + { + throw Error("The transport-fault data boundary is invalid", _recordIndex, recordOffset, ex); + } + } + + private DateTimeOffset ParseTimestamp(string value, int offset) + { + if (!DateTimeOffset.TryParseExact( + value, + "O", + CultureInfo.InvariantCulture, + DateTimeStyles.RoundtripKind, + out var timestamp)) + { + throw Error("The transport-fault timestamp must use the round-trip format", _recordIndex, offset); + } + + return timestamp; + } + + private TEnum ParseEnum(string value, string fieldName, int offset) + where TEnum : struct, Enum + { + if (!Enum.TryParse(value, ignoreCase: false, out TEnum parsed) || + !Enum.IsDefined(typeof(TEnum), parsed) || + !string.Equals(value, parsed.ToString(), StringComparison.Ordinal)) + { + throw Error($"The {fieldName} is unknown", _recordIndex, offset); + } + + return parsed; + } + + private int ParseInt32(string value, string fieldName, int offset) + { + if (!int.TryParse(value, NumberStyles.None, CultureInfo.InvariantCulture, out var parsed) || + parsed <= 0 || + !string.Equals(value, parsed.ToString(CultureInfo.InvariantCulture), StringComparison.Ordinal)) + { + throw Error($"The {fieldName} must be a canonical positive 32-bit integer", _recordIndex, offset); + } + + return parsed; + } + + private long ParseInt64(string value, string fieldName, int offset) + { + if (!long.TryParse(value, NumberStyles.None, CultureInfo.InvariantCulture, out var parsed) || + parsed <= 0 || + !string.Equals(value, parsed.ToString(CultureInfo.InvariantCulture), StringComparison.Ordinal)) + { + throw Error($"The {fieldName} must be a canonical positive 64-bit integer", _recordIndex, offset); + } + + return parsed; + } + + private IReadOnlyList ParseRanges( + string value, + int offset) + { + if (string.Equals(value, "-", StringComparison.Ordinal)) + return Array.Empty(); + + var parts = value.Split(','); + var ranges = new SecsTraceByteRedactionRange[parts.Length]; + for (var index = 0; index < parts.Length; index++) + { + var fields = parts[index].Split(':'); + if (fields.Length != 2) + throw Error("A redaction range must use the exact form offset:length", _recordIndex, offset); + var rangeOffset = ParseCanonicalRangeValue(fields[0], "redaction offset", offset, allowZero: true); + var length = ParseCanonicalRangeValue(fields[1], "redaction length", offset, allowZero: false); + try + { + ranges[index] = new SecsTraceByteRedactionRange(rangeOffset, length); + } + catch (ArgumentOutOfRangeException ex) + { + throw Error("A redaction range is outside the supported integer range", _recordIndex, offset, ex); + } + } + + return ranges; + } + + private int ParseCanonicalRangeValue( + string value, + string fieldName, + int offset, + bool allowZero) + { + if (!int.TryParse(value, NumberStyles.None, CultureInfo.InvariantCulture, out var parsed) || + parsed < (allowZero ? 0 : 1) || + !string.Equals(value, parsed.ToString(CultureInfo.InvariantCulture), StringComparison.Ordinal)) + { + throw Error($"The {fieldName} is not canonical", _recordIndex, offset); + } + + return parsed; + } + + private byte[] ParseSnapshot(string value, int offset) + { + if (string.Equals(value, "-", StringComparison.Ordinal)) + return Array.Empty(); + if (value.Length % 2 != 0) + throw Error("The snapshot must contain an even number of uppercase hexadecimal digits", _recordIndex, offset); + + var snapshotLength = value.Length / 2; + if (snapshotLength > _options.MaxSnapshotBytes) + throw Error($"The transport-fault snapshot length exceeds the configured maximum {_options.MaxSnapshotBytes}", _recordIndex, offset); + + var snapshot = new byte[snapshotLength]; + for (var index = 0; index < snapshot.Length; index++) + { + var high = ParseHexDigit(value[index * 2]); + var low = ParseHexDigit(value[(index * 2) + 1]); + if (high < 0 || low < 0) + throw Error("The snapshot must contain only uppercase hexadecimal digits", _recordIndex, offset); + snapshot[index] = (byte)((high << 4) | low); + } + + return snapshot; + } + + private static int ParseHexDigit(char value) + => value switch + { + >= '0' and <= '9' => value - '0', + >= 'A' and <= 'F' => value - 'A' + 10, + _ => -1, + }; + + private string ReadLine() + { + var start = _index; + while (!IsEnd && _text[_index] != '\n') + _index++; + + var length = _index - start; + if (length > 0 && _text[start + length - 1] == '\r') + length--; + var line = _text.Substring(start, length); + if (!IsEnd) + _index++; + return line; + } + + private bool IsEnd => _index >= _text.Length; + + private static SecsTraceParseException Error( + string message, + int recordIndex, + int offset, + Exception? innerException = null) + => new(message, recordIndex, offset, innerException); + } +} diff --git a/src/SecsFrame.Trace/SecsTraceTransportFaultRecord.cs b/src/SecsFrame.Trace/SecsTraceTransportFaultRecord.cs new file mode 100644 index 0000000..707782f --- /dev/null +++ b/src/SecsFrame.Trace/SecsTraceTransportFaultRecord.cs @@ -0,0 +1,230 @@ +using System.Collections.ObjectModel; + +namespace SecsFrame.Trace; + +/// Describes an explicitly classified T8 prefix-snapshot sample. +/// +/// The sample never contains an exception or a claimed complete TCP fragment. +/// StreamFrame provides at most the first 8 KiB and does not expose the original +/// buffered length or a completeness flag. +/// +public sealed class SecsTraceTransportFaultRecord +{ + private readonly byte[] _snapshot; + + /// Creates an immutable T8 transport-fault record. + public SecsTraceTransportFaultRecord( + DateTimeOffset timestamp, + HsmsTransportFaultKind kind, + HsmsSessionState state, + long transportSessionId, + SecsTraceFaultSampleDataClassification dataClassification, + int observedSnapshotLength, + ReadOnlySpan snapshot = default, + IEnumerable? redactionRanges = null) + { + if (!Enum.IsDefined(typeof(HsmsTransportFaultKind), kind)) + throw new ArgumentOutOfRangeException(nameof(kind), kind, "Unknown HSMS transport-fault kind."); + if (!Enum.IsDefined(typeof(HsmsSessionState), state)) + throw new ArgumentOutOfRangeException(nameof(state), state, "Unknown HSMS session state."); + if (transportSessionId <= 0) + throw new ArgumentOutOfRangeException(nameof(transportSessionId), transportSessionId, "The transport session identifier must be positive."); + if (!Enum.IsDefined(typeof(SecsTraceFaultSampleDataClassification), dataClassification)) + throw new ArgumentOutOfRangeException(nameof(dataClassification), dataClassification, "Unknown fault-sample data classification."); + if (observedSnapshotLength is <= 0 or > + HsmsTransportFaultObservation.MaxSnapshotBytes) + { + throw new ArgumentOutOfRangeException( + nameof(observedSnapshotLength), + observedSnapshotLength, + $"The observed snapshot length must be between 1 and {HsmsTransportFaultObservation.MaxSnapshotBytes} bytes."); + } + + var ranges = PrepareRanges(redactionRanges); + ValidateDataBoundary( + dataClassification, + observedSnapshotLength, + snapshot, + ranges); + + Timestamp = timestamp.ToUniversalTime(); + Kind = kind; + State = state; + TransportSessionId = transportSessionId; + DataClassification = dataClassification; + ObservedSnapshotLength = observedSnapshotLength; + _snapshot = snapshot.ToArray(); + RedactionRanges = ranges; + } + + /// Gets the UTC timestamp associated with the local observation. + public DateTimeOffset Timestamp { get; } + + /// Gets the transport fault kind. + public HsmsTransportFaultKind Kind { get; } + + /// Gets the HSMS session state observed with the fault. + public HsmsSessionState State { get; } + + /// Gets the StreamFrame TCP-session generation. + public long TransportSessionId { get; } + + /// Gets how much snapshot data the record contains. + public SecsTraceFaultSampleDataClassification DataClassification { get; } + + /// Gets the number of prefix bytes supplied by StreamFrame. + public int ObservedSnapshotLength { get; } + + /// Gets the copied snapshot bytes, empty for metadata-only records. + public ReadOnlySpan Snapshot => _snapshot; + + /// Gets the ordered ranges that were zeroed before capture. + public IReadOnlyList RedactionRanges { get; } + + /// Creates a record from a public T8 transport-fault observation. + public static SecsTraceTransportFaultRecord Create( + DateTimeOffset timestamp, + HsmsTransportFaultObservation observation, + SecsTraceTransportFaultCaptureOptions options) + { + if (observation is null) + throw new ArgumentNullException(nameof(observation)); + if (options is null) + throw new ArgumentNullException(nameof(options)); + + return options.DataClassification switch + { + SecsTraceFaultSampleDataClassification.MetadataOnly => + new SecsTraceTransportFaultRecord( + timestamp, + observation.Kind, + observation.State, + observation.TransportSessionId, + options.DataClassification, + observation.Snapshot.Length), + SecsTraceFaultSampleDataClassification.RedactedPayload => + CreateRedacted(timestamp, observation, options), + SecsTraceFaultSampleDataClassification.RawPayload => + CreateRaw(timestamp, observation, options), + _ => throw new ArgumentOutOfRangeException( + nameof(options), + options.DataClassification, + "Unknown fault-sample data classification."), + }; + } + + private static SecsTraceTransportFaultRecord CreateRedacted( + DateTimeOffset timestamp, + HsmsTransportFaultObservation observation, + SecsTraceTransportFaultCaptureOptions options) + { + ValidateSnapshotSize(observation.Snapshot.Length, options.MaxSnapshotBytes); + var snapshot = observation.Snapshot.ToArray(); + foreach (var range in options.RedactionRanges) + { + if (range.Offset + range.Length > snapshot.Length) + { + throw new ArgumentOutOfRangeException( + nameof(options), + range, + "A byte redaction range exceeds the T8 prefix snapshot."); + } + + Array.Clear(snapshot, range.Offset, range.Length); + } + + return new SecsTraceTransportFaultRecord( + timestamp, + observation.Kind, + observation.State, + observation.TransportSessionId, + options.DataClassification, + snapshot.Length, + snapshot, + options.RedactionRanges); + } + + private static SecsTraceTransportFaultRecord CreateRaw( + DateTimeOffset timestamp, + HsmsTransportFaultObservation observation, + SecsTraceTransportFaultCaptureOptions options) + { + ValidateSnapshotSize(observation.Snapshot.Length, options.MaxSnapshotBytes); + return new SecsTraceTransportFaultRecord( + timestamp, + observation.Kind, + observation.State, + observation.TransportSessionId, + options.DataClassification, + observation.Snapshot.Length, + observation.Snapshot); + } + + private static void ValidateSnapshotSize( + int snapshotLength, + int maxSnapshotBytes) + { + if (snapshotLength > maxSnapshotBytes) + { + throw new InvalidOperationException( + $"The T8 prefix snapshot length {snapshotLength} exceeds the configured maximum {maxSnapshotBytes}."); + } + } + + private static IReadOnlyList PrepareRanges( + IEnumerable? ranges) + { + if (ranges is null) + return Array.Empty(); + + var prepared = ranges.ToArray(); + for (var index = 1; index < prepared.Length; index++) + { + var previous = prepared[index - 1]; + if (previous.Offset + previous.Length > prepared[index].Offset) + { + throw new ArgumentException( + "Byte redaction ranges must be ordered and cannot overlap.", + nameof(ranges)); + } + } + + return new ReadOnlyCollection(prepared); + } + + private static void ValidateDataBoundary( + SecsTraceFaultSampleDataClassification dataClassification, + int observedSnapshotLength, + ReadOnlySpan snapshot, + IReadOnlyList ranges) + { + if (dataClassification == SecsTraceFaultSampleDataClassification.MetadataOnly) + { + if (!snapshot.IsEmpty || ranges.Count != 0) + throw new ArgumentException("A metadata-only sample cannot contain snapshot bytes or redaction ranges.", nameof(snapshot)); + return; + } + + if (snapshot.Length != observedSnapshotLength) + throw new ArgumentException("A payload sample must retain every observed snapshot byte.", nameof(snapshot)); + if (dataClassification == SecsTraceFaultSampleDataClassification.RawPayload) + { + if (ranges.Count != 0) + throw new ArgumentException("A raw payload sample cannot declare redaction ranges.", nameof(ranges)); + return; + } + + if (ranges.Count == 0) + throw new ArgumentException("A redacted payload sample requires at least one byte range.", nameof(ranges)); + foreach (var range in ranges) + { + if (range.Offset + range.Length > snapshot.Length) + throw new ArgumentOutOfRangeException(nameof(ranges), range, "A byte redaction range exceeds the T8 prefix snapshot."); + for (var index = range.Offset; index < range.Offset + range.Length; index++) + { + if (snapshot[index] != 0) + throw new ArgumentException("Every declared redaction range must contain only zero bytes.", nameof(snapshot)); + } + } + } +} diff --git a/src/SecsFrame/Hsms/Connection/HsmsConnection.cs b/src/SecsFrame/Hsms/Connection/HsmsConnection.cs index e5872d2..3a3122b 100644 --- a/src/SecsFrame/Hsms/Connection/HsmsConnection.cs +++ b/src/SecsFrame/Hsms/Connection/HsmsConnection.cs @@ -25,6 +25,7 @@ public sealed class HsmsConnection : IAsyncDisposable private int _state = (int)HsmsSessionState.Disconnected; private int _eventReaderClaimed; private int _controlMessageObservationReaderClaimed; + private int _transportFaultObservationReaderClaimed; private int _started; private int _disposed; @@ -115,6 +116,32 @@ public IAsyncEnumerable return ReadControlMessageObservationsAsync(cancellationToken); } + /// + /// Gets the single-consumer stream of bounded T8 prefix snapshots. + /// + /// Cancels only this observation-stream read. + /// + /// This stream is independent from and is available + /// only when + /// is enabled. Each item contains up to the first 8 KiB retained by StreamFrame, + /// can include the four-byte HSMS length prefix, and is not guaranteed complete. + /// The queue discards its oldest item at the configured capacity. Only one reader + /// can be active at a time. + /// + public IAsyncEnumerable + GetTransportFaultObservationsAsync( + CancellationToken cancellationToken = default) + { + ThrowIfNotRunning(); + if (!Options.EnableTransportFaultObservation) + { + throw new InvalidOperationException( + "HSMS transport-fault observation is not enabled in the connection options."); + } + + return ReadTransportFaultObservationsAsync(cancellationToken); + } + /// Waits until this connection next reaches Selected state. /// Cancels only this wait operation. public Task WaitUntilSelectedAsync( @@ -225,14 +252,19 @@ private static HsmsDataTransactionManager CreateTransactions( options.IpAddress, options.Port, isActive, - new HsmsTransportOptions(options.T5, options.T8)); + new HsmsTransportOptions( + options.T5, + options.T8, + options.EnableTransportFaultObservation)); var session = new HsmsSessionStateMachine( transport, new HsmsSessionOptions( options.ConnectionMode, options.T6, options.T7, - options.EnableControlMessageObservation)); + options.EnableControlMessageObservation, + options.EnableTransportFaultObservation, + options.TransportFaultObservationCapacity)); return new HsmsDataTransactionManager( session, new HsmsDataTransactionOptions(options.T3)); @@ -363,6 +395,35 @@ private async IAsyncEnumerable } } + private async IAsyncEnumerable + ReadTransportFaultObservationsAsync( + [EnumeratorCancellation] CancellationToken cancellationToken) + { + if (Interlocked.Exchange( + ref _transportFaultObservationReaderClaimed, + 1) != 0) + { + throw new InvalidOperationException( + "The HSMS transport-fault observation stream already has a consumer."); + } + + try + { + await foreach (var observation in _transactions + .GetTransportFaultObservationsAsync(cancellationToken) + .ConfigureAwait(false)) + { + yield return observation; + } + } + finally + { + Interlocked.Exchange( + ref _transportFaultObservationReaderClaimed, + 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 16ba5f6..c550145 100644 --- a/src/SecsFrame/Hsms/Connection/HsmsConnectionOptions.cs +++ b/src/SecsFrame/Hsms/Connection/HsmsConnectionOptions.cs @@ -8,6 +8,9 @@ namespace SecsFrame; /// public sealed class HsmsConnectionOptions { + /// The default number of retained T8 observations. + public const int DefaultTransportFaultObservationCapacity = 16; + /// Creates explicit HSMS connection settings. /// The remote address in Active mode or local bind address in Passive mode. /// The TCP port, from 1 through 65535. @@ -21,6 +24,12 @@ public sealed class HsmsConnectionOptions /// /// Enables the separate restricted-metadata control-message observation stream. /// + /// + /// Enables the separate bounded T8 prefix-snapshot observation stream. + /// + /// + /// The maximum queued T8 observations before the oldest item is discarded. + /// public HsmsConnectionOptions( IPAddress ipAddress, int port, @@ -31,7 +40,10 @@ public HsmsConnectionOptions( TimeSpan t6, TimeSpan t7, TimeSpan t8, - bool enableControlMessageObservation = false) + bool enableControlMessageObservation = false, + bool enableTransportFaultObservation = false, + int transportFaultObservationCapacity = + DefaultTransportFaultObservationCapacity) { IpAddress = ipAddress ?? throw new ArgumentNullException(nameof(ipAddress)); if (port is < 1 or > ushort.MaxValue) @@ -64,6 +76,13 @@ public HsmsConnectionOptions( ValidatePositive(t6, nameof(t6), "T6"); ValidatePositive(t7, nameof(t7), "T7"); ValidateStreamFrameMilliseconds(t8, nameof(t8), "T8"); + if (transportFaultObservationCapacity <= 0) + { + throw new ArgumentOutOfRangeException( + nameof(transportFaultObservationCapacity), + transportFaultObservationCapacity, + "The transport-fault observation capacity must be positive."); + } Port = port; ConnectionMode = connectionMode; @@ -74,6 +93,8 @@ public HsmsConnectionOptions( T7 = t7; T8 = t8; EnableControlMessageObservation = enableControlMessageObservation; + EnableTransportFaultObservation = enableTransportFaultObservation; + TransportFaultObservationCapacity = transportFaultObservationCapacity; } /// Gets the remote or local bind IP address. @@ -108,6 +129,12 @@ public HsmsConnectionOptions( /// public bool EnableControlMessageObservation { get; } + /// Gets whether the bounded T8 prefix-snapshot stream is enabled. + public bool EnableTransportFaultObservation { get; } + + /// Gets the maximum number of queued T8 observations. + public int TransportFaultObservationCapacity { get; } + private static void ValidatePositive( TimeSpan value, string parameterName, diff --git a/src/SecsFrame/Hsms/Connection/HsmsTransportFaultKind.cs b/src/SecsFrame/Hsms/Connection/HsmsTransportFaultKind.cs new file mode 100644 index 0000000..17cebc8 --- /dev/null +++ b/src/SecsFrame/Hsms/Connection/HsmsTransportFaultKind.cs @@ -0,0 +1,8 @@ +namespace SecsFrame; + +/// Identifies an explicitly observed HSMS transport fault. +public enum HsmsTransportFaultKind +{ + /// T8 expired while an incomplete HSMS frame was buffered. + IncompleteFrameTimeout, +} diff --git a/src/SecsFrame/Hsms/Connection/HsmsTransportFaultObservation.cs b/src/SecsFrame/Hsms/Connection/HsmsTransportFaultObservation.cs new file mode 100644 index 0000000..33f5271 --- /dev/null +++ b/src/SecsFrame/Hsms/Connection/HsmsTransportFaultObservation.cs @@ -0,0 +1,52 @@ +namespace SecsFrame; + +/// +/// Describes one explicitly enabled, session-associated HSMS transport fault. +/// +/// +/// The snapshot is a defensive copy of at most the first 8 KiB retained by +/// StreamFrame. It can include the four-byte HSMS length prefix and is not +/// guaranteed to contain the complete network fragment. +/// +public sealed class HsmsTransportFaultObservation +{ + private readonly byte[] _snapshot; + + /// The maximum prefix snapshot retained by StreamFrame. + public const int MaxSnapshotBytes = 8 * 1024; + + internal HsmsTransportFaultObservation( + HsmsTransportFaultKind kind, + long transportSessionId, + HsmsSessionState state, + ReadOnlySpan snapshot) + { + if (!Enum.IsDefined(typeof(HsmsTransportFaultKind), kind)) + throw new ArgumentOutOfRangeException(nameof(kind), kind, "Unknown HSMS transport-fault kind."); + if (transportSessionId <= 0) + throw new ArgumentOutOfRangeException(nameof(transportSessionId), transportSessionId, "The transport session identifier must be positive."); + if (!Enum.IsDefined(typeof(HsmsSessionState), state)) + throw new ArgumentOutOfRangeException(nameof(state), state, "Unknown HSMS session state."); + if (snapshot.IsEmpty) + throw new ArgumentException("A T8 observation requires a nonempty prefix snapshot.", nameof(snapshot)); + if (snapshot.Length > MaxSnapshotBytes) + throw new ArgumentOutOfRangeException(nameof(snapshot), snapshot.Length, $"A T8 prefix snapshot cannot exceed {MaxSnapshotBytes} bytes."); + + Kind = kind; + TransportSessionId = transportSessionId; + State = state; + _snapshot = snapshot.ToArray(); + } + + /// Gets the transport fault kind. + public HsmsTransportFaultKind Kind { get; } + + /// Gets the StreamFrame TCP-session generation. + public long TransportSessionId { get; } + + /// Gets the HSMS session state observed with the fault. + public HsmsSessionState State { get; } + + /// Gets the copied, bounded network-prefix snapshot. + public ReadOnlySpan Snapshot => _snapshot; +} diff --git a/src/SecsFrame/Hsms/Session/HsmsSessionOptions.cs b/src/SecsFrame/Hsms/Session/HsmsSessionOptions.cs index 6b1dfd4..0ef9f47 100644 --- a/src/SecsFrame/Hsms/Session/HsmsSessionOptions.cs +++ b/src/SecsFrame/Hsms/Session/HsmsSessionOptions.cs @@ -6,7 +6,10 @@ public HsmsSessionOptions( HsmsConnectionMode connectionMode, TimeSpan controlReplyTimeout, TimeSpan selectionTimeout, - bool enableControlMessageObservation = false) + bool enableControlMessageObservation = false, + bool enableTransportFaultObservation = false, + int transportFaultObservationCapacity = + HsmsConnectionOptions.DefaultTransportFaultObservationCapacity) { if (connectionMode != HsmsConnectionMode.Active && connectionMode != HsmsConnectionMode.Passive) @@ -33,10 +36,20 @@ public HsmsSessionOptions( "The selection timeout must be positive."); } + if (transportFaultObservationCapacity <= 0) + { + throw new ArgumentOutOfRangeException( + nameof(transportFaultObservationCapacity), + transportFaultObservationCapacity, + "The transport-fault observation capacity must be positive."); + } + ConnectionMode = connectionMode; ControlReplyTimeout = controlReplyTimeout; SelectionTimeout = selectionTimeout; EnableControlMessageObservation = enableControlMessageObservation; + EnableTransportFaultObservation = enableTransportFaultObservation; + TransportFaultObservationCapacity = transportFaultObservationCapacity; } public HsmsConnectionMode ConnectionMode { get; } @@ -46,4 +59,8 @@ public HsmsSessionOptions( public TimeSpan SelectionTimeout { get; } public bool EnableControlMessageObservation { get; } + + public bool EnableTransportFaultObservation { get; } + + public int TransportFaultObservationCapacity { get; } } diff --git a/src/SecsFrame/Hsms/Session/HsmsSessionStateMachine.cs b/src/SecsFrame/Hsms/Session/HsmsSessionStateMachine.cs index d561a2c..a6451f6 100644 --- a/src/SecsFrame/Hsms/Session/HsmsSessionStateMachine.cs +++ b/src/SecsFrame/Hsms/Session/HsmsSessionStateMachine.cs @@ -13,6 +13,7 @@ internal sealed class HsmsSessionStateMachine : IAsyncDisposable private readonly Channel _inputs; private readonly Channel _events; private readonly Channel? _controlMessageObservations; + private readonly Channel? _transportFaultObservations; private readonly HashSet _pendingDataSends = new(); private CancellationTokenSource? _lifetime; private Task? _transportPump; @@ -62,6 +63,19 @@ public HsmsSessionStateMachine( AllowSynchronousContinuations = false, }); } + if (options.EnableTransportFaultObservation) + { + _transportFaultObservations = + Channel.CreateBounded( + new BoundedChannelOptions( + options.TransportFaultObservationCapacity) + { + SingleReader = true, + SingleWriter = true, + AllowSynchronousContinuations = false, + FullMode = BoundedChannelFullMode.DropOldest, + }); + } } public HsmsSessionState State @@ -114,6 +128,21 @@ public async IAsyncEnumerable } } + public async IAsyncEnumerable + GetTransportFaultObservationsAsync( + [EnumeratorCancellation] CancellationToken cancellationToken) + { + var observations = _transportFaultObservations ?? + throw new InvalidOperationException( + "HSMS transport-fault 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); @@ -210,6 +239,7 @@ public async ValueTask DisposeAsync() _separateCompletion?.TrySetCanceled(); _events.Writer.TryComplete(); _controlMessageObservations?.Writer.TryComplete(); + _transportFaultObservations?.Writer.TryComplete(); _lifetime?.Dispose(); } @@ -263,6 +293,7 @@ private async Task ProcessInputsAsync() CancelT7(); _events.Writer.TryComplete(); _controlMessageObservations?.Writer.TryComplete(); + _transportFaultObservations?.Writer.TryComplete(); } } @@ -334,6 +365,18 @@ private void ProcessTransportEvent(HsmsTransportEvent transportEvent) if (transportEvent.SessionId == _sessionId) ProcessFrame(transportEvent.Frame!); break; + case HsmsTransportEventKind.TransportFaultObserved: + if (transportEvent.SessionId == _sessionId && + _transportFaultObservations is not null) + { + _transportFaultObservations.Writer.TryWrite( + new HsmsTransportFaultObservation( + transportEvent.FaultKind!.Value, + transportEvent.SessionId.Value, + State, + transportEvent.Snapshot.Span)); + } + break; case HsmsTransportEventKind.SessionClosed: if (transportEvent.SessionId == _sessionId) ProcessSessionClosed(transportEvent.Error); diff --git a/src/SecsFrame/Hsms/Transactions/HsmsDataTransactionManager.cs b/src/SecsFrame/Hsms/Transactions/HsmsDataTransactionManager.cs index 5b68725..84ce791 100644 --- a/src/SecsFrame/Hsms/Transactions/HsmsDataTransactionManager.cs +++ b/src/SecsFrame/Hsms/Transactions/HsmsDataTransactionManager.cs @@ -86,6 +86,10 @@ public IAsyncEnumerable GetControlMessageObservationsAsync(CancellationToken cancellationToken) => _session.GetControlMessageObservationsAsync(cancellationToken); + public IAsyncEnumerable + GetTransportFaultObservationsAsync(CancellationToken cancellationToken) + => _session.GetTransportFaultObservationsAsync(cancellationToken); + public Task SendAsync( ushort sessionId, SecsMessage primary, diff --git a/src/SecsFrame/Transport/HsmsTransportEvent.cs b/src/SecsFrame/Transport/HsmsTransportEvent.cs index d58d3c0..c6b3f2c 100644 --- a/src/SecsFrame/Transport/HsmsTransportEvent.cs +++ b/src/SecsFrame/Transport/HsmsTransportEvent.cs @@ -6,7 +6,9 @@ private HsmsTransportEvent( HsmsTransportEventKind kind, HsmsTransportSessionId sessionId, HsmsFrame? frame, - Exception? error) + Exception? error, + HsmsTransportFaultKind? faultKind, + ReadOnlyMemory snapshot) { if (!sessionId.IsValid) throw new ArgumentOutOfRangeException(nameof(sessionId), sessionId, "The transport session identifier must be positive."); @@ -16,11 +18,24 @@ private HsmsTransportEvent( throw new ArgumentException("Only a frame-received event can carry an HSMS frame.", nameof(frame)); if (kind != HsmsTransportEventKind.SessionClosed && error is not null) throw new ArgumentException("Only a session-closed event can carry an error.", nameof(error)); + var isFault = kind == HsmsTransportEventKind.TransportFaultObserved; + if (isFault && faultKind is null) + throw new ArgumentNullException(nameof(faultKind)); + if (!isFault && faultKind is not null) + throw new ArgumentException("Only a transport-fault event can carry a fault kind.", nameof(faultKind)); + if (isFault && snapshot.IsEmpty) + throw new ArgumentException("A transport-fault event requires a nonempty prefix snapshot.", nameof(snapshot)); + if (!isFault && !snapshot.IsEmpty) + throw new ArgumentException("Only a transport-fault event can carry a prefix snapshot.", nameof(snapshot)); + if (snapshot.Length > HsmsTransportFaultObservation.MaxSnapshotBytes) + throw new ArgumentOutOfRangeException(nameof(snapshot), snapshot.Length, "The transport-fault prefix snapshot exceeds the supported limit."); Kind = kind; SessionId = sessionId; Frame = frame; Error = error; + FaultKind = faultKind; + Snapshot = snapshot; } public HsmsTransportEventKind Kind { get; } @@ -31,14 +46,30 @@ private HsmsTransportEvent( public Exception? Error { get; } + public HsmsTransportFaultKind? FaultKind { get; } + + public ReadOnlyMemory Snapshot { get; } + public static HsmsTransportEvent SessionOpened(HsmsTransportSessionId sessionId) - => new(HsmsTransportEventKind.SessionOpened, sessionId, null, null); + => new(HsmsTransportEventKind.SessionOpened, sessionId, null, null, null, default); public static HsmsTransportEvent FrameReceived(HsmsTransportSessionId sessionId, HsmsFrame frame) - => new(HsmsTransportEventKind.FrameReceived, sessionId, frame, null); + => new(HsmsTransportEventKind.FrameReceived, sessionId, frame, null, null, default); + + public static HsmsTransportEvent TransportFaultObserved( + HsmsTransportSessionId sessionId, + HsmsTransportFaultKind faultKind, + ReadOnlySpan snapshot) + => new( + HsmsTransportEventKind.TransportFaultObserved, + sessionId, + null, + null, + faultKind, + snapshot.ToArray()); public static HsmsTransportEvent SessionClosed( HsmsTransportSessionId sessionId, Exception? error = null) - => new(HsmsTransportEventKind.SessionClosed, sessionId, null, error); + => new(HsmsTransportEventKind.SessionClosed, sessionId, null, error, null, default); } diff --git a/src/SecsFrame/Transport/HsmsTransportEventKind.cs b/src/SecsFrame/Transport/HsmsTransportEventKind.cs index 638dde0..406a9a7 100644 --- a/src/SecsFrame/Transport/HsmsTransportEventKind.cs +++ b/src/SecsFrame/Transport/HsmsTransportEventKind.cs @@ -4,5 +4,6 @@ internal enum HsmsTransportEventKind { SessionOpened, FrameReceived, + TransportFaultObserved, SessionClosed, } diff --git a/src/SecsFrame/Transport/HsmsTransportOptions.cs b/src/SecsFrame/Transport/HsmsTransportOptions.cs index 46c2ec9..e0f8395 100644 --- a/src/SecsFrame/Transport/HsmsTransportOptions.cs +++ b/src/SecsFrame/Transport/HsmsTransportOptions.cs @@ -2,7 +2,10 @@ namespace SecsFrame; internal sealed class HsmsTransportOptions { - public HsmsTransportOptions(TimeSpan t5, TimeSpan t8) + public HsmsTransportOptions( + TimeSpan t5, + TimeSpan t8, + bool enableT8FaultObservation = false) { if (t5 <= TimeSpan.Zero) { @@ -58,12 +61,15 @@ public HsmsTransportOptions(TimeSpan t5, TimeSpan t8) T8 = t8; T5Milliseconds = (int)t5Milliseconds; T8Milliseconds = (int)t8Milliseconds; + EnableT8FaultObservation = enableT8FaultObservation; } public TimeSpan T5 { get; } public TimeSpan T8 { get; } + public bool EnableT8FaultObservation { get; } + internal int T5Milliseconds { get; } internal int T8Milliseconds { get; } diff --git a/src/SecsFrame/Transport/StreamFrameHsmsTransport.cs b/src/SecsFrame/Transport/StreamFrameHsmsTransport.cs index 953d239..3fc8630 100644 --- a/src/SecsFrame/Transport/StreamFrameHsmsTransport.cs +++ b/src/SecsFrame/Transport/StreamFrameHsmsTransport.cs @@ -10,6 +10,7 @@ internal sealed class StreamFrameHsmsTransport : IHsmsTransport { private readonly ISessionAwareStreamConnection _connection; private readonly Channel _events; + private readonly bool _enableT8FaultObservation; #if NET9_0_OR_GREATER private readonly Lock _lifecycleGate = new(); private readonly Lock _sessionGate = new(); @@ -24,9 +25,11 @@ internal sealed class StreamFrameHsmsTransport : IHsmsTransport private int _disposed; internal StreamFrameHsmsTransport( - ISessionAwareStreamConnection connection) + ISessionAwareStreamConnection connection, + bool enableT8FaultObservation = false) { _connection = connection ?? throw new ArgumentNullException(nameof(connection)); + _enableT8FaultObservation = enableT8FaultObservation; _events = Channel.CreateUnbounded(new UnboundedChannelOptions { SingleReader = false, @@ -60,7 +63,9 @@ public static StreamFrameHsmsTransport Create( port, isActive, adaptedOptions); - return new StreamFrameHsmsTransport(connection); + return new StreamFrameHsmsTransport( + connection, + hsmsOptions.EnableT8FaultObservation); } public void Start(CancellationToken cancellationToken) @@ -280,17 +285,28 @@ private void OnFrameError(object? sender, FrameErrorEventArgs args) if (sessionValue <= 0) return; + var isCurrentSession = false; lock (_sessionGate) { if (_currentSessionId == sessionValue && + _connection.CurrentSessionId == sessionValue && _sessions.TryGetValue(sessionValue, out var context) && - !context.IsClosed && - context.CloseReason is null) + !context.IsClosed) { - context.CloseReason = new HsmsT8TimeoutException( + isCurrentSession = true; + context.CloseReason ??= new HsmsT8TimeoutException( new HsmsTransportSessionId(sessionValue)); } } + + if (isCurrentSession && _enableT8FaultObservation) + { + _events.Writer.TryWrite( + HsmsTransportEvent.TransportFaultObserved( + new HsmsTransportSessionId(sessionValue), + HsmsTransportFaultKind.IncompleteFrameTimeout, + args.Bytes.Span)); + } } private void RegisterPendingSend(HsmsTransportSessionId sessionId) diff --git a/test/SecsFrame.Tests/HsmsConnectionOptionsTests.cs b/test/SecsFrame.Tests/HsmsConnectionOptionsTests.cs index 113ca1b..ec50948 100644 --- a/test/SecsFrame.Tests/HsmsConnectionOptionsTests.cs +++ b/test/SecsFrame.Tests/HsmsConnectionOptionsTests.cs @@ -29,6 +29,10 @@ public void Explicit_values_are_preserved() Assert.Equal(TimeSpan.FromSeconds(10), options.T7); Assert.Equal(TimeSpan.FromSeconds(5), options.T8); Assert.False(options.EnableControlMessageObservation); + Assert.False(options.EnableTransportFaultObservation); + Assert.Equal( + HsmsConnectionOptions.DefaultTransportFaultObservationCapacity, + options.TransportFaultObservationCapacity); } [Fact] @@ -49,6 +53,38 @@ public void Control_message_observation_can_be_explicitly_enabled() Assert.True(options.EnableControlMessageObservation); } + [Fact] + public void Transport_fault_observation_can_be_explicitly_bounded() + { + 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), + enableTransportFaultObservation: true, + transportFaultObservationCapacity: 3); + + Assert.True(options.EnableTransportFaultObservation); + Assert.Equal(3, options.TransportFaultObservationCapacity); + Assert.Throws(() => + new HsmsConnectionOptions( + IPAddress.Loopback, + 5000, + HsmsConnectionMode.Active, + 10, + TimeSpan.FromSeconds(45), + TimeSpan.FromSeconds(10), + TimeSpan.FromSeconds(5), + TimeSpan.FromSeconds(10), + TimeSpan.FromSeconds(5), + transportFaultObservationCapacity: 0)); + } + [Fact] public void Null_address_is_rejected() { diff --git a/test/SecsFrame.Tests/HsmsConnectionTests.cs b/test/SecsFrame.Tests/HsmsConnectionTests.cs index 96a4d80..12244ca 100644 --- a/test/SecsFrame.Tests/HsmsConnectionTests.cs +++ b/test/SecsFrame.Tests/HsmsConnectionTests.cs @@ -77,6 +77,41 @@ await AssertLinktestObservationsAsync( activeObservations).ConfigureAwait(true); } + [Fact] + public async Task Enabled_transport_fault_observation_captures_real_T8_prefix() + { + var port = GetFreePort(); + await using var connection = new HsmsConnection( + CreateOptions( + port, + HsmsConnectionMode.Passive, + enableTransportFaultObservation: true, + t8: TimeSpan.FromMilliseconds(250))); + using var cancellation = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + connection.Start(); + await using var observations = connection + .GetTransportFaultObservationsAsync(cancellation.Token) + .GetAsyncEnumerator(); + using var client = new System.Net.Sockets.TcpClient(); + await client.ConnectAsync(IPAddress.Loopback, port).ConfigureAwait(true); + var prefix = new byte[] { 0x00, 0x00, 0x00, 0x0A, 0x00, 0x01 }; + + await client.GetStream().WriteAsync( + prefix, + 0, + prefix.Length, + cancellation.Token).ConfigureAwait(true); + Assert.True(await observations.MoveNextAsync().ConfigureAwait(true)); + var observation = observations.Current; + + Assert.Equal( + HsmsTransportFaultKind.IncompleteFrameTimeout, + observation.Kind); + Assert.True(observation.TransportSessionId > 0); + Assert.Equal(HsmsSessionState.Connected, observation.State); + Assert.Equal(prefix, observation.Snapshot.ToArray()); + } + private static async Task AssertSelectObservationsAsync( IAsyncEnumerator passiveObservations, IAsyncEnumerator activeObservations) @@ -248,6 +283,17 @@ public async Task Control_observation_is_disabled_by_default() () => connection.GetControlMessageObservationsAsync()); } + [Fact] + public async Task Transport_fault_observation_is_disabled_by_default() + { + await using var connection = new HsmsConnection( + CreateOptions(GetFreePort(), HsmsConnectionMode.Active)); + connection.Start(); + + Assert.Throws( + () => connection.GetTransportFaultObservationsAsync()); + } + [Fact] public async Task Control_observation_enforces_one_reader_and_allows_replacement() { @@ -283,6 +329,41 @@ await Assert.ThrowsAnyAsync( () => replacement.MoveNextAsync().AsTask()).ConfigureAwait(true); } + [Fact] + public async Task Transport_fault_observation_enforces_one_reader_and_allows_replacement() + { + await using var connection = new HsmsConnection( + CreateOptions( + GetFreePort(), + HsmsConnectionMode.Active, + enableTransportFaultObservation: true)); + using var cancellation = new CancellationTokenSource(); + connection.Start(); + await using var first = connection + .GetTransportFaultObservationsAsync(cancellation.Token) + .GetAsyncEnumerator(); + var firstRead = first.MoveNextAsync().AsTask(); + var concurrent = connection + .GetTransportFaultObservationsAsync() + .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 + .GetTransportFaultObservationsAsync(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() { @@ -310,7 +391,9 @@ public async Task Connection_rejects_commands_before_start_and_duplicate_start() private static HsmsConnectionOptions CreateOptions( int port, HsmsConnectionMode connectionMode, - bool enableControlMessageObservation = false) + bool enableControlMessageObservation = false, + bool enableTransportFaultObservation = false, + TimeSpan? t8 = null) => new( IPAddress.Loopback, port, @@ -320,8 +403,9 @@ private static HsmsConnectionOptions CreateOptions( t5: TimeSpan.FromMilliseconds(10), t6: TimeSpan.FromSeconds(5), t7: TimeSpan.FromSeconds(10), - t8: TimeSpan.FromSeconds(5), - enableControlMessageObservation: enableControlMessageObservation); + t8: t8 ?? TimeSpan.FromSeconds(5), + enableControlMessageObservation: enableControlMessageObservation, + enableTransportFaultObservation: enableTransportFaultObservation); private static async Task AssertSelectedEventAsync( IAsyncEnumerator passiveEvents, diff --git a/test/SecsFrame.Tests/HsmsSessionStateMachineTests.TransportFaultObservation.cs b/test/SecsFrame.Tests/HsmsSessionStateMachineTests.TransportFaultObservation.cs new file mode 100644 index 0000000..8241add --- /dev/null +++ b/test/SecsFrame.Tests/HsmsSessionStateMachineTests.TransportFaultObservation.cs @@ -0,0 +1,112 @@ +namespace SecsFrame.Tests; + +public sealed partial class HsmsSessionStateMachineTests +{ + [Fact] + public async Task Enabled_T8_observation_filters_stale_transport_sessions() + { + var transport = new FakeHsmsTransport(); + await using var machine = CreateMachine( + transport, + HsmsConnectionMode.Passive, + new ManualTimerFactory(), + enableTransportFaultObservation: true); + using var cancellation = new CancellationTokenSource(TimeSpan.FromSeconds(5)); + var observations = machine + .GetTransportFaultObservationsAsync(cancellation.Token) + .GetAsyncEnumerator(); + await using var observationScope = observations.ConfigureAwait(true); + machine.Start(cancellation.Token); + var firstSession = new HsmsTransportSessionId(1); + var secondSession = new HsmsTransportSessionId(2); + + transport.Open(firstSession); + await WaitUntilAsync(() => machine.State == HsmsSessionState.Connected) + .ConfigureAwait(true); + transport.CloseCurrent(); + await WaitUntilAsync(() => machine.State == HsmsSessionState.Disconnected) + .ConfigureAwait(true); + transport.Open(secondSession); + await WaitUntilAsync(() => machine.State == HsmsSessionState.Connected) + .ConfigureAwait(true); + + transport.ObserveT8(firstSession, new byte[] { 0x01 }); + transport.ObserveT8(secondSession, new byte[] { 0x02 }); + var observation = await NextTransportFaultObservationAsync(observations) + .ConfigureAwait(true); + + Assert.Equal(secondSession.Value, observation.TransportSessionId); + Assert.Equal(HsmsSessionState.Connected, observation.State); + Assert.Equal( + HsmsTransportFaultKind.IncompleteFrameTimeout, + observation.Kind); + Assert.Equal(new byte[] { 0x02 }, observation.Snapshot.ToArray()); + } + + [Fact] + public async Task T8_observation_queue_drops_oldest_without_blocking_state_machine() + { + var transport = new FakeHsmsTransport(); + await using var machine = CreateMachine( + transport, + HsmsConnectionMode.Passive, + new ManualTimerFactory(), + enableTransportFaultObservation: true, + transportFaultObservationCapacity: 2); + using var cancellation = new CancellationTokenSource(TimeSpan.FromSeconds(5)); + machine.Start(cancellation.Token); + var sessionId = new HsmsTransportSessionId(1); + transport.Open(sessionId); + await WaitUntilAsync(() => machine.State == HsmsSessionState.Connected) + .ConfigureAwait(true); + + transport.ObserveT8(sessionId, new byte[] { 0x01 }); + transport.ObserveT8(sessionId, new byte[] { 0x02 }); + transport.ObserveT8(sessionId, new byte[] { 0x03 }); + transport.CloseCurrent(); + await WaitUntilAsync(() => machine.State == HsmsSessionState.Disconnected) + .ConfigureAwait(true); + var observations = machine + .GetTransportFaultObservationsAsync(cancellation.Token) + .GetAsyncEnumerator(); + await using var observationScope = observations.ConfigureAwait(true); + + var second = await NextTransportFaultObservationAsync(observations) + .ConfigureAwait(true); + var third = await NextTransportFaultObservationAsync(observations) + .ConfigureAwait(true); + + Assert.Equal(new byte[] { 0x02 }, second.Snapshot.ToArray()); + Assert.Equal(new byte[] { 0x03 }, third.Snapshot.ToArray()); + Assert.Equal(HsmsSessionState.Disconnected, machine.State); + } + + [Fact] + public async Task T8_observation_is_disabled_by_default() + { + var transport = new FakeHsmsTransport(); + await using var machine = CreateMachine( + transport, + HsmsConnectionMode.Passive, + new ManualTimerFactory()); + + await Assert.ThrowsAsync( + async () => await machine + .GetTransportFaultObservationsAsync(CancellationToken.None) + .GetAsyncEnumerator() + .MoveNextAsync() + .AsTask() + .ConfigureAwait(true)).ConfigureAwait(true); + } + + private static async Task + NextTransportFaultObservationAsync( + IAsyncEnumerator observations) + { + if (await observations.MoveNextAsync().ConfigureAwait(true)) + return observations.Current; + + Assert.Fail("The expected HSMS transport-fault observation was not received."); + return null!; + } +} diff --git a/test/SecsFrame.Tests/HsmsSessionStateMachineTests.cs b/test/SecsFrame.Tests/HsmsSessionStateMachineTests.cs index fb9ffc7..f284cfb 100644 --- a/test/SecsFrame.Tests/HsmsSessionStateMachineTests.cs +++ b/test/SecsFrame.Tests/HsmsSessionStateMachineTests.cs @@ -554,14 +554,19 @@ private static HsmsSessionStateMachine CreateMachine( HsmsConnectionMode mode, ManualTimerFactory timers, IHsmsSystemBytesProvider? systemBytesProvider = null, - bool enableControlMessageObservation = false) + bool enableControlMessageObservation = false, + bool enableTransportFaultObservation = false, + int transportFaultObservationCapacity = + HsmsConnectionOptions.DefaultTransportFaultObservationCapacity) => new( transport, new HsmsSessionOptions( mode, T6, T7, - enableControlMessageObservation), + enableControlMessageObservation, + enableTransportFaultObservation, + transportFaultObservationCapacity), timers, systemBytesProvider); @@ -885,6 +890,15 @@ public void Receive(HsmsFrame frame) HsmsTransportEvent.FrameReceived(sessionId, frame)); } + public void ObserveT8( + HsmsTransportSessionId sessionId, + ReadOnlySpan snapshot) + => _events.Writer.TryWrite( + HsmsTransportEvent.TransportFaultObserved( + sessionId, + HsmsTransportFaultKind.IncompleteFrameTimeout, + snapshot)); + public void CompleteSend(int index) { PendingSend pending; diff --git a/test/SecsFrame.Tests/SecsTraceTransportFaultTests.cs b/test/SecsFrame.Tests/SecsTraceTransportFaultTests.cs new file mode 100644 index 0000000..8ae2f92 --- /dev/null +++ b/test/SecsFrame.Tests/SecsTraceTransportFaultTests.cs @@ -0,0 +1,277 @@ +using SecsFrame.Trace; + +namespace SecsFrame.Tests; + +public sealed class SecsTraceTransportFaultTests +{ + [Fact] + public void Redacted_capture_produces_a_deterministic_verifiable_vector() + { + var observation = CreateObservation( + new byte[] + { + 0x00, 0x00, 0x00, 0x0A, + 0x53, 0x45, 0x43, 0x52, 0x45, 0x54, + }); + var options = SecsTraceTransportFaultCaptureOptions.RedactedPayload( + new[] { new SecsTraceByteRedactionRange(4, 6) }); + var timestamp = new DateTimeOffset( + 2026, + 8, + 29, + 13, + 0, + 0, + TimeSpan.FromHours(8)); + var record = SecsTraceTransportFaultRecord.Create( + timestamp, + observation, + options); + var codec = new SecsTraceTransportFaultCodec( + allowPayloadRecords: true); + + var text = codec.Encode(new[] { record }); + + Assert.Equal( + "SecsFrame-TransportFaultTrace/1\n" + + "TransportFault 2026-08-29T05:00:00.0000000Z IncompleteFrameTimeout Connected 17 RedactedPayload 10 4:6 0000000A000000000000\n", + text); + Assert.DoesNotContain("534543524554", text, StringComparison.Ordinal); + var decoded = Assert.Single(codec.Decode(text)); + AssertRecordEqual(record, decoded); + Assert.All( + decoded.Snapshot.Slice(4, 6).ToArray(), + static value => Assert.Equal((byte)0, value)); + } + + [Fact] + public void Default_codec_allows_metadata_only_and_rejects_payload_records() + { + var observation = CreateObservation(new byte[] { 0x00, 0x01 }); + var metadata = SecsTraceTransportFaultRecord.Create( + Epoch, + observation, + SecsTraceTransportFaultCaptureOptions.MetadataOnly()); + var raw = SecsTraceTransportFaultRecord.Create( + Epoch, + observation, + SecsTraceTransportFaultCaptureOptions.RawPayload()); + var defaultCodec = new SecsTraceTransportFaultCodec(); + var payloadCodec = new SecsTraceTransportFaultCodec( + allowPayloadRecords: true); + + var metadataText = defaultCodec.Encode(new[] { metadata }); + var decoded = Assert.Single(defaultCodec.Decode(metadataText)); + var rawText = payloadCodec.Encode(new[] { raw }); + + Assert.Equal( + SecsTraceFaultSampleDataClassification.MetadataOnly, + decoded.DataClassification); + Assert.Equal(2, decoded.ObservedSnapshotLength); + Assert.True(decoded.Snapshot.IsEmpty); + Assert.Throws( + () => defaultCodec.Encode(new[] { raw })); + Assert.Throws( + () => defaultCodec.Decode(rawText)); + } + + [Fact] + public void Raw_capture_is_defensively_copied_and_round_trips_when_enabled() + { + var snapshot = new byte[] { 0x00, 0x00, 0x00, 0x0A, 0x20 }; + var observation = CreateObservation(snapshot); + snapshot[0] = 0xFF; + var record = SecsTraceTransportFaultRecord.Create( + Epoch, + observation, + SecsTraceTransportFaultCaptureOptions.RawPayload()); + var codec = new SecsTraceTransportFaultCodec( + allowPayloadRecords: true); + + var decoded = Assert.Single(codec.Decode(codec.Encode(new[] { record }))); + + Assert.Equal( + new byte[] { 0x00, 0x00, 0x00, 0x0A, 0x20 }, + record.Snapshot.ToArray()); + AssertRecordEqual(record, decoded); + } + + [Fact] + public void Capture_factory_rejects_null_oversize_and_outside_ranges() + { + var observation = CreateObservation(new byte[4]); + + Assert.Throws(() => + SecsTraceTransportFaultRecord.Create( + Epoch, + null!, + SecsTraceTransportFaultCaptureOptions.MetadataOnly())); + Assert.Throws(() => + SecsTraceTransportFaultRecord.Create(Epoch, observation, null!)); + Assert.Throws(() => + SecsTraceTransportFaultRecord.Create( + Epoch, + observation, + SecsTraceTransportFaultCaptureOptions.RawPayload( + maxSnapshotBytes: 3))); + Assert.Throws(() => + SecsTraceTransportFaultRecord.Create( + Epoch, + observation, + SecsTraceTransportFaultCaptureOptions.RedactedPayload( + new[] { new SecsTraceByteRedactionRange(3, 2) }))); + } + + [Fact] + public void Codec_rejects_malformed_fields_and_unverifiable_redaction() + { + const string valid = + "SecsFrame-TransportFaultTrace/1\n" + + "TransportFault 1970-01-01T00:00:00.0000000Z IncompleteFrameTimeout Connected 17 RedactedPayload 4 1:2 41000044\n"; + var codec = new SecsTraceTransportFaultCodec( + allowPayloadRecords: true); + + Assert.Throws(() => codec.Decode(string.Empty)); + Assert.Throws(() => codec.Decode(valid.Replace("TransportFaultTrace/1", "TransportFaultTrace/2"))); + Assert.Throws(() => codec.Decode(valid.Replace("IncompleteFrameTimeout", "DecodeFailed"))); + Assert.Throws(() => codec.Decode(valid.Replace("Connected", "connected"))); + Assert.Throws(() => codec.Decode(valid.Replace(" 17 ", " 01 "))); + Assert.Throws(() => codec.Decode(valid.Replace("RedactedPayload", "Payload"))); + Assert.Throws(() => codec.Decode(valid.Replace(" 4 1:2 ", " 04 1:2 "))); + Assert.Throws(() => codec.Decode(valid.Replace("1:2", "01:2"))); + Assert.Throws(() => codec.Decode(valid.Replace("1:2 41000044", "3:1,0:1 00000000"))); + Assert.Throws(() => codec.Decode(valid.Replace("41000044", "41424344"))); + Assert.Throws(() => codec.Decode(valid.Replace("41000044", "4100044"))); + Assert.Throws(() => codec.Decode(valid.Replace("41000044", "4100004a"))); + Assert.Throws(() => codec.Decode(valid.Replace("TransportFault ", "TransportFault "))); + } + + [Fact] + public void Options_observations_and_records_enforce_data_boundaries() + { + Assert.Throws(() => + SecsTraceTransportFaultCaptureOptions.RedactedPayload(null!)); + Assert.Throws(() => + SecsTraceTransportFaultCaptureOptions.RedactedPayload( + Array.Empty())); + Assert.Throws(() => + SecsTraceTransportFaultCaptureOptions.RedactedPayload( + new[] + { + new SecsTraceByteRedactionRange(0, 2), + new SecsTraceByteRedactionRange(1, 2), + })); + Assert.Throws(() => + SecsTraceTransportFaultCaptureOptions.RawPayload(0)); + Assert.Throws(() => + SecsTraceTransportFaultCaptureOptions.RawPayload( + HsmsTransportFaultObservation.MaxSnapshotBytes + 1)); + Assert.Throws(() => + new HsmsTransportFaultObservation( + HsmsTransportFaultKind.IncompleteFrameTimeout, + 1, + HsmsSessionState.Connected, + ReadOnlySpan.Empty)); + Assert.Throws(() => + new HsmsTransportFaultObservation( + HsmsTransportFaultKind.IncompleteFrameTimeout, + 1, + HsmsSessionState.Connected, + new byte[HsmsTransportFaultObservation.MaxSnapshotBytes + 1])); + Assert.Throws(() => + new SecsTraceTransportFaultRecord( + Epoch, + HsmsTransportFaultKind.IncompleteFrameTimeout, + HsmsSessionState.Connected, + 17, + SecsTraceFaultSampleDataClassification.RedactedPayload, + 2, + new byte[] { 1, 2 }, + new[] { new SecsTraceByteRedactionRange(0, 1) })); + Assert.Throws(() => + new SecsTraceTransportFaultRecord( + Epoch, + HsmsTransportFaultKind.IncompleteFrameTimeout, + HsmsSessionState.Connected, + 17, + SecsTraceFaultSampleDataClassification.RedactedPayload, + 4, + new byte[4], + new[] + { + new SecsTraceByteRedactionRange(2, 1), + new SecsTraceByteRedactionRange(0, 1), + })); + } + + [Fact] + public void Codec_enforces_record_text_and_snapshot_resource_limits() + { + var observation = CreateObservation(new byte[4]); + var metadata = SecsTraceTransportFaultRecord.Create( + Epoch, + observation, + SecsTraceTransportFaultCaptureOptions.MetadataOnly()); + var raw = SecsTraceTransportFaultRecord.Create( + Epoch, + observation, + SecsTraceTransportFaultCaptureOptions.RawPayload()); + var records = new[] { metadata, metadata }; + var metadataText = new SecsTraceTransportFaultCodec().Encode(records); + var rawText = new SecsTraceTransportFaultCodec( + allowPayloadRecords: true).Encode(new[] { raw }); + + Assert.Throws(() => + new SecsTraceTransportFaultCodec().Encode(null!)); + Assert.Throws(() => + new SecsTraceTransportFaultCodec().Decode(null!)); + Assert.Throws(() => + new SecsTraceTransportFaultCodec().Encode( + new SecsTraceTransportFaultRecord[] { metadata, null! })); + Assert.Throws(() => + new SecsTraceTransportFaultCodec(maxRecordCount: 1).Encode(records)); + Assert.Throws(() => + new SecsTraceTransportFaultCodec(maxRecordCount: 1).Decode(metadataText)); + Assert.Throws(() => + new SecsTraceTransportFaultCodec(maxTextLength: 20) + .Encode(Array.Empty())); + Assert.Throws(() => + new SecsTraceTransportFaultCodec(maxTextLength: 20) + .Decode(metadataText)); + Assert.Throws(() => + new SecsTraceTransportFaultCodec( + allowPayloadRecords: true, + maxSnapshotBytes: 3).Encode(new[] { raw })); + Assert.Throws(() => + new SecsTraceTransportFaultCodec( + allowPayloadRecords: true, + maxSnapshotBytes: 3).Decode(rawText)); + } + + private static HsmsTransportFaultObservation CreateObservation( + byte[] snapshot) + => new( + HsmsTransportFaultKind.IncompleteFrameTimeout, + 17, + HsmsSessionState.Connected, + snapshot); + + private static DateTimeOffset Epoch + => new(1970, 1, 1, 0, 0, 0, TimeSpan.Zero); + + private static void AssertRecordEqual( + SecsTraceTransportFaultRecord expected, + SecsTraceTransportFaultRecord actual) + { + Assert.Equal(expected.Timestamp, actual.Timestamp); + Assert.Equal(expected.Kind, actual.Kind); + Assert.Equal(expected.State, actual.State); + Assert.Equal(expected.TransportSessionId, actual.TransportSessionId); + Assert.Equal(expected.DataClassification, actual.DataClassification); + Assert.Equal( + expected.ObservedSnapshotLength, + actual.ObservedSnapshotLength); + Assert.Equal(expected.Snapshot.ToArray(), actual.Snapshot.ToArray()); + Assert.Equal(expected.RedactionRanges, actual.RedactionRanges); + } +} diff --git a/test/SecsFrame.Tests/StreamFrameHsmsTransportTests.cs b/test/SecsFrame.Tests/StreamFrameHsmsTransportTests.cs index 5624802..5432440 100644 --- a/test/SecsFrame.Tests/StreamFrameHsmsTransportTests.cs +++ b/test/SecsFrame.Tests/StreamFrameHsmsTransportTests.cs @@ -175,6 +175,42 @@ public async Task Native_incomplete_frame_timeout_reports_HSMS_T8_reason() await events.DisposeAsync().ConfigureAwait(true); } + [Fact] + public async Task Enabled_native_T8_observation_preserves_session_and_snapshot() + { + var connection = new FakeStreamConnection(); + await using var transport = CreateTransport( + connection, + enableT8FaultObservation: true); + 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 snapshot = new byte[] { 0x00, 0x00, 0x00, 0x0A, 0x00 }; + + connection.RaiseFrameError( + FrameErrorKind.IncompleteFrameTimeout, + snapshot); + snapshot[0] = 0xFF; + var fault = await NextAsync(events).ConfigureAwait(true); + connection.Reconnect(); + var closed = await NextAsync(events).ConfigureAwait(true); + + Assert.Equal(HsmsTransportEventKind.TransportFaultObserved, fault.Kind); + Assert.Equal(opened.SessionId, fault.SessionId); + Assert.Equal( + HsmsTransportFaultKind.IncompleteFrameTimeout, + fault.FaultKind); + Assert.Equal( + new byte[] { 0x00, 0x00, 0x00, 0x0A, 0x00 }, + fault.Snapshot.ToArray()); + Assert.Equal(HsmsTransportEventKind.SessionClosed, closed.Kind); + Assert.Equal(opened.SessionId, closed.SessionId); + Assert.IsType(closed.Error); + await events.DisposeAsync().ConfigureAwait(true); + } + [Fact] public async Task Other_native_frame_errors_do_not_become_T8_close_reasons() { @@ -500,8 +536,9 @@ public async Task Real_passive_reconnect_race_keeps_listener_available() } private static StreamFrameHsmsTransport CreateTransport( - FakeStreamConnection connection) - => new(connection); + FakeStreamConnection connection, + bool enableT8FaultObservation = false) + => new(connection, enableT8FaultObservation); private static StreamFrameHsmsTransport CreateControlledTransport( int port, @@ -845,10 +882,12 @@ public void RaiseState(ConnectionState state) ConnectionChanged?.Invoke(this, state); } - public void RaiseFrameError(FrameErrorKind kind) + public void RaiseFrameError( + FrameErrorKind kind, + ReadOnlyMemory bytes = default) => FrameError?.Invoke( this, - new FrameErrorEventArgs(kind, ReadOnlyMemory.Empty)); + new FrameErrorEventArgs(kind, bytes)); public void Emit(long sessionId, HsmsFrame frame) => _messages.Writer.TryWrite(new SessionMessage(sessionId, frame));