diff --git a/docs/reference/entity-protocol.md b/docs/reference/entity-protocol.md new file mode 100644 index 000000000..42773c1f0 --- /dev/null +++ b/docs/reference/entity-protocol.md @@ -0,0 +1,165 @@ +# Entity replication protocol + +`` converts owning [nested publications](kernel.md#nested-baselines-and-deltas) +to and from the canonical [`entity_sync.proto`](../../src/proto/proto/entity_sync.proto). +Enable `ARK_BUILD_PROTO=ON` and link `ark::kernel` and `ark::proto`. +Generated C++ messages use `ark::protobuf::entity_sync`; generated C# messages use +`Ark.Client.Protocol.EntitySync`. + + + +## Binding before state + +The application supplies a fixed `nested_stream_binding` before any remote +baseline: Schema ID, root GUID/incarnation, stream generation and an authorized +`nested_entity_layout`. Layouts include the field types of empty tables and the +individual child GUIDs allowed at each container. State messages never define or +expand this binding. Session authentication and projection authorization remain +application responsibilities. + +```cpp +#include + +using namespace ark::entity_sync; + +// trusted and initial come from the application's authorized capture path. +nested_wire::limits budget; +nested_publisher sender(trusted, initial, budget.stream); +nested_replica receiver(trusted, budget.stream); + +std::string body = nested_wire::encode(*sender.baseline(), trusted, budget); +auto baseline = nested_wire::decode_baseline(body, trusted, budget); +auto result = receiver.apply(baseline); +``` + +| API in `ark::entity_sync::nested_wire` | Result | +| --- | --- | +| `encode(native, proto_output, trusted, limits)` | Baseline/delta overloads validate and stage a generated message, then swap it into the output | +| `decode(proto_input, trusted, limits)` | Baseline/delta overloads return a new owning native publication | +| `encode(native, trusted, limits)` | Returns the Protobuf body as `std::string`; adds no transport header | +| `decode_baseline(bytes, trusted, limits)` / `decode_delta(bytes, trusted, limits)` | Parse a bounded `std::string_view` body and validate the resulting publication | + +The `limits` argument is optional. The caller chooses baseline versus delta from +its transport contract. The codec does not authenticate a session, manage a sender +queue or apply a publication. A returned publication acknowledges no delivery. + +## Values and structure + +`scalar_field.value` is an explicit oneof. Its selected alternative carries zero +and false values without losing presence. C++ variant indices and +`nested_scalar_type` ordinals are not a wire ABI. + +| Native value | Protobuf alternative | Constraint | +| --- | --- | --- | +| `bool` | `bool_value` | Explicit false is a present value | +| `int32_t` / `uint32_t` | `int32_value` (`sint32`) / `uint32_value` | Preserve signedness and width | +| `int64_t` / `uint64_t` | `int64_value` (`sint64`) / `uint64_value` | Preserve the full native range | +| `float` / `double` | `float_value` / `double_value` | Finite values only | +| `std::string` | `string_value` (`bytes`) | Raw bytes, including NUL and non-UTF-8; presentation validates its own text rules | +| `guid_value` | `guid_value` (`sint64`) | Signed 64-bit wire range and native `guid_t` range must both fit | + +Entity identities also use `sint64`; they must be nonzero. Scalar GUID values may +be zero. Default native GUIDs are signed 64-bit; 32-bit builds reject narrowing +on decode, and wider builds reject values that cannot fit the wire on encode. + +| Publication | Contents | +| --- | --- | +| `nested_baseline` | Generation, nonzero revision and complete `nested_snapshot` | +| `nested_snapshot` | Schema ID and root Entity snapshot | +| Entity snapshot | GUID/incarnation, class name, complete selected fields/tables/containers; selected empty tables and containers remain present | +| Table snapshot | Member index, row class and rows keyed by positive row number, each with incarnation and complete selected columns | +| `nested_delta` | Schema ID, generation, required revision, next revision, root patch and optional root-removal flag | +| Entity patch | GUID/incarnation and changed scalar, table-row and container records | + +Indexed collections use repeated records. Decoders reject duplicate field/table/ +container indices, row numbers and slots before installing any state. Indices are +class-local; table columns resolve in the row class. Complete roots reject repeated +Entity GUIDs, shared children and cycles. Unknown Protobuf fields may be ignored +within the same byte/recursion budgets; they grant no authorization. + +## Row and container changes + +| `row_delta` incarnations | Meaning | +| --- | --- | +| `required_incarnation == 0`, `incarnation > 0` | Add an absent row with every selected column | +| `required_incarnation > 0`, `incarnation == 0` | Remove the matching row; fields must be empty | +| Equal nonzero incarnations | Patch a nonempty subset of selected columns | +| New incarnation greater than the required nonzero incarnation | Replace the matching row with every selected column | + +Row identity includes its owner lifetime, table index, row number and incarnation. +Deleting and reusing a row number, even with identical values between flushes, +requires a new incarnation. The replica checks required identities and remembers +each table's incarnation watermark; retired row lifetimes cannot be reintroduced. + +`container_delta.update` selects exactly one representation: + +| Alternative | Meaning | +| --- | --- | +| `replacement` | Complete final authorized slot membership with owning child snapshots; an empty list clears the container | +| `patch` | Nonempty changed-slot list with recursive patches; each slot's existing child GUID/incarnation must match | + +The publisher chooses replacement whenever a slot, child GUID or incarnation +changes. Otherwise it recursively patches child values. Slot moves/swaps and +changes across multiple containers commit together within the root. Detaching a +child removes a relationship; it does not declare the source Entity destroyed. + +## Commit and recovery + +The codec validates binding, envelope, types, structure and publication budgets. +It cannot decide whether a delta fits a receiver's current revision, reconstructed +state or lifetime history; pass the result to `nested_replica::apply()`. + +| Receiver condition | Result | +| --- | --- | +| First complete matching baseline | Install once | +| New delta with `required_revision == current_revision` and `revision == required_revision + 1` | Validate a private complete candidate and its history, then commit the entire root once | +| Structurally valid delta at or below the current revision | `duplicate`; state never moves backwards | +| Older baseline / equal-revision identical active baseline | `duplicate`; a newer baseline is not an in-place reset | +| Gap, mismatched lifetime/binding or illegal state transition | `rejected`; hide state and invalidate that replica permanently | +| Ordered root tombstone | Remove state while retaining valid duplicate handling; later traffic cannot revive the root | + +The application closes an affected stream on malformed input, overflow, encoding +failure or admission/send rejection. Recovery uses a new connection, independently +trusted binding and complete baseline. Fixed projections do not support an in-place +generation reset or permission expansion. Atomicity covers one selected root; +there is no cross-root transaction, ACK or replay log. + +## Independent limits and failures + +| Budget | Default / scope | +| --- | --- | +| `limits.stream.snapshot` | Entity depth 8 with root depth 0; 1024 Entities, 65536 rows, 65536 selected fields/table/container members, 1 MiB of Schema IDs, class names and byte-string values; binding and complete state checked separately | +| Patch work | 65536 Entity/table/container/row operations, 65536 scalar values and 1 MiB of Schema IDs, class names and byte-string values, including replacement subtrees | +| `max_lifetime_entries` | 132096 remembered Entity/table/last-visible-row records; enforced at baseline validation and by the stateful publisher/replica | +| `max_message_bytes` | 4 MiB of encoded body, independently of native state and transport pending queues | +| `max_recursion_depth` | 128 Protobuf message levels, including record wrappers and unknown groups; distinct from Entity depth | + +Byte parsing uses `CodedInputStream` with the configured total-byte and recursion +limits. Generated-message overloads also check nesting and body size. Native +state/patch validation does not measure exact heap use: parsed messages, old state, +candidate state and copied history can coexist. Applications separately bound +pending frames/bytes/age, retained publications, simultaneous work and peak memory. +When encoding into an arena-owned output, staged allocations remain subject to +that arena's lifetime, including allocations from a failed attempt. + +| Failure | C++ behavior | +| --- | --- | +| Missing alternatives, duplicate indices, wrong binding/type, malformed bytes | Throw `std::invalid_argument` | +| State/patch/body limit or generated-message depth exceeded | Throw `std::length_error`; raw parser recursion failure reports `std::invalid_argument` | +| Allocation failure | Propagate `std::bad_alloc`; no partial generated output or native return is published | +| Stateful `apply()` rejects a decoded publication | Return `rejected` and discard active receiver state; previously retained immutable handles remain readable | + +Encoding stages on the destination's Protobuf arena and swaps only after complete +validation. Codec exceptions do not call or invalidate a replica; the application +handles stream closure and state disposal. A replica allocation failure separately +preserves its prior committed state/revision; this local guarantee does not create +a transport retry policy. + +`check-nested-wire` covers round trips, all scalar alternatives, binary values, +duplicate records, malformed structures, independent budgets, arena outputs and +stateful rejection with both default and 32-bit native GUIDs. These are local C++ +correctness checks; client and network integration require their own acceptance. diff --git a/docs/reference/entity-protocol_CN.md b/docs/reference/entity-protocol_CN.md new file mode 100644 index 000000000..0ab102105 --- /dev/null +++ b/docs/reference/entity-protocol_CN.md @@ -0,0 +1,145 @@ +# Entity 复制协议 + +`` 将拥有型[嵌套发布](kernel_CN.md#嵌套基线与增量) +与规范 [`entity_sync.proto`](../../src/proto/proto/entity_sync.proto) 相互转换。 +启用 `ARK_BUILD_PROTO=ON`，并链接 `ark::kernel` 和 `ark::proto`。 +生成的 C++ 消息使用 `ark::protobuf::entity_sync`，C# 消息使用 +`Ark.Client.Protocol.EntitySync`。 + + + +## 先绑定，再接收状态 + +应用在接收任何远端基线前提供固定的 `nested_stream_binding`：Schema ID、根 GUID／ +实例标记、流代次及已授权的 `nested_entity_layout`。布局包括空表的字段类型和每个 +容器逐个允许的子 GUID。状态消息不能定义或扩大该绑定。应用负责会话认证和投影授权。 + +```cpp +#include + +using namespace ark::entity_sync; + +// trusted 和 initial 来自应用已授权的捕获路径。 +nested_wire::limits budget; +nested_publisher sender(trusted, initial, budget.stream); +nested_replica receiver(trusted, budget.stream); + +std::string body = nested_wire::encode(*sender.baseline(), trusted, budget); +auto baseline = nested_wire::decode_baseline(body, trusted, budget); +auto result = receiver.apply(baseline); +``` + +| `ark::entity_sync::nested_wire` 中的 API | 结果 | +| --- | --- | +| `encode(native, proto_output, trusted, limits)` | 基线／增量重载校验并暂存生成的消息，完成后交换到输出 | +| `decode(proto_input, trusted, limits)` | 基线／增量重载返回新的拥有型原生发布 | +| `encode(native, trusted, limits)` | 返回 `std::string` Protobuf 消息体，不添加传输头 | +| `decode_baseline(bytes, trusted, limits)` / `decode_delta(bytes, trusted, limits)` | 解析有界的 `std::string_view` 消息体并校验发布 | + +`limits` 参数可省略。调用方按传输契约区分基线与增量。编解码器不认证会话、不管理 +发送队列，也不应用发布。返回发布不代表已经交付。 + +## 值与结构 + +`scalar_field.value` 使用显式 oneof；已选择的类型分支保留零值和 false 的存在性。 +C++ variant 索引与 `nested_scalar_type` 序号不是线协议 ABI。 + +| 原生值 | Protobuf 分支 | 约束 | +| --- | --- | --- | +| `bool` | `bool_value` | 显式 false 也是已存在的值 | +| `int32_t` / `uint32_t` | `int32_value`（`sint32`）/ `uint32_value` | 保留符号与宽度 | +| `int64_t` / `uint64_t` | `int64_value`（`sint64`）/ `uint64_value` | 保留完整原生范围 | +| `float` / `double` | `float_value` / `double_value` | 仅允许有限值 | +| `std::string` | `string_value`（`bytes`）| 原始字节，包括 NUL 和非 UTF-8；展示层自行校验文本规则 | +| `guid_value` | `guid_value`（`sint64`）| 同时满足有符号 64 位线协议范围及原生 `guid_t` 范围 | + +Entity 身份也使用 `sint64`，且不得为零；标量 GUID 值可以为零。默认原生 GUID 为 +有符号 64 位；32 位构建在解码时拒绝窄化，较宽构建在编码时拒绝线协议无法表示的值。 + +| 发布 | 内容 | +| --- | --- | +| `nested_baseline` | 流代次、非零修订号及完整 `nested_snapshot` | +| `nested_snapshot` | Schema ID 和根 Entity 快照 | +| Entity 快照 | GUID／实例标记、类名、完整的选定字段／表格／容器；保留已选择的空表和空容器 | +| 表格快照 | 成员索引、行类及正行号索引的行，每行包含实例标记和全部选定列 | +| `nested_delta` | Schema ID、流代次、基准修订、下一修订、根补丁及可选整根删除标志 | +| Entity 补丁 | GUID／实例标记及变化的标量、表格行和容器记录 | + +索引集合使用 repeated 记录。解码器在安装任何状态前拒绝重复字段／表格／容器索引、 +行号及槽位。索引在类内解析，表格列属于行类。完整根拒绝重复 Entity GUID、共享子 +引用和循环。未知 Protobuf 字段可以在同一字节／递归上限内被忽略，不授予任何权限。 + +## 行与容器变化 + +| `row_delta` 实例标记 | 含义 | +| --- | --- | +| `required_incarnation == 0`，`incarnation > 0` | 完整提供所有选定列，新增原先不存在的行 | +| `required_incarnation > 0`，`incarnation == 0` | 移除匹配的行，fields 必须为空 | +| 两者非零且相等 | 修改非空的选定列子集 | +| 新标记大于非零基准标记 | 完整提供所有选定列，替换匹配的行 | + +行身份包括所属 Entity 存续期、表格索引、行号及实例标记。删除后复用行号，即使两次 +刷新之间最终值相同，也必须分配新标记。接收端核对基准身份并记录各表的实例标记水位， +已退役行存续期不能重新引入。 + +`container_delta.update` 恰好选择一种表示： + +| 分支 | 含义 | +| --- | --- | +| `replacement` | 最终授权槽位的完整集合，附带拥有型子快照；空列表清空容器 | +| `patch` | 非空的变化槽位列表，附带递归补丁；槽位原有子 GUID／实例标记必须匹配 | + +槽位、子 GUID 或实例标记任一变化，发布端选择完整替换；否则递归修改子值。 +槽位移动／交换及多个容器的变化在同一根中一起提交。解除子关系不等于声明源 Entity 已销毁。 + +## 提交与恢复 + +编解码器校验绑定、消息封装、类型、结构及发布上限，无法判定增量是否匹配接收端的 +当前修订、重建状态和身份历史；须将结果传给 `nested_replica::apply()`。 + +| 接收端条件 | 结果 | +| --- | --- | +| 首个完整且匹配的基线 | 安装一次 | +| 新增量满足 `required_revision == current_revision` 且 `revision == required_revision + 1` | 验证私有完整候选和身份历史，再一次提交整根 | +| 结构合法且修订不高于当前修订的增量 | `duplicate`，状态不回退 | +| 旧基线／与当前有效状态相同的同修订基线 | `duplicate`；更新的基线不能用于原地重置 | +| 缺口、存续期／绑定不符或非法状态转换 | `rejected`，隐藏状态并永久使该接收端失效 | +| 有序整根删除标记 | 移除状态，保留有效的重复处理能力；后续流量不能复活根 | + +应用在非法输入、溢出、编码失败或入队／发送拒绝时关闭受影响流。恢复需要新连接、 +独立可信绑定及完整基线。固定投影不支持原地流代次重置或扩权。原子性仅覆盖一棵 +选定根，不提供跨根事务、ACK 或重放日志。 + +## 独立上限与失败处理 + +| 预算 | 默认值／范围 | +| --- | --- | +| `limits.stream.snapshot` | 根深度为 0，Entity 深度 8；1024 个 Entity、65536 行、65536 个选定字段／表格／容器成员，1 MiB 的 Schema ID／类名／字节串值；绑定与完整状态分别核算 | +| 补丁工作量 | 65536 个 Entity／表格／容器／行操作、65536 个标量值及 1 MiB 的 Schema ID／类名／字节串值，包括替代子树 | +| `max_lifetime_entries` | 132096 条已记录的 Entity／表格／最近可见行身份；基线校验及有状态发布端／接收端均执行上限 | +| `max_message_bytes` | 4 MiB 编码后消息体，与原生状态和传输待发队列分别限制 | +| `max_recursion_depth` | 128 层 Protobuf 消息，含记录包装层和未知 group，区别于 Entity 深度 | + +字节解析使用 `CodedInputStream` 设置总字节数和递归上限。生成消息重载也核对层数 +和消息体大小。原生状态／补丁校验不测量精确堆用量；已解析消息、旧状态、候选状态和 +历史副本可能同时存在。应用须分别限制待发帧数／字节数／年龄、保留发布、并行工作量 +和内存峰值。向 arena 所有的输出编码时，暂存分配受该 arena 的生命周期约束， +失败尝试的分配也包括在内。 + +| 失败 | C++ 行为 | +| --- | --- | +| 缺失类型分支、重复索引、绑定／类型错误或非法字节 | 抛出 `std::invalid_argument` | +| 状态／补丁／消息体超限或生成消息层数超限 | 抛出 `std::length_error`；原始字节解析的递归失败报告 `std::invalid_argument` | +| 分配失败 | 向上传播 `std::bad_alloc`，不发布部分生成输出或原生返回值 | +| 有状态 `apply()` 拒绝已解码发布 | 返回 `rejected` 并丢弃当前接收状态；此前保留的不可变句柄仍可读取 | + +编码在目标的 Protobuf arena 上暂存，完整校验后才交换。编解码异常不调用或使接收端 +失效，应用负责关闭流和释放状态。接收端自身的分配失败另有保持旧提交状态／修订的 +保证；这个本地保证不创建传输重试策略。 + +`check-nested-wire` 在默认及 32 位原生 GUID 构建下覆盖往返、全部标量类型、二进制值、 +重复记录、非法结构、独立预算、arena 输出及有状态拒绝。这些属于本地 C++ 正确性 +检查，客户端与网络集成仍需各自验收。 diff --git a/docs/reference/kernel.md b/docs/reference/kernel.md index 8f16089d9..53405234a 100644 --- a/docs/reference/kernel.md +++ b/docs/reference/kernel.md @@ -361,7 +361,7 @@ Entity incarnations are allocated monotonically within one loaded `entity_impl` **Breaking ABI change:** the new virtual Kernel method and Entity/row/table implementation layout changes require rebuilding all plugins, hosts and consumers together. Do not load old binaries with these headers. Source subclasses may keep the default method, which throws `std::logic_error`. The independent `check-kernel-plugin` host exercises the actual loaded virtual method and retains nested values after child destruction, table clearing and normal stop. -Capture provides the owning source values used by the [nested publisher/replica](#nested-baselines-and-deltas). Existing scalar APIs and room v1 messages are unchanged; no nested sender queue or codec is provided. Applications bind authorization and retain fixed projections for their stream. The [replication design](../project/entity-replication-design.md) records the selected consistency/recovery boundaries. +Capture provides the owning source values used by the [nested publisher/replica](#nested-baselines-and-deltas) and the [Entity protocol codec](entity-protocol.md). Applications bind authorization and retain fixed projections for their stream; sender queues remain application-owned. The [replication design](../project/entity-replication-design.md) records the selected consistency/recovery boundaries. ## Maps and data conversion @@ -566,9 +566,11 @@ peak and bound its own retained handles/queues. Use each publisher/replica on on owner thread. Release plugin-created shared handles before unloading their code; these handles do not certify arbitrary DSO unloading. -The header requires no Protobuf or transport. There is no nested codec, sender -queue, ACK/replay log, in-place reset, dynamic permission change, cross-root commit -or room v1 change. A returned publication acknowledges no delivery; an adapter +The stream header requires no Protobuf or transport. The separate +[Entity protocol codec](entity-protocol.md) maps publications to canonical Protobuf +messages against the same trusted binding. There is no sender queue, ACK/replay +log, in-place reset, dynamic permission change or cross-root commit. A returned +publication acknowledges no delivery; an adapter must separately bind sessions, bound encoded/pending work and close the affected stream on admission/send failure. `check-entity-stream` covers typed binding, recursive patches, row/Entity reuse, duplicates, gaps, atomic invalidation and diff --git a/docs/reference/kernel_CN.md b/docs/reference/kernel_CN.md index b0c9a9738..4c12f2884 100644 --- a/docs/reference/kernel_CN.md +++ b/docs/reference/kernel_CN.md @@ -355,7 +355,7 @@ Entity 实例标记在一个已加载的 `entity_impl` 实现副本内单调分 **ABI 不兼容变更：** 新 Kernel 虚方法及 Entity／row／table 实现布局变化要求同时重建全部插件、宿主和消费者，不能将旧二进制与新头文件混用。源代码子类可保留默认方法，该方法抛出 `std::logic_error`。独立的 `check-kernel-plugin` 宿主通过真实已加载虚方法捕获，并在子 Entity 销毁、表格清空和正常停止后读取保留值。 -捕获提供[嵌套发布端／接收端](#嵌套基线与增量)使用的拥有型源值。现有标量 API 和房间 v1 消息保持不变，不提供嵌套发送队列或编解码。应用负责绑定授权，并在流内保留固定投影。[复制设计](../project/entity-replication-design_CN.md)记录已选定的一致性／恢复边界。 +捕获提供[嵌套发布端／接收端](#嵌套基线与增量)及 [Entity 协议编解码](entity-protocol_CN.md)使用的拥有型源值。应用负责绑定授权，并在流内保留固定投影；发送队列仍由应用管理。[复制设计](../project/entity-replication-design_CN.md)记录已选定的一致性／恢复边界。 ## 地图与数据转换 @@ -536,8 +536,9 @@ if (auto delta = sender.update(next_authorized_capture)) 旧已提交状态、候选、数据包和历史副本；应用须计入组合峰值，并限制自行保留的句柄／队列。 每个发布端／接收端只在一个所属线程使用。卸载代码前须释放插件创建的共享句柄；句柄不证明任意 DSO 卸载安全。 -该头文件不依赖 Protobuf 或传输。不提供嵌套编解码、发送队列、ACK／重放日志、原地重置、 -动态权限变更或跨根提交，不改变房间 v1。返回发布不代表交付；适配器须单独绑定会话、 +该流头文件不依赖 Protobuf 或传输。独立的 [Entity 协议编解码](entity-protocol_CN.md) +按同一可信绑定将发布映射为规范 Protobuf 消息。不提供发送队列、ACK／重放日志、原地重置、 +动态权限变更或跨根提交。返回发布不代表交付；适配器须单独绑定会话、 限制编码后／待发工作，并在接收或发送失败时关闭受影响流。 `check-entity-stream` 覆盖类型绑定、递归补丁、行／Entity 复用、重复、缺口、原子失效及上限。 其依赖 `check-nested-stream-allocation` 对路径中的每个分配点注入失败。 diff --git a/docs/site/src/navigation.ts b/docs/site/src/navigation.ts index 36b98f437..88a95fe2b 100644 --- a/docs/site/src/navigation.ts +++ b/docs/site/src/navigation.ts @@ -156,6 +156,11 @@ const tree: Entry[] = [ "Kernel plugin", "Kernel 插件" ], + [ + "reference/entity-protocol", + "Entity protocol", + "Entity 协议" + ], [ "reference/network", "Network plugin", diff --git a/src/plugins/CMakeLists.txt b/src/plugins/CMakeLists.txt index 174e99dd6..0330f7571 100755 --- a/src/plugins/CMakeLists.txt +++ b/src/plugins/CMakeLists.txt @@ -79,6 +79,7 @@ set(ark_kernel_public_headers "${CMAKE_CURRENT_SOURCE_DIR}/kernel/entity_nested_snapshot.hpp" "${CMAKE_CURRENT_SOURCE_DIR}/kernel/entity_stream.hpp" "${CMAKE_CURRENT_SOURCE_DIR}/kernel/nested_stream.hpp" + "${CMAKE_CURRENT_SOURCE_DIR}/kernel/nested_wire.hpp" "${CMAKE_CURRENT_SOURCE_DIR}/kernel/class_call_back_manager.hpp" "${CMAKE_CURRENT_SOURCE_DIR}/kernel/class_meta.hpp" "${CMAKE_CURRENT_SOURCE_DIR}/kernel/class_meta_manager.hpp" diff --git a/src/plugins/kernel/nested_wire.hpp b/src/plugins/kernel/nested_wire.hpp new file mode 100644 index 000000000..5930fd420 --- /dev/null +++ b/src/plugins/kernel/nested_wire.hpp @@ -0,0 +1,520 @@ +// Copyright (c) 2026 OpenArkStudio. Licensed under Apache-2.0. See LICENSE. +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +// A caller may already have Win32's object-like GetMessage macro in scope. +// Keep Protobuf declarations independent of that macro's include order. +#pragma push_macro("GetMessage") +#undef GetMessage +#include +#include +#include +#pragma pop_macro("GetMessage") + +#include "ark/kernel/nested_stream.hpp" + +// nested_stream can introduce Win32 macros through ark/platform.hpp after the +// Protobuf declarations above. Protect generated headers and our implementation +// separately, then restore the caller's macro state at the end of this header. +#pragma push_macro("GetMessage") +#undef GetMessage +#include "ark/proto/entity_sync.pb.h" + +namespace ark::entity_sync::nested_wire { + +namespace proto = protobuf::entity_sync; + +struct limits +{ + nested_stream_limits stream; + // Independent of native state/patch budgets and transport pending budgets. + std::size_t max_message_bytes{4 * 1024 * 1024}; + // Protobuf message nesting, including record wrappers, not Entity depth. + int max_recursion_depth{128}; +}; + +namespace detail { + +template +inline std::int64_t encode_guid(guid_type input) { + if constexpr (std::numeric_limits::digits > std::numeric_limits::digits) + { + if (input < std::numeric_limits::min() || input > std::numeric_limits::max()) + { + throw std::invalid_argument("Nested GUID exceeds signed 64-bit wire width"); + } + } + return static_cast(input); +} + +inline guid_t decode_guid(std::int64_t input) { + if (input < std::numeric_limits::min() || input > std::numeric_limits::max()) + { + throw std::invalid_argument("Nested GUID exceeds native width"); + } + return static_cast(input); +} + +inline void check_limits(const limits& budget) { + if (!budget.max_message_bytes || + budget.max_message_bytes > static_cast(std::numeric_limits::max()) || + budget.max_recursion_depth <= 0) + { + throw std::invalid_argument("Invalid nested wire byte or recursion limit"); + } +} + +inline void check_unknown( + const google::protobuf::UnknownFieldSet& input, int depth, const limits& budget, std::size_t& records) { + if (depth > budget.max_recursion_depth) + { + throw std::length_error("Nested wire unknown-group recursion limit exceeded"); + } + for (int index = 0; index < input.field_count(); ++index) + { + if (!records) + { + throw std::length_error("Nested wire record limit exceeded"); + } + --records; + const auto& field = input.field(index); + if (field.type() == google::protobuf::UnknownField::TYPE_GROUP) + { + check_unknown(field.group(), depth + 1, budget, records); + } + } +} + +// Check a caller-supplied generated message before recursively computing its +// size or constructing native maps. Unknown fields remain forward compatible; +// their size and any unknown groups still obey the independent wire budgets. +inline void check_depth(const google::protobuf::Message& input, int depth, const limits& budget, std::size_t& records) { + if (depth > budget.max_recursion_depth || !records) + { + throw std::length_error("Nested wire recursion or record limit exceeded"); + } + --records; + const auto* reflection = input.GetReflection(); + check_unknown(reflection->GetUnknownFields(input), depth, budget, records); + std::vector fields; + reflection->ListFields(input, &fields); + for (const auto* field : fields) + { + if (field->cpp_type() != google::protobuf::FieldDescriptor::CPPTYPE_MESSAGE) + { + continue; + } + if (field->is_repeated()) + { + const auto count = reflection->FieldSize(input, field); + for (int index = 0; index < count; ++index) + { + check_depth(reflection->GetRepeatedMessage(input, field, index), depth + 1, budget, records); + } + } + else + { + check_depth(reflection->GetMessage(input, field), depth + 1, budget, records); + } + } +} + +inline void check_message(const google::protobuf::Message& input, const limits& budget) { + check_limits(budget); + // Each nested record needs at least a tag and a length byte on the wire. + auto records = budget.max_message_bytes; + check_depth(input, 0, budget, records); + if (input.ByteSizeLong() > budget.max_message_bytes) + { + throw std::length_error("Nested wire message byte limit exceeded"); + } +} + +inline void validate(const nested_baseline& input, const nested_stream_binding& trusted, const limits& budget) { + nested_detail::binding(trusted, budget.stream); + nested_detail::require(input.generation == trusted.generation && input.revision); + nested_detail::snapshot(input.snapshot, trusted, budget.stream); + nested_detail::lifetime_history history; + nested_detail::remember(input.snapshot.root, history, budget.stream.max_lifetime_entries); +} + +inline void validate(const nested_delta& input, const nested_stream_binding& trusted, const limits& budget) { + nested_detail::binding(trusted, budget.stream); + nested_detail::delta(input, trusted, budget.stream); +} + +inline void encode_fields(const std::map& input, + google::protobuf::RepeatedPtrField& output) { + for (const auto& [index, value] : input) + { + auto& field = *output.Add(); + field.set_index(index); + std::visit( + [&](const auto& item) { + using type = std::decay_t; + if constexpr (std::is_same_v) + { + field.set_bool_value(item); + } + else if constexpr (std::is_same_v) + { + field.set_int32_value(item); + } + else if constexpr (std::is_same_v) + { + field.set_uint32_value(item); + } + else if constexpr (std::is_same_v) + { + field.set_int64_value(item); + } + else if constexpr (std::is_same_v) + { + field.set_uint64_value(item); + } + else if constexpr (std::is_same_v) + { + field.set_float_value(item); + } + else if constexpr (std::is_same_v) + { + field.set_double_value(item); + } + else if constexpr (std::is_same_v) + { + field.set_string_value(item); + } + else if constexpr (std::is_same_v) + { + field.set_guid_value(encode_guid(item.value)); + } + }, + value); + } +} + +inline scalar_value decode_field(const proto::scalar_field& input) { + switch (input.value_case()) + { + case proto::scalar_field::kBoolValue: + return input.bool_value(); + case proto::scalar_field::kInt32Value: + return input.int32_value(); + case proto::scalar_field::kUint32Value: + return input.uint32_value(); + case proto::scalar_field::kInt64Value: + return input.int64_value(); + case proto::scalar_field::kUint64Value: + return input.uint64_value(); + case proto::scalar_field::kFloatValue: + return input.float_value(); + case proto::scalar_field::kDoubleValue: + return input.double_value(); + case proto::scalar_field::kStringValue: + return input.string_value(); + case proto::scalar_field::kGuidValue: + return guid_value{decode_guid(input.guid_value())}; + default: + throw std::invalid_argument("Missing nested scalar alternative"); + } +} + +inline std::map decode_fields( + const google::protobuf::RepeatedPtrField& input) { + std::map output; + for (const auto& field : input) + { + nested_detail::require( + field.index() && + output.emplace(field.index(), entity_sync::detail::direct_value{[&] { return decode_field(field); }}) + .second); + } + return output; +} + +inline void encode_entity(const nested_entity_snapshot& input, proto::entity_snapshot& output) { + output.set_entity_id(encode_guid(input.scalars.entity_id)); + output.set_incarnation(input.incarnation); + output.set_class_name(input.scalars.class_name); + encode_fields(input.scalars.fields, *output.mutable_fields()); + for (const auto& [index, table] : input.tables) + { + auto& result = *output.add_tables(); + result.set_index(index); + result.set_class_name(table.class_name); + for (const auto& [number, row] : table.rows) + { + auto& record = *result.add_rows(); + record.set_number(number); + record.set_incarnation(row.incarnation); + encode_fields(row.fields, *record.mutable_fields()); + } + } + for (const auto& [index, children] : input.containers) + { + auto& result = *output.add_containers(); + result.set_index(index); + for (const auto& [slot, child] : children) + { + auto& record = *result.add_slots(); + record.set_slot(slot); + encode_entity(child, *record.mutable_state()); + } + } +} + +inline nested_entity_snapshot decode_entity(const proto::entity_snapshot& input) { + std::map tables; + for (const auto& table : input.tables()) + { + std::map rows; + for (const auto& row : table.rows()) + { + nested_detail::require(row.number() && rows.emplace(row.number(), entity_sync::detail::direct_value{[&] { + return nested_row_snapshot{ + row.incarnation(), decode_fields(row.fields())}; + }}) + .second); + } + nested_detail::require(table.index() && tables + .emplace(table.index(), entity_sync::detail::direct_value{[&] { + return nested_table_snapshot{ + table.class_name(), std::move(rows)}; + }}) + .second); + } + std::map> containers; + for (const auto& container : input.containers()) + { + std::map children; + for (const auto& slot : container.slots()) + { + nested_detail::require(slot.slot() && slot.has_state() && + children + .emplace(slot.slot(), entity_sync::detail::direct_value{[&] { + return decode_entity(slot.state()); + }}) + .second); + } + nested_detail::require(container.index() && containers.emplace(container.index(), std::move(children)).second); + } + return {input.incarnation(), {decode_guid(input.entity_id()), input.class_name(), decode_fields(input.fields())}, + std::move(tables), std::move(containers)}; +} + +inline void encode_change(const nested_entity_delta& input, proto::entity_delta& output) { + output.set_entity_id(encode_guid(input.entity_id)); + output.set_incarnation(input.incarnation); + encode_fields(input.fields, *output.mutable_fields()); + for (const auto& [index, rows] : input.tables) + { + auto& result = *output.add_tables(); + result.set_index(index); + for (const auto& [number, row] : rows) + { + auto& record = *result.add_rows(); + record.set_number(number); + record.set_required_incarnation(row.required_incarnation); + record.set_incarnation(row.incarnation); + encode_fields(row.fields, *record.mutable_fields()); + } + } + for (const auto& [index, change] : input.containers) + { + auto& result = *output.add_containers(); + result.set_index(index); + if (change.replace) + { + auto& replacement = *result.mutable_replacement(); + for (const auto& [slot, child] : change.slots) + { + auto& record = *replacement.add_slots(); + record.set_slot(slot); + encode_entity(child, *record.mutable_state()); + } + } + else + { + auto& patch = *result.mutable_patch(); + for (const auto& [slot, child] : change.changed) + { + auto& record = *patch.add_changed(); + record.set_slot(slot); + encode_change(child, *record.mutable_change()); + } + } + } +} + +inline nested_entity_delta decode_change(const proto::entity_delta& input) { + std::map> tables; + for (const auto& table : input.tables()) + { + std::map rows; + for (const auto& row : table.rows()) + { + nested_detail::require(row.number() && rows.emplace(row.number(), entity_sync::detail::direct_value{[&] { + return nested_row_delta{row.required_incarnation(), + row.incarnation(), decode_fields(row.fields())}; + }}) + .second); + } + nested_detail::require(table.index() && tables.emplace(table.index(), std::move(rows)).second); + } + std::map containers; + for (const auto& container : input.containers()) + { + nested_container_delta change; + if (container.has_replacement()) + { + change.replace = true; + for (const auto& slot : container.replacement().slots()) + { + nested_detail::require(slot.slot() && slot.has_state() && + change.slots + .emplace(slot.slot(), entity_sync::detail::direct_value{[&] { + return decode_entity(slot.state()); + }}) + .second); + } + } + else + { + nested_detail::require(container.has_patch()); + for (const auto& slot : container.patch().changed()) + { + nested_detail::require(slot.slot() && slot.has_change() && + change.changed + .emplace(slot.slot(), entity_sync::detail::direct_value{[&] { + return decode_change(slot.change()); + }}) + .second); + } + } + nested_detail::require(container.index() && containers.emplace(container.index(), std::move(change)).second); + } + return {decode_guid(input.entity_id()), input.incarnation(), decode_fields(input.fields()), std::move(tables), + std::move(containers)}; +} + +template +inline void parse(std::string_view input, message_type& output, const limits& budget) { + check_limits(budget); + if (input.size() > budget.max_message_bytes) + { + throw std::length_error("Nested wire message byte limit exceeded"); + } + google::protobuf::io::CodedInputStream stream( + reinterpret_cast(input.data()), static_cast(input.size())); + stream.SetTotalBytesLimit(static_cast(budget.max_message_bytes)); + stream.SetRecursionLimit(budget.max_recursion_depth); + if (!output.ParseFromCodedStream(&stream) || !stream.ConsumedEntireMessage()) + { + throw std::invalid_argument("Malformed or excessively nested Entity message"); + } +} + +} // namespace detail + +// Encoding validates the independent binding and stages the complete result. +// A codec validates packet structure; ordering and lifetime history belong to +// nested_replica::apply(). A successful encode is not a delivery acknowledgement. +inline void encode(const nested_baseline& input, proto::nested_baseline& output, const nested_stream_binding& trusted, + const limits& budget = {}) { + detail::check_limits(budget); + detail::validate(input, trusted, budget); + // Sharing the destination arena makes the final Swap allocation-free. + auto* candidate = google::protobuf::Arena::Create(output.GetArena()); + std::unique_ptr owner(output.GetArena() ? nullptr : candidate); + candidate->set_generation(input.generation); + candidate->set_revision(input.revision); + auto& snapshot = *candidate->mutable_snapshot(); + snapshot.set_schema_id(input.snapshot.schema_id); + detail::encode_entity(input.snapshot.root, *snapshot.mutable_root()); + detail::check_message(*candidate, budget); + output.Swap(candidate); +} + +inline void encode(const nested_delta& input, proto::nested_delta& output, const nested_stream_binding& trusted, + const limits& budget = {}) { + detail::check_limits(budget); + detail::validate(input, trusted, budget); + auto* candidate = google::protobuf::Arena::Create(output.GetArena()); + std::unique_ptr owner(output.GetArena() ? nullptr : candidate); + candidate->set_schema_id(input.schema_id); + candidate->set_generation(input.generation); + candidate->set_required_revision(input.required_revision); + candidate->set_revision(input.revision); + candidate->set_removed(input.removed); + detail::encode_change(input.root, *candidate->mutable_root()); + detail::check_message(*candidate, budget); + output.Swap(candidate); +} + +inline nested_baseline decode( + const proto::nested_baseline& input, const nested_stream_binding& trusted, const limits& budget = {}) { + detail::check_message(input, budget); + nested_detail::binding(trusted, budget.stream); + nested_detail::require(input.has_snapshot() && input.snapshot().has_root()); + nested_baseline candidate{input.generation(), input.revision(), + {input.snapshot().schema_id(), detail::decode_entity(input.snapshot().root())}}; + detail::validate(candidate, trusted, budget); + return {candidate.generation, candidate.revision, nested_detail::take_snapshot(candidate.snapshot)}; +} + +inline nested_delta decode( + const proto::nested_delta& input, const nested_stream_binding& trusted, const limits& budget = {}) { + detail::check_message(input, budget); + nested_detail::binding(trusted, budget.stream); + nested_detail::require(input.has_root()); + nested_delta candidate{input.schema_id(), input.generation(), input.required_revision(), input.revision(), + detail::decode_change(input.root()), input.removed()}; + detail::validate(candidate, trusted, budget); + return {candidate.schema_id, candidate.generation, candidate.required_revision, candidate.revision, + std::move(candidate.root), candidate.removed}; +} + +template +inline std::string encode( + const publication_type& input, const nested_stream_binding& trusted, const limits& budget = {}) { + using message_type = std::conditional_t, proto::nested_baseline, + proto::nested_delta>; + static_assert(std::is_same_v || std::is_same_v); + message_type message; + encode(input, message, trusted, budget); + std::string output; + if (!message.SerializeToString(&output)) + { + throw std::invalid_argument("Cannot serialize nested Entity message"); + } + return output; +} + +inline nested_baseline decode_baseline( + std::string_view input, const nested_stream_binding& trusted, const limits& budget = {}) { + proto::nested_baseline message; + detail::parse(input, message, budget); + return decode(message, trusted, budget); +} + +inline nested_delta decode_delta( + std::string_view input, const nested_stream_binding& trusted, const limits& budget = {}) { + proto::nested_delta message; + detail::parse(input, message, budget); + return decode(message, trusted, budget); +} + +} // namespace ark::entity_sync::nested_wire + +#pragma pop_macro("GetMessage") diff --git a/src/proto/proto/entity_sync.proto b/src/proto/proto/entity_sync.proto new file mode 100644 index 000000000..011b79c39 --- /dev/null +++ b/src/proto/proto/entity_sync.proto @@ -0,0 +1,115 @@ +syntax = "proto3"; +package ark.protobuf.entity_sync; +option csharp_namespace = "Ark.Client.Protocol.EntitySync"; + +// Indexed collections are repeated records, not maps. Decoders reject duplicate +// indices before installing any state. A schema and authorized projection must +// be trusted independently of these messages. +message scalar_field { + uint32 index = 1; + oneof value { + bool bool_value = 2; + sint32 int32_value = 3; + uint32 uint32_value = 4; + sint64 int64_value = 5; + uint64 uint64_value = 6; + float float_value = 7; + double double_value = 8; + bytes string_value = 9; + sint64 guid_value = 10; + } +} + +message row_snapshot { + uint32 number = 1; + uint64 incarnation = 2; + repeated scalar_field fields = 3; +} + +message table_snapshot { + uint32 index = 1; + string class_name = 2; + repeated row_snapshot rows = 3; +} + +message entity_slot { + uint32 slot = 1; + entity_snapshot state = 2; +} + +message container_snapshot { + uint32 index = 1; + repeated entity_slot slots = 2; +} + +message entity_snapshot { + sint64 entity_id = 1; + uint64 incarnation = 2; + string class_name = 3; + repeated scalar_field fields = 4; + repeated table_snapshot tables = 5; + repeated container_snapshot containers = 6; +} + +message nested_snapshot { + string schema_id = 1; + entity_snapshot root = 2; +} + +message nested_baseline { + uint64 generation = 1; + uint64 revision = 2; + nested_snapshot snapshot = 3; +} + +// required=0 adds a complete row; incarnation=0 removes a row; equal nonzero +// incarnations patch columns; a newer incarnation completely replaces a row. +message row_delta { + uint32 number = 1; + uint64 required_incarnation = 2; + uint64 incarnation = 3; + repeated scalar_field fields = 4; +} + +message table_delta { + uint32 index = 1; + repeated row_delta rows = 2; +} + +message entity_change { + uint32 slot = 1; + entity_delta change = 2; +} + +message container_replacement { + repeated entity_slot slots = 1; +} + +message container_patch { + repeated entity_change changed = 1; +} + +message container_delta { + uint32 index = 1; + oneof update { + container_replacement replacement = 2; + container_patch patch = 3; + } +} + +message entity_delta { + sint64 entity_id = 1; + uint64 incarnation = 2; + repeated scalar_field fields = 3; + repeated table_delta tables = 4; + repeated container_delta containers = 5; +} + +message nested_delta { + string schema_id = 1; + uint64 generation = 2; + uint64 required_revision = 3; + uint64 revision = 4; + entity_delta root = 5; + bool removed = 6; +} diff --git a/tests/app/business_runtime_plugin.cpp b/tests/app/business_runtime_plugin.cpp index c8778afc4..263ec1b1b 100644 --- a/tests/app/business_runtime_plugin.cpp +++ b/tests/app/business_runtime_plugin.cpp @@ -173,7 +173,7 @@ class probe_module final : public ark::module trace("ready"); if (mode_ == "boundary_readiness") { - std::this_thread::sleep_for(std::chrono::milliseconds(100)); + expire_shutdown_deadline(); } trace("ready-return"); return true; @@ -206,7 +206,7 @@ class probe_module final : public ark::module cleanup_started_ = true; if (mode_ == "boundary_pre_shut") { - std::this_thread::sleep_for(std::chrono::milliseconds(100)); + expire_shutdown_deadline(); } trace("pre-shut-return"); return true; @@ -223,7 +223,7 @@ class probe_module final : public ark::module trace("shut"); if (mode_ == "boundary_shut") { - std::this_thread::sleep_for(std::chrono::milliseconds(100)); + expire_shutdown_deadline(); trace("shut-return"); } if (mode_ == "shut_throw") @@ -236,6 +236,19 @@ class probe_module final : public ark::module private: using callback = std::function; + void expire_shutdown_deadline() const { + // Tighten the active stop only after reaching the hook under test. + // Earlier tracing or scheduling must not choose a different boundary. + // Reentrant stop cannot clean up this hook's active stack; its caller + // must notice expiry after the successful hook returns. + if (manager_.stop_until(manager_.get_now_time(), std::chrono::steady_clock::time_point::min()) || + manager_.get_shutdown_state() != ark::plugin_manager::shutdown_state::stopping) + { + throw std::runtime_error("Shutdown deadline entered the active lifecycle hook"); + } + trace("deadline-expired"); + } + std::uint64_t start(callback complete) { if (pending_) { diff --git a/tests/app/shutdown_deadlines.cmake b/tests/app/shutdown_deadlines.cmake index da40fc478..a1699bbfd 100644 --- a/tests/app/shutdown_deadlines.cmake +++ b/tests/app/shutdown_deadlines.cmake @@ -18,10 +18,13 @@ foreach(mode IN LISTS modes) file(WRITE "${directory}/application.toml" "[application]\nplugin_library_dir = '${plugin_directory}/'\nplugin_config_dir = './'\n" "[[plugins]]\nname = 'ark_business_runtime_probe'\nconfig = '${mode}'\n") + # The selected hook tightens stop_until() to an expired deadline on its + # active stack. A short launcher budget could expire during earlier trace + # writes or scheduling and select the wrong boundary. CLI elapsed-time + # behavior is covered separately by business_runtime.cmake. execute_process( COMMAND "${PROGRAM}" --busid=1.0.1.1 --name=shutdown-deadline-test "--app_conf=${directory}/application.toml" - --shutdown_timeout_ms=50 WORKING_DIRECTORY "${directory}" INPUT_FILE "${directory}/exit.txt" RESULT_VARIABLE result OUTPUT_VARIABLE output ERROR_VARIABLE errors TIMEOUT 5) file(WRITE "${directory}/output.log" "${output}\n${errors}") @@ -31,15 +34,15 @@ foreach(mode IN LISTS modes) endif() file(STRINGS "${directory}/business-runtime.trace" events) if(mode STREQUAL "boundary_readiness") - set(expected request-stop ready ready-return) + set(expected request-stop ready deadline-expired ready-return) set(forbidden pre-shut pre-shut-return shut shut-return) set(phase ready_to_shut) elseif(mode STREQUAL "boundary_pre_shut") - set(expected request-stop ready ready-return pre-shut pre-shut-return) + set(expected request-stop ready ready-return pre-shut deadline-expired pre-shut-return) set(forbidden shut shut-return) set(phase pre_shut) else() - set(expected request-stop ready ready-return pre-shut pre-shut-return shut shut-return) + set(expected request-stop ready ready-return pre-shut pre-shut-return shut deadline-expired shut-return) set(forbidden) set(phase shut) endif() @@ -66,8 +69,8 @@ foreach(mode IN LISTS modes) string(REGEX MATCHALL "Shutdown (failed at |wait timed out |update failed)" diagnostics "${output}\n${errors}") list(LENGTH diagnostics count) if(NOT count EQUAL 1 OR NOT "${output}\n${errors}" MATCHES - "Shutdown wait timed out after [0-9]+ ms.*limit=50 ms.*pending=0.*phase=${phase}.*process and resources are retained") + "Shutdown wait timed out after [0-9]+ ms.*limit=0 ms.*pending=0.*phase=${phase}.*process and resources are retained") message(FATAL_ERROR "${mode}: expected one phase-specific timeout diagnostic\n${output}\n${errors}") endif() - message(STATUS "${mode}: successful slow hook returned; no later cleanup or successful exit") + message(STATUS "${mode}: successful hook returned after controlled expiry; no later cleanup or successful exit") endforeach() diff --git a/tests/consumer/kernel/CMakeLists.txt b/tests/consumer/kernel/CMakeLists.txt index 6e9e3f66c..1425acc62 100644 --- a/tests/consumer/kernel/CMakeLists.txt +++ b/tests/consumer/kernel/CMakeLists.txt @@ -36,6 +36,7 @@ set(kernel_headers entity_nested_snapshot.hpp entity_stream.hpp nested_stream.hpp + nested_wire.hpp event_manager.hpp event_manager_impl.hpp kernel_module.hpp @@ -58,7 +59,7 @@ set(kernel_headers ) if(NOT ARK_BUILD_PROTO) # These existing headers require ark/proto.hpp; value/template headers do not. - list(REMOVE_ITEM kernel_headers kernel_module.hpp kernel_module_impl.hpp map_module_impl.hpp) + list(REMOVE_ITEM kernel_headers kernel_module.hpp kernel_module_impl.hpp map_module_impl.hpp nested_wire.hpp) endif() set(header_checks) foreach(header IN LISTS kernel_headers) diff --git a/tests/core/CMakeLists.txt b/tests/core/CMakeLists.txt index b0a78b34a..990b3c2f6 100644 --- a/tests/core/CMakeLists.txt +++ b/tests/core/CMakeLists.txt @@ -159,6 +159,19 @@ add_custom_target(check-entity-stream COMMAND $ DEPENDS ark_entity_stream_tests USES_TERMINAL) add_dependencies(check-core check-entity-stream) +if(ARK_BUILD_PROTO) + set(ark_nested_wire_test_sources nested_wire.cpp nested_wire_macro.cpp nested_wire_macro_first.cpp) + add_executable(ark_nested_wire_tests $ ${ark_nested_wire_test_sources}) + set_property(TARGET ark_nested_wire_tests PROPERTY LINK_LIBRARIES ark::kernel ark::proto) + add_executable(ark_nested_wire_guid32_tests main.cpp ${ark_nested_wire_test_sources}) + set_property(TARGET ark_nested_wire_guid32_tests PROPERTY LINK_LIBRARIES ark::kernel ark::proto) + target_compile_definitions(ark_nested_wire_guid32_tests PRIVATE ARK_USE_GUID_32) + add_custom_target(check-nested-wire + COMMAND $ + COMMAND $ + DEPENDS ark_nested_wire_tests ark_nested_wire_guid32_tests USES_TERMINAL VERBATIM) + add_dependencies(check-core check-nested-wire) +endif() add_executable(ark_scalar_stream_allocation_tests $ entity_scalar_stream_allocation.cpp) set_property(TARGET ark_scalar_stream_allocation_tests PROPERTY LINK_LIBRARIES ark::kernel) add_custom_target(check-scalar-stream-allocation diff --git a/tests/core/nested_wire.cpp b/tests/core/nested_wire.cpp new file mode 100644 index 000000000..6a120f5a2 --- /dev/null +++ b/tests/core/nested_wire.cpp @@ -0,0 +1,330 @@ +// Copyright (c) 2026 OpenArkStudio. Licensed under Apache-2.0. See LICENSE. +#include +#include + +#include "../network/doctest.h" + +#include "ark/kernel/nested_wire.hpp" +#include "nested_stream_fixture.hpp" + +namespace { +using namespace ark::entity_sync; +namespace wire = nested_wire; +namespace proto = ark::protobuf::entity_sync; + +proto::nested_baseline encoded_initial() { + proto::nested_baseline output; + wire::encode(nested_test::initial(), output, nested_test::binding()); + return output; +} + +proto::nested_delta encoded_change() { + nested_publisher publisher(nested_test::binding(), nested_test::initial()); + proto::nested_delta output; + wire::encode(*publisher.update(nested_test::changed()), output, nested_test::binding()); + return output; +} +} // namespace + +TEST_CASE("Nested wire preserves every scalar width zero presence and binary bytes") { + auto baseline = nested_test::initial(); + auto trusted = nested_test::binding(); + baseline.snapshot.root.scalars.fields = {{1, false}, {2, std::int32_t{0}}, {3, std::uint32_t{0}}, + {4, std::numeric_limits::min()}, {5, std::numeric_limits::max()}, {6, 0.0f}, + {7, 0.0}, {8, std::string("\xff\0\x80", 3)}, {9, guid_value{-1}}}; + trusted.root.fields = {{1, nested_scalar_type::boolean}, {2, nested_scalar_type::int32}, + {3, nested_scalar_type::uint32}, {4, nested_scalar_type::int64}, {5, nested_scalar_type::uint64}, + {6, nested_scalar_type::real32}, {7, nested_scalar_type::real64}, {8, nested_scalar_type::bytes}, + {9, nested_scalar_type::guid}}; + proto::nested_baseline message; + wire::encode(baseline, message, trusted); + REQUIRE(message.snapshot().root().fields_size() == 9); + CHECK(message.snapshot().root().fields(0).value_case() == proto::scalar_field::kBoolValue); + CHECK(message.snapshot().root().fields(1).value_case() == proto::scalar_field::kInt32Value); + CHECK(message.snapshot().root().fields(2).value_case() == proto::scalar_field::kUint32Value); + CHECK(message.snapshot().root().fields(5).value_case() == proto::scalar_field::kFloatValue); + CHECK(message.snapshot().root().fields(6).value_case() == proto::scalar_field::kDoubleValue); + CHECK(wire::decode(message, trusted).snapshot == baseline.snapshot); + const auto bytes = wire::encode(baseline, trusted); + CHECK(wire::decode_baseline(bytes, trusted).snapshot == baseline.snapshot); + nested_publisher publisher(trusted, baseline); + auto next = baseline.snapshot; + next.root.scalars.fields.at(1) = true; + next.root.scalars.fields.at(8) = std::string("\0\xfe", 2); + nested_replica receiver(trusted); + REQUIRE(receiver.apply(wire::decode_baseline(bytes, trusted)) == nested_apply_result::applied); + const auto delta = wire::decode_delta(wire::encode(*publisher.update(next), trusted), trusted); + REQUIRE(receiver.apply(delta) == nested_apply_result::applied); + CHECK(*receiver.state() == next); + CHECK(receiver.apply(delta) == nested_apply_result::duplicate); +} + +TEST_CASE("Nested wire row lifetimes relationship replacements and tombstones compose with the replica") { + const auto trusted = nested_test::binding(); + nested_publisher publisher(trusted, nested_test::initial()); + nested_replica receiver(trusted); + REQUIRE(receiver.apply(wire::decode_baseline(wire::encode(*publisher.baseline(), trusted), trusted)) == + nested_apply_result::applied); + auto next = nested_test::changed(); + auto packet = wire::decode_delta(wire::encode(*publisher.update(next), trusted), trusted); + REQUIRE(receiver.apply(packet) == nested_apply_result::applied); + CHECK(*receiver.state() == next); + next.root.tables.at(200).rows.at(1).incarnation = 3; + packet = wire::decode_delta(wire::encode(*publisher.update(next), trusted), trusted); + CHECK(packet.root.tables.at(200).at(1).required_incarnation == 1); + CHECK(packet.root.tables.at(200).at(1).incarnation == 3); + CHECK(packet.root.tables.at(200).at(1).fields.size() == 2); + REQUIRE(receiver.apply(packet) == nested_apply_result::applied); + next.root.tables.at(200).rows.clear(); + packet = wire::decode_delta(wire::encode(*publisher.update(next), trusted), trusted); + CHECK(packet.root.tables.at(200).at(1).incarnation == 0); + REQUIRE(receiver.apply(packet) == nested_apply_result::applied); + const auto child = next.root.containers.at(201).at(1); + next.root.containers.at(201).clear(); + next.root.containers.at(202).emplace(7, child); + proto::nested_delta message; + wire::encode(*publisher.update(next), message, trusted); + REQUIRE(message.root().containers_size() == 2); + CHECK(message.root().containers(0).has_replacement()); + CHECK(message.root().containers(0).replacement().slots_size() == 0); + CHECK(message.root().containers(1).has_replacement()); + REQUIRE(receiver.apply(wire::decode(message, trusted)) == nested_apply_result::applied); + CHECK(*receiver.state() == next); + const auto removed = wire::decode_delta(wire::encode(*publisher.remove(), trusted), trusted); + REQUIRE(receiver.apply(removed) == nested_apply_result::applied); + CHECK(receiver.valid()); + CHECK(receiver.state() == nullptr); + CHECK(receiver.apply(removed) == nested_apply_result::duplicate); +} + +TEST_CASE("Nested wire rejects duplicate baseline indices without collapsing records") { + auto message = encoded_initial(); + auto& root = *message.mutable_snapshot()->mutable_root(); + SUBCASE("scalar") { + *root.add_fields() = root.fields(0); + } + SUBCASE("table") { + *root.add_tables() = root.tables(0); + } + SUBCASE("row") { + auto& table = *root.mutable_tables(0); + *table.add_rows() = table.rows(0); + } + SUBCASE("container") { + *root.add_containers() = root.containers(0); + } + SUBCASE("slot") { + auto& container = *root.mutable_containers(0); + *container.add_slots() = container.slots(0); + } + CHECK_THROWS_AS(wire::decode(message, nested_test::binding()), std::invalid_argument); + CHECK_THROWS_AS(wire::decode_baseline(message.SerializeAsString(), nested_test::binding()), std::invalid_argument); +} + +TEST_CASE("Nested wire rejects duplicate delta indices and missing container alternatives") { + auto message = encoded_change(); + auto& root = *message.mutable_root(); + SUBCASE("scalar") { + *root.add_fields() = root.fields(0); + } + SUBCASE("table") { + *root.add_tables() = root.tables(0); + } + SUBCASE("row") { + auto& table = *root.mutable_tables(0); + *table.add_rows() = table.rows(0); + } + SUBCASE("container") { + *root.add_containers() = root.containers(0); + } + SUBCASE("changed slot") { + auto& patch = *root.mutable_containers(0)->mutable_patch(); + *patch.add_changed() = patch.changed(0); + } + SUBCASE("replacement slot") { + auto* slots = root.mutable_containers(0)->mutable_replacement(); + auto* first = slots->add_slots(); + first->set_slot(1); + *first->mutable_state() = encoded_initial().snapshot().root().containers(0).slots(0).state(); + *slots->add_slots() = slots->slots(0); + } + SUBCASE("missing alternative") { + root.mutable_containers(0)->clear_update(); + } + CHECK_THROWS_AS(wire::decode(message, nested_test::binding()), std::invalid_argument); +} + +TEST_CASE("Nested wire rejects malformed envelopes values and independently untrusted schemas") { + auto message = encoded_initial(); + auto& root = *message.mutable_snapshot()->mutable_root(); + SUBCASE("missing scalar alternative") { + root.mutable_fields(0)->clear_value(); + } + SUBCASE("wrong scalar type") { + root.mutable_fields(0)->set_uint32_value(0); + } + SUBCASE("nonfinite") { + root.mutable_fields(0)->set_double_value(std::numeric_limits::infinity()); + } + SUBCASE("zero index") { + root.mutable_fields(0)->set_index(0); + } + SUBCASE("zero identity") { + root.set_entity_id(0); + } + SUBCASE("zero slot") { + root.mutable_containers(0)->mutable_slots(0)->set_slot(0); + } + SUBCASE("missing child state") { + root.mutable_containers(0)->mutable_slots(0)->clear_state(); + } + SUBCASE("missing root") { + message.mutable_snapshot()->clear_root(); + } + SUBCASE("missing snapshot") { + message.clear_snapshot(); + } + SUBCASE("schema mismatch") { + message.mutable_snapshot()->set_schema_id("other"); + } + SUBCASE("wrong class on empty table") { + root.mutable_tables(0)->clear_rows(); + root.mutable_tables(0)->set_class_name("Other"); + } + CHECK_THROWS_AS(wire::decode(message, nested_test::binding()), std::invalid_argument); + auto trusted = nested_test::binding(); + trusted.root.tables.at(200).columns.at(1) = nested_scalar_type::int32; + CHECK_THROWS_AS(wire::decode(encoded_initial(), trusted), std::invalid_argument); +} + +TEST_CASE("Nested wire retains unknown-field compatibility within byte and recursion budgets") { + auto message = encoded_initial(); + message.GetReflection()->MutableUnknownFields(&message)->AddVarint(999, 1); + CHECK(wire::decode(message, nested_test::binding()).snapshot == nested_test::initial().snapshot); + auto* group = message.GetReflection()->MutableUnknownFields(&message)->AddGroup(998); + for (int depth = 0; depth < 10; ++depth) + { + group = group->AddGroup(998); + } + wire::limits budget; + budget.max_recursion_depth = 8; + CHECK_THROWS_AS(wire::decode(message, nested_test::binding(), budget), std::length_error); +} + +TEST_CASE("Nested wire validates before replacing caller output and bounds bytes and Protobuf nesting") { + const auto baseline = nested_test::initial(); + const auto trusted = nested_test::binding(); + const auto original = encoded_initial(); + proto::nested_baseline output = original; + auto invalid = baseline; + invalid.snapshot.root.scalars.fields.at(1) = true; + CHECK_THROWS_AS(wire::encode(invalid, output, trusted), std::invalid_argument); + CHECK(output.SerializeAsString() == original.SerializeAsString()); + auto budget = wire::limits{}; + budget.max_message_bytes = original.ByteSizeLong() - 1; + CHECK_THROWS_AS(wire::encode(baseline, output, trusted, budget), std::length_error); + CHECK(output.SerializeAsString() == original.SerializeAsString()); + CHECK_THROWS_AS(wire::decode(original, trusted, budget), std::length_error); + CHECK_THROWS_AS(wire::decode_baseline(original.SerializeAsString(), trusted, budget), std::length_error); + budget.max_message_bytes = original.ByteSizeLong(); + CHECK(wire::decode(original, trusted, budget).snapshot == baseline.snapshot); + budget.max_recursion_depth = 1; + CHECK_THROWS_AS(wire::encode(baseline, output, trusted, budget), std::length_error); + CHECK(output.SerializeAsString() == original.SerializeAsString()); + CHECK_THROWS_AS(wire::decode(original, trusted, budget), std::length_error); + CHECK_THROWS_AS(wire::decode_baseline(original.SerializeAsString(), trusted, budget), std::invalid_argument); + CHECK_THROWS_AS(wire::decode_baseline(std::string("\xff", 1), trusted), std::invalid_argument); +} + +TEST_CASE("Nested wire applies complete state patch and lifetime budgets separately") { + const auto trusted = nested_test::binding(); + const auto initial = encoded_initial(); + const auto change = encoded_change(); + wire::limits budget; + SUBCASE("complete rows") { + budget.stream.snapshot.max_rows = 1; + CHECK_THROWS_AS(wire::decode(initial, trusted, budget), std::length_error); + } + SUBCASE("complete fields") { + budget.stream.snapshot.max_fields = 1; + CHECK_THROWS_AS(wire::decode(initial, trusted, budget), std::length_error); + } + SUBCASE("complete text") { + budget.stream.snapshot.max_string_bytes = 1; + CHECK_THROWS_AS(wire::decode(initial, trusted, budget), std::length_error); + } + SUBCASE("lifetime history") { + budget.stream.max_lifetime_entries = 1; + CHECK_THROWS_AS(wire::decode(initial, trusted, budget), std::length_error); + } + SUBCASE("patch operations") { + budget.stream.max_patch_operations = 1; + CHECK_THROWS_AS(wire::decode(change, trusted, budget), std::length_error); + } + SUBCASE("patch values") { + budget.stream.max_patch_values = 1; + CHECK_THROWS_AS(wire::decode(change, trusted, budget), std::length_error); + } + SUBCASE("patch text") { + budget.stream.max_patch_string_bytes = 1; + CHECK_THROWS_AS(wire::decode(change, trusted, budget), std::length_error); + } +} + +TEST_CASE("Nested wire stages publications on the destination Protobuf arena") { + google::protobuf::Arena arena; + auto* output = google::protobuf::Arena::Create(&arena); + output->set_revision(99); + const auto before = output->SerializeAsString(); + auto invalid = nested_test::initial(); + invalid.revision = 0; + CHECK_THROWS_AS(wire::encode(invalid, *output, nested_test::binding()), std::invalid_argument); + CHECK(output->SerializeAsString() == before); + wire::encode(nested_test::initial(), *output, nested_test::binding()); + CHECK(wire::decode(*output, nested_test::binding()).snapshot == nested_test::initial().snapshot); + CHECK(output->GetArena() == &arena); +} + +TEST_CASE("Decoded nested packets retain atomic rejection and terminal recovery semantics") { + const auto trusted = nested_test::binding(); + nested_replica receiver(trusted); + REQUIRE(receiver.apply(wire::decode(encoded_initial(), trusted)) == nested_apply_result::applied); + const auto retained = receiver.state(); + auto message = encoded_change(); + SUBCASE("revision gap") { + message.set_required_revision(2); + message.set_revision(3); + } + SUBCASE("late stale child incarnation") { + message.mutable_root() + ->mutable_containers(0) + ->mutable_patch() + ->mutable_changed(0) + ->mutable_change() + ->set_incarnation(99); + } + SUBCASE("stale required row incarnation") { + auto* row = message.mutable_root()->mutable_tables(0)->mutable_rows(0); + row->set_required_incarnation(99); + row->set_incarnation(99); + } + const auto decoded = wire::decode(message, trusted); + CHECK(receiver.apply(decoded) == nested_apply_result::rejected); + CHECK(receiver.state() == nullptr); + CHECK_FALSE(receiver.valid()); + CHECK(*retained == nested_test::initial().snapshot); + CHECK(receiver.apply(wire::decode(encoded_initial(), trusted)) == nested_apply_result::rejected); +} + +TEST_CASE("Nested wire GUIDs preserve signed identity and reject native narrowing") { + auto trusted = nested_test::binding(); + auto baseline = nested_test::initial(); + trusted.root_id = -10; + baseline.snapshot.root.scalars.entity_id = -10; + CHECK(wire::decode_baseline(wire::encode(baseline, trusted), trusted).snapshot == baseline.snapshot); +#if defined(ARK_USE_GUID_32) + auto message = encoded_initial(); + message.mutable_snapshot()->mutable_root()->set_entity_id(std::numeric_limits::max()); + CHECK_THROWS_AS(wire::decode(message, nested_test::binding()), std::invalid_argument); +#endif +} diff --git a/tests/core/nested_wire_macro.cpp b/tests/core/nested_wire_macro.cpp new file mode 100644 index 000000000..071b5b0d0 --- /dev/null +++ b/tests/core/nested_wire_macro.cpp @@ -0,0 +1,36 @@ +// Copyright (c) 2026 OpenArkStudio. Licensed under Apache-2.0. See LICENSE. +#include + +#include "../network/doctest.h" +#include + +#include "ark/platform.hpp" + +// Reproduce Win32's object-like macro after Reflection's declaration is fixed. +#pragma push_macro("GetMessage") +#undef GetMessage +#define GetMessage ark_nested_wire_get_message_sentinel + +#include "ark/kernel/nested_wire.hpp" + +#define ARK_NESTED_WIRE_STRINGIZE_IMPL(value) #value +#define ARK_NESTED_WIRE_STRINGIZE(value) ARK_NESTED_WIRE_STRINGIZE_IMPL(value) +static_assert(std::string_view(ARK_NESTED_WIRE_STRINGIZE(GetMessage)) == "ark_nested_wire_get_message_sentinel"); +#undef ARK_NESTED_WIRE_STRINGIZE +#undef ARK_NESTED_WIRE_STRINGIZE_IMPL +#pragma pop_macro("GetMessage") + +#include "nested_stream_fixture.hpp" + +TEST_CASE("Nested wire preserves the caller GetMessage macro while round-tripping a baseline") { + using namespace ark::entity_sync; + const auto trusted = nested_test::binding(); + const auto initial = nested_test::initial(); + const auto bytes = nested_wire::encode(initial, trusted); + const auto decoded = nested_wire::decode_baseline(bytes, trusted); + CHECK(decoded.snapshot == initial.snapshot); + nested_replica receiver(trusted); + CHECK(receiver.apply(decoded) == nested_apply_result::applied); + REQUIRE(receiver.state() != nullptr); + CHECK(*receiver.state() == initial.snapshot); +} diff --git a/tests/core/nested_wire_macro_first.cpp b/tests/core/nested_wire_macro_first.cpp new file mode 100644 index 000000000..4c3d306f6 --- /dev/null +++ b/tests/core/nested_wire_macro_first.cpp @@ -0,0 +1,35 @@ +// Copyright (c) 2026 OpenArkStudio. Licensed under Apache-2.0. See LICENSE. +#include + +#include "../network/doctest.h" + +#include "ark/platform.hpp" + +// The public codec must also protect a first Protobuf include from the macro. +#pragma push_macro("GetMessage") +#undef GetMessage +#define GetMessage ark_nested_wire_first_get_message_sentinel + +#include "ark/kernel/nested_wire.hpp" + +#define ARK_NESTED_WIRE_STRINGIZE_IMPL(value) #value +#define ARK_NESTED_WIRE_STRINGIZE(value) ARK_NESTED_WIRE_STRINGIZE_IMPL(value) +static_assert(std::string_view(ARK_NESTED_WIRE_STRINGIZE(GetMessage)) == "ark_nested_wire_first_get_message_sentinel"); +#undef ARK_NESTED_WIRE_STRINGIZE +#undef ARK_NESTED_WIRE_STRINGIZE_IMPL +#pragma pop_macro("GetMessage") + +#include "nested_stream_fixture.hpp" + +TEST_CASE("Nested wire protects first Protobuf declarations from a caller GetMessage macro") { + using namespace ark::entity_sync; + const auto trusted = nested_test::binding(); + const auto initial = nested_test::initial(); + const auto bytes = nested_wire::encode(initial, trusted); + const auto decoded = nested_wire::decode_baseline(bytes, trusted); + CHECK(decoded.snapshot == initial.snapshot); + nested_replica receiver(trusted); + CHECK(receiver.apply(decoded) == nested_apply_result::applied); + REQUIRE(receiver.state() != nullptr); + CHECK(*receiver.state() == initial.snapshot); +}