diff --git a/docs/roadmap/DISTRIBUTED_STATE_ROADMAP.md b/docs/roadmap/DISTRIBUTED_STATE_ROADMAP.md index 2257d1f90..cec2174f7 100644 --- a/docs/roadmap/DISTRIBUTED_STATE_ROADMAP.md +++ b/docs/roadmap/DISTRIBUTED_STATE_ROADMAP.md @@ -54,16 +54,14 @@ later that week. Also present on `master` today: These are implemented on the receive side or in isolation, and are easy to mistake for finished: -- **`data()` does not replicate end to end.** `state_sync_adapter::attach_node` - connects only `valueChanged` and `childChanged`, so nothing journals - `dataChanged`, `commentChanged`, `hiddenChanged` or `readOnlyChanged` — even - though the op enum, the wire proto and `apply_remote` all support them. Those - mutations can currently only be produced by hand-constructing a - `state_mutation`. Additionally, `apply_remote`'s `set_data` case is an - explicit no-op pending a codec registry - (`src/cvc/state/state_sync_adapter.cpp:293`). -- **`state_data_hydrator` is never driven from the ingest path.** It works and - is tested, but nothing calls it automatically on receiving a blob reference. +- **`data()` object replication — LANDED** (status corrected 2026-10-22; this was previously listed + here as a gap). The sync adapter hooks `dataChanged` and emits a codec-encoded `set_data` (inline, + or a content-addressed `blob_ref` above the shard's inline threshold); `apply_remote` decodes inline + via the per-shard `state_codec_registry` or hydrates a `blob_ref`; `snapshot()` captures it and + `ingest_snapshot_entry` replays it; and `state_data_hydrator` IS driven automatically from + `apply_remote`'s blob branch (the session wires it on join). See Phase 10 below for the full chain + and the residual deferred items (WebSocket late-joiner blob hydration, the async hydrator install + guard, the legacy oversized-`value()` resolver). - **Backpressure and slow-peer isolation are `inproc`-only.** The bounded outbox, drop policies and quarantine live in `state_transport_inproc`; the IPC and gRPC transports publish synchronously with no bounded queue. @@ -421,59 +419,86 @@ Delivered so far on `feature/phase9-telemetry`: - Failure mode: an aggregator stops receiving from a peer for > 3× publish interval and marks the peer `stale`. - Bench: 100-node simulated cluster, 1 Hz telemetry, assert per-node CPU overhead is < 1% and aggregator memory is bounded. -### Phase 10: Replicated structured/binary node `data()` (codec-backed) — NOT STARTED +### Phase 10: Replicated structured/binary node `data()` (codec-backed) — LANDED (residual items below) Surfaced 2026-10-01 while reviewing the binary-messaging/streaming work (originally floated as the -"snapshot binary gap"). On investigation the gap is **not** a snapshot-field tweak — it is the whole -`data()` replication path, which is stubbed out end to end. - -**What already works (so this phase does NOT need to touch it):** a node's `value()` (a `std::string`) -replicates byte-safely over both transports *and* survives a full-tree snapshot resync. IPC frames every -string length-prefixed (`state_transport_ipc.cpp` `put_string`/`reader::str`), and proto3 `string` is a -length-delimited field the C++ runtime passes arbitrary octets through. So a binary *value()* is fine on -both the mutation and snapshot paths — there is no asymmetry to fix there. - -**The actual gap:** a node's `data()` (a `boost::any` holding structured/typed/binary payload — e.g. the -`format_desc` a stream descriptor stores) does **not replicate at all**: - -- Nothing emits a `SET_DATA` mutation. `state_sync_adapter` only hooks `valueChanged` and emits - `set_value`; it never hooks `dataChanged`. -- The apply side is an explicit no-op: `state_sync_adapter`'s `set_data` case is `// accept it but do not - decode … phase 2 will provide a codec registry`. -- The snapshot serializer (`state_cluster_shard::snapshot()`) captures only `value()`, `comment`, - `hidden`, `read_only` — never `data()`. -- The *carriers* already exist but are unused end to end: the journal's `state_payload` - (`none|inline_bytes|blob`), the proto `Mutation.SET_DATA` + `Payload{inline_bytes|blob}`, the blob store, - and `state_data_hydrator` (which can already reassemble + decode a blob-ref payload — but only once a - codec registry exists to decode it). - -So "binary data survives resync" presupposes a `data()`-replication path that was never built; it is the -deferred **codec registry** the no-op comment names, not an extra `bytes` field on `SnapshotEntry`. - -**Design to close it:** - -1. **Codec registry** — `encode(const boost::any&, type_name) -> bytes` / `decode(bytes, type_name) -> - boost::any`, keyed by a registered `type_name`; pluggable per data type (volume, geometry, image, the - small structured types first). An unregistered type is skipped with a structured diagnostic, never a - silent corruption. -2. **Emit** — `state_sync_adapter` hooks `dataChanged` and emits a `set_data` mutation carrying - `type_name` + `state_payload`: `inline_bytes` for small payloads, a content-addressed `blob_ref` - (through the existing blob store) above a threshold. -3. **Apply** — the `set_data` case decodes via the registry (inline) or routes a `blob_ref` through - `state_data_hydrator` (fetch chunks → reassemble → decode → `node->data(decoded)`), replacing the - current no-op. -4. **Snapshot** — add a `payload` (type_name + `state_payload`) to the `snapshot_entry` struct, the proto - `SnapshotEntry`, and the IPC snapshot wire format; populate it from `node.data()` via the registry in - `state_cluster_shard::snapshot()`; apply it on receive via the same registry/hydrator path as a - `set_data` mutation. (Also carry `type_name` for values, which the snapshot currently drops.) - -**Testing:** a typed `data()` value round-trips through (a) a live mutation and (b) a full snapshot -resync, on both IPC and gRPC; a large payload round-trips as a blob-ref through the hydrator; an -unregistered type degrades gracefully; echo-suppression still holds (a `set_data` apply must not re-emit). - -**Dependency note:** this is independent of the cvc::ariadne::stream work but complements it — a replicated -`/streams/` descriptor (Phase 4 of the streaming roadmap) stores `format_desc` via `data()`, so it -needs this to survive a cross-node resync. +"snapshot binary gap"). **Status corrected 2026-10-22:** an earlier revision of this section said the +`data()` replication path was "NOT STARTED / stubbed out end to end." That is stale — the path LANDED with +the cross-node streaming critical-path work and is live across all four transports. This section now +records what is wired and the small items that remained to close §52 to its own design + testing bullets. + +**What already works (value()):** a node's `value()` (a `std::string`) replicates byte-safely over both +transports and survives a full-tree snapshot resync (IPC length-prefixes every string; proto3 `string` +passes arbitrary octets). A binary *value()* is fine on both the mutation and snapshot paths. + +**What is LANDED for `data()`** (a `boost::any` holding a structured/typed/binary payload — e.g. the +`format_desc` a stream descriptor stores): + +- **Codec registry** — `cvc::state_codec_registry` (`type_name -> encode/decode`), built per shard with + `register_builtin_codecs` (bool / the integer family / float / double / std::string) + + `register_volume_geometry_codecs`; the shard ctor sets it on the adapter so it is non-null in production. + An unregistered type is skipped (the object does not replicate; value + child structure still converge), + never a silent corruption. +- **Emit** — `state_sync_adapter` hooks `dataChanged` and emits a `set_data` carrying `type_name` + + `state_payload`: `inline_bytes` for small payloads, a content-addressed `blob_ref` (via the blob store) + above the shard's `max_inline_payload_bytes`. +- **Apply** — the `set_data` case decodes inline via the registry or routes a `blob_ref` through + `state_data_hydrator` (fetch chunks → reassemble → decode → `node->data(decoded)`), under the adapter's + thread-suppression guard (no echo). +- **Snapshot** — `snapshot_entry` / proto `SnapshotEntry` carry `data_type_name` + `data_payload`; + `state_cluster_shard::snapshot()` captures `node.data()` mirroring the live encode+offload byte-for-byte, + and `ingest_snapshot_entry` replays it as a `set_data`. The value's own `type_name` is carried too. +- **Wire** — the proto `Payload` oneof (`none|inline_bytes|BlobRef`) and the `SnapshotEntry` payload fields + serialize on all four transports (inproc by-value, IPC `put_payload`/`read_payload`, gRPC + `set_pb_payload`, WS). +- **Session wiring** — `distributed_state_session` sets the blob store + `max_inline_payload_bytes`, builds + the hydrator over the shard codecs + blob store, and requests + ingests a snapshot on join. +- **Echo suppression** — transport origin-skip + the shard `(origin,sequence)` seen-set + the adapter's + thread-suppression guard. + +**Closing items (this change):** + +1. **Integer `data()` key fix (LANDED).** `register_builtin_codecs` keyed the integer codecs on the C++ + spellings (`int32_t`…), but `state::dataTypeName()` returns the app's registered names (`int`, + `unsigned int`, `unsigned short`, `int64`, `uint64`), so an integer `data()` silently failed to + replicate (encode threw "no codec", swallowed). Codec aliases under the app spellings fix it. Regression + test: `TwoNodeInprocIntegerDataReplication`. +2. **Value `type_name` across apply + snapshot (LANDED).** The emit carried `valueTypeName()`, but + `apply_remote::set_value` stamped `"std::string"` and `snapshot()` never captured the value type. A + signal-free `state::valueTypeName(const std::string&)` setter + apply/snapshot wiring preserve it. Test: + `TwoNodeValueTypeNamePreserved`. +3. **Transport coverage (LANDED).** A typed `data()` round-trip over IPC — inline, builtin-int, and a + blob-ref — is `StateTransportIpcTest.DataPayloadReplicatesOverIpc`. gRPC has equivalent coverage + (`SnapshotCarriesData` / `BlobPayloadRoundTrip`), built only under `CVC_ENABLE_GRPC`. +4. **`set_data` echo-suppression pinned (LANDED).** `DistributedStateSession.SetDataApplyDoesNotEcho` + asserts an applied `set_data` emits no local mutation and a quiescent re-pump moves nothing. + +**Deferred (real, but outside this closure):** + +- **`format_desc` on a production shard.** `register_stream_codecs` + the `format_desc` codec exist and the + descriptor is written to a `/streams/` node's `data()`, but NO production shard registers the codec — + because no production shard currently *replicates* those descriptors: the conference relay scopes + replication to `conf.*` (`add_interest("conf")` + `set_enforce_interest(true)`), descriptor nodes are + rooted at the bare stream token (outside `conf.*`), and cross-node media rides `stream_bridge` / + `media_frame`, not the state tree. When cross-node descriptor *discovery* is wired (a holon, or a + conference feature placing descriptors under a replicated root), register the codec on both shards' codec + registries, keyed by the stable `"cvc::ariadne::stream::format_desc"` string, and treat the decoder as an + adversarial-bytes surface (`origin_node_id` is forgeable — bounds-check, do not trust length fields). + Covered in isolation today by `TwoNodeStreamDescriptorDiscovery` (test-local codec registration). +- **WebSocket late-joiner blob hydration** — the WS leg does the ~1 s full-shard broadcast with bounded + inline bytes only; a large blob-ref payload to a browser late joiner is not yet pullable. Belongs to the + conference/browser phase. +- **Async hydrator late-install** — `state_data_hydrator::retry_pending` / `on_hydrated` install `data()` + off the ingest thread with no suppression guard and no conflict-order re-check; inproc/IPC/gRPC + `fetch_chunk` completes inline (under the guard), so this is unused today but MUST be closed before any + non-blocking chunk transport ships. +- **Oversized `value()` blob offload** — `set_value` above the threshold emits a `blob_ref` with an empty + `string_value` and NO receive-side resolver, so a large `value()` arrives unresolved. Either wire a + `set_value` hydrator or cap/forbid oversized `value()` (large payloads should use `data()`). + +**Dependency note:** independent of the cvc::ariadne::stream work but complements it — a replicated +`/streams/` descriptor stores `format_desc` via `data()`, so cross-node descriptor discovery will need +the deferred `format_desc` registration above once such a replication route exists. ## Testing Requirements diff --git a/inc/cvc/state/state.h b/inc/cvc/state/state.h index a253a4810..503391f57 100644 --- a/inc/cvc/state/state.h +++ b/inc/cvc/state/state.h @@ -245,6 +245,11 @@ class state : public boost::enable_shared_from_this { std::string value(); std::string valueTypeName(); + // Set the value's declared type name directly, WITHOUT changing the value or emitting a change + // signal. Replication uses this on apply/snapshot to preserve a value's source type across a node + // that only receives the string form (the plain value(const std::string&) setter would stamp + // "std::string"). Not for ordinary code, which sets the type by writing a typed value(). + void valueTypeName(const std::string &typeName); std::vector values(bool unique = false); // shortcut for comma separated values in value() state &value(const std::string &v, bool setValueType = true); diff --git a/src/cvc/state/state.cpp b/src/cvc/state/state.cpp index bf272140d..ff083063a 100644 --- a/src/cvc/state/state.cpp +++ b/src/cvc/state/state.cpp @@ -336,6 +336,16 @@ std::string state::valueTypeName() { return _valueTypeName; } +// Set the declared value type directly, without touching the value or emitting a change signal. +// Replication apply/snapshot uses this to preserve a value's source type across a receive-only node +// (see the header). Deliberately signal-free: the value itself is set separately, under the sync +// adapter's thread-suppression guard, so this must not re-enter the journal. +void state::valueTypeName(const std::string &typeName) { + boost::this_thread::interruption_point(); + boost::mutex::scoped_lock lock(_mutex); + _valueTypeName = typeName; +} + // -------------- // state::comment // -------------- diff --git a/src/cvc/state/state_cluster_shard.cpp b/src/cvc/state/state_cluster_shard.cpp index cfde4af46..9859a83ce 100644 --- a/src/cvc/state/state_cluster_shard.cpp +++ b/src/cvc/state/state_cluster_shard.cpp @@ -697,6 +697,7 @@ state_cluster_shard::snapshot(const std::string &path_prefix) const { snapshot_entry e; e.path = child_path; e.string_value = node.value(); + e.type_name = node.valueTypeName(); // the value's declared type, so it survives a resync e.comment = node.comment(); e.hidden = node.hidden(); e.read_only = node.readOnly(); @@ -744,7 +745,17 @@ void state_cluster_shard::ingest_snapshot_entry(const snapshot_entry &e) { }; // value() always (also creates the node on the joiner, matching the source even when empty). - apply_scalar(state_mutation_op::set_value, e.string_value); + // Carry the declared value type too (apply_remote honors m.type_name on set_value), so a typed + // read on the joiner sees the source type rather than the "std::string" the string setter stamps. + { + state_mutation m; + m.cluster_id = _cluster_id; + m.path = e.path; + m.op = state_mutation_op::set_value; + m.string_value = e.string_value; + m.type_name = e.type_name; + _adapter->apply_remote(m); + } // Metadata: apply only the non-default flags (a fresh joiner's nodes start at the defaults). if (!e.comment.empty()) diff --git a/src/cvc/state/state_codec_registry.cpp b/src/cvc/state/state_codec_registry.cpp index f7de537a8..5b66d538c 100644 --- a/src/cvc/state/state_codec_registry.cpp +++ b/src/cvc/state/state_codec_registry.cpp @@ -9,6 +9,7 @@ */ #include +#include #include #include #include @@ -189,6 +190,61 @@ void state_codec_registry::register_builtin_codecs() { return boost::any(std::string(b.begin(), b.end())); }, "cvc.str.v1"); + + // Integer data() under the app's REGISTERED spellings. state::dataTypeName() returns the name + // app::registerDefaultTypes gave the type — "int", "unsigned int", "unsigned short", "int64", + // "uint64" — NOT the C++ "_t" spellings the codecs above are keyed under. Without these aliases, + // _codecs->encode()/decode() miss for an integer data(), the exception is swallowed in the sync + // adapter, and the integer object silently fails to replicate (value() + structure still flow). + // Each alias any_casts the EXACT stored type (boost::any_cast is exact-typeid, not conversion): + // "int64"/"uint64" are boost::int64_t/uint64_t (cvc::int64/uint64 in core/types.h). float, + // double, std::string and bool already match the app's spelling and need no alias. + register_codec( + "int", [](const boost::any &v) { return encode_pod(boost::any_cast(v)); }, + [](const std::vector &b) -> boost::any { + return boost::any(decode_pod(b)); + }, + "cvc.int.v1"); + + register_codec( + "unsigned int", + [](const boost::any &v) { + return encode_pod(boost::any_cast(v)); + }, + [](const std::vector &b) -> boost::any { + return boost::any(decode_pod(b)); + }, + "cvc.uint.v1"); + + register_codec( + "unsigned short", + [](const boost::any &v) { + return encode_pod(boost::any_cast(v)); + }, + [](const std::vector &b) -> boost::any { + return boost::any(decode_pod(b)); + }, + "cvc.ushort.v1"); + + register_codec( + "int64", + [](const boost::any &v) { + return encode_pod(boost::any_cast(v)); + }, + [](const std::vector &b) -> boost::any { + return boost::any(decode_pod(b)); + }, + "cvc.int64.v1"); + + register_codec( + "uint64", + [](const boost::any &v) { + return encode_pod(boost::any_cast(v)); + }, + [](const std::vector &b) -> boost::any { + return boost::any(decode_pod(b)); + }, + "cvc.uint64.v1"); } } // namespace cvc diff --git a/src/cvc/state/state_sync_adapter.cpp b/src/cvc/state/state_sync_adapter.cpp index bbf44f558..8d1d87cf4 100644 --- a/src/cvc/state/state_sync_adapter.cpp +++ b/src/cvc/state/state_sync_adapter.cpp @@ -335,6 +335,11 @@ bool state_sync_adapter::apply_remote(const state_mutation &m) { switch (m.op) { case state_mutation_op::set_value: node.value(m.string_value); + // Preserve the source value's declared type: value(const std::string&) stamps "std::string", + // but the emit carried the real type (valueChanged sends m.type_name = n.valueTypeName()). The + // setter is signal-free and we are under the suppression guard, so this does not re-journal. + if (!m.type_name.empty()) + node.valueTypeName(m.type_name); break; case state_mutation_op::set_comment: node.comment(m.string_value); diff --git a/src/cvc/tests/distributed_state_session_test.cpp b/src/cvc/tests/distributed_state_session_test.cpp index 853e009f5..8f9469471 100644 --- a/src/cvc/tests/distributed_state_session_test.cpp +++ b/src/cvc/tests/distributed_state_session_test.cpp @@ -3,6 +3,8 @@ Tests for the distributed_state_session convenience API. */ +#include +#include #include #include #include @@ -13,6 +15,7 @@ #include #include #include +#include // state_mutation / state_mutation_op (echo test) #include #include #include @@ -189,6 +192,72 @@ TEST(DistributedStateSession, TwoNodeInprocDataReplication) { s2.detach(); } +// data() replication of the builtin integer scalars (regression for the codec-key name mismatch): +// register_builtin_codecs must key integer codecs under the SAME spellings state::dataTypeName() +// returns — the app::registerDefaultTypes names "int" / "unsigned int" / "unsigned short" / "int64" +// / "uint64", NOT the C++ "int32_t"/... spellings — or an integer data() silently fails to +// replicate (its encode throws "no codec", is swallowed in the sync adapter) while value() and +// child structure still flow. No manual codec registration here: these are builtin codecs the shard +// ctor installs, and registerDefaultTypes already names the types. +TEST(DistributedStateSession, TwoNodeInprocIntegerDataReplication) { + app a1, a2; + + state_transport_inproc shared_transport; + state_cluster_shard s1(a1, "cluster", "node1"); + state_cluster_shard s2(a2, "cluster", "node2"); + s1.attach(); + s2.attach(); + shared_transport.register_shard(&s1); + shared_transport.register_shard(&s2); + + // data() has no equality guard (unlike value()), so a plain double-write beats the creation- + // consume of a fresh node's first signal. + { + state &n = state::instance(a1)("nums.i"); + n.value(std::string("p1")); + n.value(std::string("p2")); + n.data(boost::any(int{42})); + n.data(boost::any(int{42})); + EXPECT_EQ(n.dataTypeName(), "int") << "app int spelling changed; update the codec alias key"; + } + { + state &n = state::instance(a1)("nums.i64"); + n.value(std::string("p1")); + n.value(std::string("p2")); + n.data(boost::any(boost::int64_t{-9000000000LL})); + n.data(boost::any(boost::int64_t{-9000000000LL})); + EXPECT_EQ(n.dataTypeName(), "int64") + << "app int64 spelling changed; update the codec alias key"; + } + { + state &n = state::instance(a1)("nums.u64"); + n.value(std::string("p1")); + n.value(std::string("p2")); + n.data(boost::any(boost::uint64_t{18000000000ULL})); + n.data(boost::any(boost::uint64_t{18000000000ULL})); + EXPECT_EQ(n.dataTypeName(), "uint64") << "app uint64 spelling changed; update the codec alias"; + } + + shared_transport.pump_all(); + + const boost::any gi = state::instance(a2)("nums.i").data(); + ASSERT_FALSE(gi.empty()) << "int data() did not replicate (codec key mismatch regressed?)"; + EXPECT_EQ(boost::any_cast(gi), 42); + + const boost::any g64 = state::instance(a2)("nums.i64").data(); + ASSERT_FALSE(g64.empty()) << "int64 data() did not replicate"; + EXPECT_EQ(boost::any_cast(g64), boost::int64_t{-9000000000LL}); + + const boost::any gu64 = state::instance(a2)("nums.u64").data(); + ASSERT_FALSE(gu64.empty()) << "uint64 data() did not replicate"; + EXPECT_EQ(boost::any_cast(gu64), boost::uint64_t{18000000000ULL}); + + shared_transport.unregister_shard(&s1); + shared_transport.unregister_shard(&s2); + s1.detach(); + s2.detach(); +} + namespace { // A larger descriptor: a small header plus a variable-length lookup table. Encoded, it exceeds // the inline threshold, so it exercises the blob-offload + hydrator path rather than inline bytes. @@ -450,6 +519,111 @@ TEST(DistributedStateSession, TwoNodeSnapshotDataMetadataReassembly) { s2.detach(); } +// value() type_name survives replication (S2): the valueChanged emit already carries +// n.valueTypeName(), but apply_remote::set_value stamped "std::string" and shard.snapshot() never +// captured the value type, so a typed value lost its declared type both on a live peer and on a +// snapshot resync. apply + snapshot now preserve it. +TEST(DistributedStateSession, TwoNodeValueTypeNamePreserved) { + app a1, a2, a3; + + state_transport_inproc shared_transport; + state_cluster_shard s1(a1, "cluster", "node1"); + state_cluster_shard s2(a2, "cluster", "node2"); + s1.attach(); + s2.attach(); + shared_transport.register_shard(&s1); + shared_transport.register_shard(&s2); + + // A typed (int) value: value records valueTypeName = dataTypeName() = "int"; the string + // setter would stamp "std::string". Set twice (first write consumed on node creation). + state &n1 = state::instance(a1)("typed.count"); + n1.value(7); + n1.value(8); + ASSERT_EQ(n1.valueTypeName(), "int"); + + shared_transport.pump_all(); + + // LIVE apply: the peer keeps both the string form AND the declared type. + EXPECT_EQ(state::instance(a2)("typed.count").value(), "8"); + EXPECT_EQ(state::instance(a2)("typed.count").valueTypeName(), "int") + << "value type_name dropped on live apply"; + + // SNAPSHOT resync: a late joiner reconstructs the declared type too (direct snapshot+ingest, the + // join path without a transport). + state_cluster_shard s3(a3, "cluster", "node3"); + s3.attach(); + const std::vector snap = s1.snapshot(std::string()); + ASSERT_FALSE(snap.empty()); + for (const auto &e : snap) + s3.ingest_snapshot_entry(e); + EXPECT_EQ(state::instance(a3)("typed.count").value(), "8"); + EXPECT_EQ(state::instance(a3)("typed.count").valueTypeName(), "int") + << "value type_name dropped on snapshot resync"; + + shared_transport.unregister_shard(&s1); + shared_transport.unregister_shard(&s2); + s1.detach(); + s2.detach(); + s3.detach(); +} + +// set_data echo-suppression (S5): §52 requires that applying a remote set_data must NOT re-emit a +// set_data back toward the origin (no ping-pong). The mechanism is apply_remote's +// thread_suppression_guard, which makes the dataChanged the apply triggers early-return instead of +// journaling. Pin it: node2, which only APPLIES node1's data(), must emit zero local mutations, and +// a quiescent re-pump must move nothing. +TEST(DistributedStateSession, SetDataApplyDoesNotEcho) { + app a1, a2; + + state_transport_inproc shared_transport; + state_cluster_shard s1(a1, "cluster", "node1"); + state_cluster_shard s2(a2, "cluster", "node2"); + + // Count every LOCAL mutation node2 emits. The emit path fires on_local_mutation; apply_remote, + // under its suppression guard, does not. An applied set_data must produce none. + int b_local_setdata = 0, b_local_any = 0; + s2.adapter().set_on_local_mutation([&](const state_mutation &m) { + ++b_local_any; + if (m.op == state_mutation_op::set_data) + ++b_local_setdata; + }); + + s1.attach(); + s2.attach(); + shared_transport.register_shard(&s1); + shared_transport.register_shard(&s2); + + state &n = state::instance(a1)("echo.n"); + n.value(std::string("p1")); + n.value(std::string("p2")); + n.data(boost::any(int{123})); + n.data(boost::any(int{123})); + + shared_transport.pump_all(); + + // node2 applied the data()... + const boost::any got = state::instance(a2)("echo.n").data(); + ASSERT_FALSE(got.empty()); + EXPECT_EQ(boost::any_cast(got), 123); + + // ...and emitted NOTHING locally in doing so (no echo, no ping-pong). + EXPECT_EQ(b_local_setdata, 0) << "applying a remote set_data re-emitted a set_data"; + EXPECT_EQ(b_local_any, 0) << "node2 emitted a local mutation while only applying remote ones"; + EXPECT_EQ(s2.adapter().local_mutation_count(), 0u); + + // A quiescent re-pump moves nothing — an echo would surface here as an outbound mutation. + const std::size_t pumped2 = shared_transport.pump_all(); + EXPECT_EQ(pumped2, 0u) + << "a set_data echo would surface as outbound mutations on the second pump"; + EXPECT_EQ(b_local_setdata, 0); + EXPECT_EQ(boost::any_cast(state::instance(a2)("echo.n").data()), 123); + + shared_transport.unregister_shard(&s1); + shared_transport.unregister_shard(&s2); + s1.detach(); + s2.detach(); +} + // Cross-node stream DISCOVERY (Phase 10, PR5 — the control-plane payoff): a real stream opened on // one node publishes its format_desc into the /streams/ descriptor node's data(); with the // stream codec registered on both shards, that descriptor replicates so a remote node discovers the diff --git a/src/cvc/tests/state_transport_ipc_test.cpp b/src/cvc/tests/state_transport_ipc_test.cpp index 26850cf4a..a08774b60 100644 --- a/src/cvc/tests/state_transport_ipc_test.cpp +++ b/src/cvc/tests/state_transport_ipc_test.cpp @@ -9,11 +9,17 @@ */ #include +#include #include #include +#include #include #include +#include #include +#include +#include +#include #include #include #include @@ -130,6 +136,119 @@ TEST(StateTransportIpcTest, TwoEndpointConvergence) { tB.stop(); } +namespace { +struct IpcDesc { + int w = 0, h = 0; + double fps = 0.0; +}; +std::vector enc_ipc_desc(const boost::any &v) { + const IpcDesc d = boost::any_cast(v); + std::vector b(sizeof(IpcDesc)); + std::memcpy(b.data(), &d, sizeof(IpcDesc)); + return b; +} +boost::any dec_ipc_desc(const std::vector &b) { + IpcDesc d{}; + if (b.size() == sizeof(IpcDesc)) + std::memcpy(&d, b.data(), sizeof(IpcDesc)); + return boost::any(d); +} +} // namespace + +// data() OBJECT replication over the IPC transport (S4 — closes §52's "both IPC and gRPC" testing +// bullet: gRPC has this coverage but compiles out by default, and the IPC tests only ever asserted +// value()/metadata). Proves the IPC wire serializes a Mutation's Payload (put_payload/read_payload) +// for a custom inline object, the builtin-int alias (S1) over a real transport, AND a blob_ref. +// Both endpoints live in one process, so a single shared blob store serves B's hydrator (no +// transport chunk-fetch; that async path is a separate follow-up, see §52 open questions). +TEST(StateTransportIpcTest, DataPayloadReplicatesOverIpc) { + cvc::app aA, aB; + aA.registerDataType("ipc_desc"); + aB.registerDataType("ipc_desc"); + + cvc::state_transport_ipc tA, tB; + auto pA = make_socket_path("A_data"); + auto pB = make_socket_path("B_data"); + tA.start(pA, "A", "C"); + tB.start(pB, "B", "C"); + ASSERT_TRUE(tA.connect_to_peer(pB, std::chrono::milliseconds(2000))); + ASSERT_TRUE(wait_connected(tA, tB, std::chrono::milliseconds(2000))); + + cvc::memory_state_blob_store store; // shared: both endpoints are in this one process + cvc::state_cluster_shard sA(aA, "C", "A"); + cvc::state_cluster_shard sB(aB, "C", "B"); + sA.set_blob_store(&store); + sB.set_blob_store(&store); + sA.set_max_inline_payload_bytes(256); + sB.set_max_inline_payload_bytes(256); + sA.codecs().register_codec("ipc_desc", enc_ipc_desc, dec_ipc_desc); + sB.codecs().register_codec("ipc_desc", enc_ipc_desc, dec_ipc_desc); + cvc::state_data_hydrator hydratorB(store, sB.codecs(), + &cvc::state_compression_registry::shared()); + sB.set_data_hydrator(&hydratorB); + sA.attach(); + sB.attach(); + tA.register_shard(&sA); + tB.register_shard(&sB); + + // INLINE: a small custom descriptor + a builtin int (the S1 alias), both ride inline bytes. value + // doubles as a sync barrier; data() has no equality guard so set it twice past the + // create-consume. + cvc::state &nd = cvc::state::instance(aA)("streams.cam.desc"); + nd.value(std::string("p1")); + nd.value(std::string("live")); + nd.data(boost::any(IpcDesc{1280, 720, 30.0})); + nd.data(boost::any(IpcDesc{1280, 720, 30.0})); + + cvc::state &ni = cvc::state::instance(aA)("nums.i"); + ni.value(std::string("p1")); + ni.value(std::string("p2")); + ni.data(boost::any(int{4242})); + ni.data(boost::any(int{4242})); + + // BLOB: a 4 KiB payload exceeds the 256-byte inline threshold, so it offloads to a chunked + // blob_ref on A; B reconstructs it from the shared store via the hydrator. + std::vector big(4096); + for (std::size_t i = 0; i < big.size(); ++i) + big[i] = static_cast(i % 251); + const std::string big_s(big.begin(), big.end()); // builtin "std::string" codec, > threshold + cvc::state &nb = cvc::state::instance(aA)("streams.cam.lut"); + nb.value(std::string("p1")); + nb.value(std::string("p2")); + nb.data(boost::any(big_s)); + nb.data(boost::any(big_s)); + + tA.pump_all(); + tA.flush(); + + ASSERT_TRUE(wait_until( + [&] { + return cvc::state::instance(aB)("streams.cam.desc").value() == "live" && + !cvc::state::instance(aB)("streams.cam.desc").data().empty() && + !cvc::state::instance(aB)("nums.i").data().empty() && + !cvc::state::instance(aB)("streams.cam.lut").data().empty(); + }, + std::chrono::milliseconds(3000))) + << "data() payload did not cross the IPC wire"; + + const IpcDesc got = boost::any_cast(cvc::state::instance(aB)("streams.cam.desc").data()); + EXPECT_EQ(got.w, 1280); + EXPECT_EQ(got.h, 720); + EXPECT_DOUBLE_EQ(got.fps, 30.0); + + EXPECT_EQ(boost::any_cast(cvc::state::instance(aB)("nums.i").data()), 4242) + << "builtin int data() (S1) did not replicate over IPC"; + + const std::string lut = + boost::any_cast(cvc::state::instance(aB)("streams.cam.lut").data()); + ASSERT_EQ(lut.size(), big_s.size()) << "blob_ref payload did not reconstruct over IPC"; + EXPECT_EQ(lut, big_s); + EXPECT_GT(store.size(), 0u) << "large data() was not offloaded to a blob"; + + tA.stop(); + tB.stop(); +} + TEST(StateTransportIpcTest, RoundTripSuppressedByDedup) { cvc::app aA, aB; cvc::state_transport_ipc tA, tB;