diff --git a/Cargo.lock b/Cargo.lock index 23aa084771..ad6d671ea5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -672,26 +672,6 @@ dependencies = [ "serde", ] -[[package]] -name = "bincode" -version = "2.0.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "36eaf5d7b090263e8150820482d5d93cd964a81e4019913c972f4edcc6edb740" -dependencies = [ - "bincode_derive", - "serde", - "unty", -] - -[[package]] -name = "bincode_derive" -version = "2.0.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bf95709a440f45e986983918d0e8a1f30a9b1df04918fc828670606804ac3c09" -dependencies = [ - "virtue", -] - [[package]] name = "bit-set" version = "0.5.3" @@ -1089,6 +1069,15 @@ dependencies = [ "cc", ] +[[package]] +name = "cobs" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0fa961b519f0b462e3a3b4a34b64d119eeaca1d59af726fe450bbba07a9fc0a1" +dependencies = [ + "thiserror 2.0.20", +] + [[package]] name = "codespan-reporting" version = "0.13.1" @@ -2019,7 +2008,6 @@ version = "1.0.0-rc.4" dependencies = [ "arrow", "axum", - "bincode 1.3.3", "ctrlc", "dora-coordinator-store", "dora-core", @@ -2050,7 +2038,6 @@ dependencies = [ name = "dora-coordinator-store" version = "1.0.0-rc.4" dependencies = [ - "bincode 2.0.1", "dora-message", "eyre", "libc", @@ -2097,7 +2084,6 @@ name = "dora-daemon" version = "1.0.0-rc.4" dependencies = [ "aligned-vec", - "bincode 1.3.3", "chrono", "clonable-command", "criterion 0.8.2", @@ -2255,7 +2241,6 @@ dependencies = [ "aligned-vec", "arrow-data", "arrow-schema", - "bincode 1.3.3", "chrono", "criterion 0.8.2", "dirs 6.0.0", @@ -2263,6 +2248,7 @@ dependencies = [ "getrandom 0.4.3", "libc", "log", + "postcard", "proptest", "schemars 1.2.2", "semver", @@ -2295,7 +2281,6 @@ dependencies = [ "arrow", "arrow-json", "arrow-schema", - "bincode 1.3.3", "chrono", "colored", "criterion 0.5.1", @@ -2470,7 +2455,6 @@ dependencies = [ name = "dora-replay-node" version = "1.0.0-rc.4" dependencies = [ - "bincode 1.3.3", "dora-message", "dora-node-api", "dora-recording", @@ -2731,6 +2715,18 @@ version = "1.17.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9e5e8f6c15a24b9a3ee5efec809ccd006d3b30e8b3bb63c39af737c7f87daa1d" +[[package]] +name = "embedded-io" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ef1a6892d9eef45c8fa6b9e0086428a2cca8491aca8f787c534a3d6d0bcb3ced" + +[[package]] +name = "embedded-io" +version = "0.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "edd0f118536f44f5ccd48bcb8b111bdc3de888b58c74639dfb034a357d0f206d" + [[package]] name = "emit-then-exit-source-node" version = "1.0.0-rc.4" @@ -5715,6 +5711,18 @@ dependencies = [ "portable-atomic", ] +[[package]] +name = "postcard" +version = "1.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6764c3b5dd454e283a30e6dfe78e9b31096d9e32036b5d1eaac7a6119ccb9a24" +dependencies = [ + "cobs", + "embedded-io 0.4.0", + "embedded-io 0.6.1", + "serde", +] + [[package]] name = "potential_utf" version = "0.1.5" @@ -8485,12 +8493,6 @@ version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" -[[package]] -name = "unty" -version = "0.0.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6d49784317cd0d1ee7ec5c716dd598ec5b4483ea832a2dced265471cc0f690ae" - [[package]] name = "unwind_safe" version = "0.1.0" @@ -8664,12 +8666,6 @@ version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" -[[package]] -name = "virtue" -version = "0.0.18" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "051eb1abcf10076295e815102942cc58f9d5e3b4560e46e53c21e8ff6f3af7b1" - [[package]] name = "vtparse" version = "0.6.2" @@ -9681,7 +9677,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4e4de171bff459816b1ca9e6657c4ae9783d2c1fc569e1721a380521425e897d" dependencies = [ "async-trait", - "bincode 1.3.3", + "bincode", "flume 0.11.1", "futures", "leb128", diff --git a/Cargo.toml b/Cargo.toml index e5bfa72d4a..7c34a2a79a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -186,7 +186,7 @@ uuid = { version = "1.23", features = ["serde", "v7"] } futures = { version = "0.3.32", default-features = false, features = ["std", "async-await"] } fs2 = "0.4.3" redb = "4.1" -bincode = "1.3.3" +postcard = { version = "1.1.3", default-features = false, features = ["use-std"] } flume = "0.12.0" tempfile = "3.27.0" proptest = "1.11" diff --git a/Changelog.md b/Changelog.md index 70dcee8506..14dc0aeac6 100644 --- a/Changelog.md +++ b/Changelog.md @@ -4,6 +4,13 @@ ### Breaking +- **Binary wire format moved from bincode to postcard.** [bincode is unmaintained](https://rustsec.org/advisories/RUSTSEC-2025-0141.html) — development stopped at 1.3.3 and all versions are flagged — so it is not something to carry into 1.0. Every binary plane moves to [postcard](https://docs.rs/postcard), which is serde-based (so the migration is encoding-only, no type changes), actively maintained, and unlike bincode has a **documented, stable wire spec** — the right property for a format 1.0 commits to. Messages shrink 25–27 bytes each (`Metadata` alone: 34 → 27 B) because postcard varint-encodes integers and length prefixes; serialization speed is at parity or slightly better once the encode buffer is pre-sized (`dora_message::to_vec_with_capacity`), and deserialization is unchanged to slightly faster. The JSON planes (CLI ↔ coordinator, coordinator ↔ daemon WebSocket) are untouched. This breaks four things, all of which fail loudly: + - **Mixed-version deployments.** `Metadata::CURRENT_VERSION` is bumped `1` → `2`, so a node and daemon on different sides of this change are rejected at register with a clear version-mismatch error. Upgrade daemon and nodes together. + - **Coordinator persisted store: `SCHEMA_VERSION` bumped `4` → `5`.** Stored records are postcard-encoded now, and postcard is positional rather than self-describing, so v4 rows cannot be read. `RedbStore::open()` rejects a v4-stamped database with a `schema version mismatch` error. **If you have an existing coordinator store** (`~/.dora/` by default; `redb` is the default `--store` backend), `dora coordinator` will refuse to start after upgrading. Delete the store file (its path is named in the error message) to start fresh, or pass `--store memory` to bypass persistence. + - **`.drec` recordings: format version bumped `1` → `2`.** Entry payloads are postcard-encoded. The container framing is unchanged, so a v1 file would otherwise pass the header check and then fail per-entry; the reader now rejects it up front, naming the cause. Re-record with this release — there is no converter. + - **The WebSocket topic-data channel** (`docs/websocket-topic-data-channel.md`) carries raw `Timestamped` bytes, now postcard. That channel has no version handshake and both encodings are positional, so a bincode-era third-party subscriber will *misparse* rather than error — match your subscriber to the dora release. + + `bincode` is gone from dora's own dependencies; it remains in the tree only transitively via `zenoh-ext`. See [RUSTSEC-2025-0141](https://rustsec.org/advisories/RUSTSEC-2025-0141.html). - **`dora-operator-api-cxx` operator interface gains `on_input_closed` and `on_stop`**: previously the C++ operator API silently dropped `Event::InputClosed { id }` and `Event::Stop` via a catch-all `_ => Continue` arm — operators had no way to react to upstream input closure or graceful shutdown. The cxx::bridge now declares two additional callbacks that the C++ side must implement: ```cpp DoraOnInputResult on_input_closed(Operator& op, rust::Str id, OutputSender& output_sender); diff --git a/apis/rust/node/Cargo.toml b/apis/rust/node/Cargo.toml index d37fb9409d..ccb6a9c271 100644 --- a/apis/rust/node/Cargo.toml +++ b/apis/rust/node/Cargo.toml @@ -21,7 +21,6 @@ eyre = { workspace = true } serde_yaml = { workspace = true } tracing = { workspace = true } flume = { workspace = true } -bincode = { workspace = true } zenoh = { workspace = true } zenoh-ext = { workspace = true } dora-tracing = { workspace = true, optional = true } diff --git a/apis/rust/node/src/daemon_connection/tcp.rs b/apis/rust/node/src/daemon_connection/tcp.rs index c08b1677b6..b259cb20fa 100644 --- a/apis/rust/node/src/daemon_connection/tcp.rs +++ b/apis/rust/node/src/daemon_connection/tcp.rs @@ -9,7 +9,7 @@ use std::{ }; enum Serializer { - Bincode, + Postcard, SerdeJson, } pub fn request( @@ -17,8 +17,8 @@ pub fn request( request: &Timestamped, ) -> eyre::Result { send_message(connection, request)?; - if request.inner.expects_tcp_bincode_reply() { - receive_reply(connection, Serializer::Bincode) + if request.inner.expects_tcp_binary_reply() { + receive_reply(connection, Serializer::Postcard) .and_then(|reply| reply.ok_or_else(|| eyre!("server disconnected unexpectedly"))) // Use serde json for message with variable length } else if request.inner.expects_tcp_json_reply() { @@ -33,7 +33,8 @@ fn send_message( connection: &mut TcpStream, message: &Timestamped, ) -> eyre::Result<()> { - let serialized = bincode::serialize(&message).wrap_err("failed to serialize DaemonRequest")?; + let serialized = dora_message::encode_presized(message, message.inner.encode_size_hint()) + .wrap_err("failed to serialize DaemonRequest")?; tcp_send(connection, &serialized).wrap_err("failed to send DaemonRequest")?; Ok(()) } @@ -57,7 +58,7 @@ fn receive_reply( }, }; match serializer { - Serializer::Bincode => bincode::deserialize(&raw) + Serializer::Postcard => dora_message::decode(&raw) .wrap_err("failed to deserialize DaemonReply") .map(Some), Serializer::SerdeJson => serde_json::from_slice(&raw) diff --git a/apis/rust/node/src/event_stream/data_conversion.rs b/apis/rust/node/src/event_stream/data_conversion.rs index 3088eed38c..9e0f33e6ae 100644 --- a/apis/rust/node/src/event_stream/data_conversion.rs +++ b/apis/rust/node/src/event_stream/data_conversion.rs @@ -45,7 +45,7 @@ mod tests { use arrow::array::{Array, Float32Array}; use dora_message::node_to_daemon::DataMessage; - /// The daemon/TCP fallback carries the IPC stream as a bincode-serialized + /// The daemon/TCP fallback carries the IPC stream as a postcard-serialized /// `DataMessage::Vec`. A round-trip through that serialization must preserve /// both the payload and its 128-byte alignment, so the receiver still /// decodes zero-copy via `decode_arrow_ipc_zero_copy`. @@ -60,9 +60,9 @@ mod tests { ipc_encode::encode_ipc_into(&data, &mut avec).unwrap(); let message = DataMessage::Vec(avec); - // bincode round-trip = what the node->daemon->node TCP hops do. - let bytes = bincode::serialize(&message).unwrap(); - let restored: DataMessage = bincode::deserialize(&bytes).unwrap(); + // Same encode/decode entry points the node->daemon->node TCP hops use. + let bytes = dora_message::encode(&message).unwrap(); + let restored: DataMessage = dora_message::decode(&bytes).unwrap(); let DataMessage::Vec(avec) = restored; assert_eq!( avec.as_ptr() as usize % 128, diff --git a/apis/rust/node/src/event_stream/mod.rs b/apis/rust/node/src/event_stream/mod.rs index 14a053d6e6..f75c5badb7 100644 --- a/apis/rust/node/src/event_stream/mod.rs +++ b/apis/rust/node/src/event_stream/mod.rs @@ -159,7 +159,7 @@ fn spawn_startup_acker( node_id.as_ref(), input_id.as_ref(), ); - let attachment = match bincode::serialize(&metadata) { + let attachment = match dora_message::encode(&metadata) { Ok(bytes) => bytes, Err(e) => { tracing::debug!(input = %input_id, "failed to serialize startup ack ({e})"); @@ -490,7 +490,7 @@ impl EventStream { use dora_message::metadata::Metadata; let metadata = match sample.attachment() { Some(att) => { - match bincode::deserialize::(&att.to_bytes()) + match dora_message::decode::(&att.to_bytes()) { // A version mismatch that still happens // to deserialize: reject with a clear @@ -514,7 +514,7 @@ impl EventStream { // A pre-1.0 peer (old ArrowTypeInfo // sidecar layout) misaligns here; name // the likely cause so the failure isn't - // a bare bincode error. + // a bare deserialization error. tracing::warn!( "zenoh metadata deserialization failed \ (possibly a peer using an incompatible \ @@ -1921,14 +1921,14 @@ mod tests { } /// Regression test for the daemon↔node wire protocol: `NodeEvent` - /// is sent over TCP with bincode, so any field type that uses + /// is sent over TCP with postcard, so any field type that uses /// `Deserializer::deserialize_any` (like `serde_json::Value`) /// breaks the channel and kills the node at the next receive. /// `NodeEvent::ParamUpdate` carries its value as JSON-encoded /// bytes for that reason. This test pins the invariant so we /// don't regress back to a `deserialize_any` field. #[test] - fn node_event_param_update_round_trips_through_bincode() { + fn node_event_param_update_round_trips_through_postcard() { let cases = [ serde_json::json!(42), serde_json::json!(1.5), @@ -1942,8 +1942,8 @@ mod tests { key: "rate".into(), value_json: serde_json::to_vec(&value).unwrap(), }; - let bytes = bincode::serialize(&event).expect("bincode serialize"); - let back: NodeEvent = bincode::deserialize(&bytes).expect("bincode deserialize"); + let bytes = dora_message::encode(&event).expect("serialize"); + let back: NodeEvent = dora_message::decode(&bytes).expect("deserialize"); match back { NodeEvent::ParamUpdate { key, value_json } => { assert_eq!(key, "rate"); diff --git a/apis/rust/node/src/node/mod.rs b/apis/rust/node/src/node/mod.rs index 58d2983108..1b06bbed86 100644 --- a/apis/rust/node/src/node/mod.rs +++ b/apis/rust/node/src/node/mod.rs @@ -406,7 +406,7 @@ fn declare_ack_subscribers( let Some(attachment) = sample.attachment() else { return; }; - let Ok(metadata) = bincode::deserialize::(&attachment.to_bytes()) else { + let Ok(metadata) = dora_message::decode::(&attachment.to_bytes()) else { // Not a dora ack (foreign publisher on the ack key): ignore. return; }; @@ -524,7 +524,7 @@ impl StartupHandshake { continue; }; let metadata = Metadata::startup_marker(clock.new_timestamp()); - let attachment = match bincode::serialize(&metadata) { + let attachment = match dora_message::encode(&metadata) { Ok(bytes) => bytes, Err(e) => { debug!(output = %state.output_id, "failed to serialize startup marker ({e})"); @@ -1880,7 +1880,7 @@ impl DoraNode { .expect("a declared publisher implies a zenoh session"); // Serialize metadata as zenoh attachment. - let metadata_bytes = match bincode::serialize(metadata) { + let metadata_bytes = match dora_message::encode(metadata) { Ok(bytes) => bytes, Err(e) => { tracing::warn!(output = %output_id, "failed to serialize metadata ({e}); falling back to daemon path"); @@ -3005,7 +3005,7 @@ fn publish_schema_once( metadata .parameters .insert(SCHEMA_HASH.to_string(), Parameter::Integer(hash as i64)); - bincode::serialize(&metadata).ok() + dora_message::encode(&metadata).ok() } } } diff --git a/binaries/cli/src/command/record.rs b/binaries/cli/src/command/record.rs index d714f6529c..32d9fa5f58 100644 --- a/binaries/cli/src/command/record.rs +++ b/binaries/cli/src/command/record.rs @@ -399,7 +399,7 @@ fn run_record_proxy(args: Record) -> eyre::Result<()> { .as_nanos() as u64; let header = RecordingHeader { - version: 1, + version: dora_recording::FORMAT_VERSION, start_nanos, dataflow_id, descriptor_yaml: yaml_bytes, @@ -456,7 +456,7 @@ fn run_record_proxy(args: Record) -> eyre::Result<()> { match data_rx.recv_timeout(std::time::Duration::from_millis(100)) { Ok(Ok(payload)) => { - // The payload is already `Timestamped` bincode bytes. + // The payload is already `Timestamped` postcard bytes. // Parse it to extract node_id and output_id for the recording entry. let event = match Timestamped::deserialize_inter_daemon_event(&payload) { Ok(e) => e, diff --git a/binaries/coordinator/Cargo.toml b/binaries/coordinator/Cargo.toml index 77507913a8..3d0913d12b 100644 --- a/binaries/coordinator/Cargo.toml +++ b/binaries/coordinator/Cargo.toml @@ -39,7 +39,6 @@ tower-http = { version = "0.7", features = ["cors"] } indexmap = "2" itertools = "0.15.0" zenoh = { workspace = true } -bincode = { workspace = true } arrow = { workspace = true, features = ["ipc"] } opentelemetry = { version = "0.32", features = ["metrics"], optional = true } dora-metrics = { workspace = true, optional = true } diff --git a/binaries/daemon/Cargo.toml b/binaries/daemon/Cargo.toml index f17996127c..252b0c2fcc 100644 --- a/binaries/daemon/Cargo.toml +++ b/binaries/daemon/Cargo.toml @@ -40,7 +40,6 @@ serde_yaml = { workspace = true } uuid = { workspace = true } futures = { workspace = true } shared_memory_extended = "0.13.0" -bincode = { workspace = true } aligned-vec = "0.6.4" ctrlc = "3.5.2" which = "8" diff --git a/binaries/daemon/src/lib.rs b/binaries/daemon/src/lib.rs index c68a8ea381..db58a3bad4 100644 --- a/binaries/daemon/src/lib.rs +++ b/binaries/daemon/src/lib.rs @@ -4154,13 +4154,11 @@ impl Daemon { future::Either::Right((sample, f)) => { finished = f; let Ok(sample) = sample else { break }; - // Node publishes raw payload + bincode metadata + // Node publishes raw payload + encoded metadata // attachment (see `DoraNode::zenoh_publish`). + use dora_message::metadata::Metadata; let Some(mut metadata) = sample.attachment().and_then(|a| { - bincode::deserialize::( - &a.to_bytes(), - ) - .ok() + dora_message::decode::(&a.to_bytes()).ok() }) else { continue; }; diff --git a/binaries/daemon/src/local_listener.rs b/binaries/daemon/src/local_listener.rs index ed8e91b7a6..85be675ab7 100644 --- a/binaries/daemon/src/local_listener.rs +++ b/binaries/daemon/src/local_listener.rs @@ -164,7 +164,7 @@ async fn receive_message( } }, }; - bincode::deserialize(&raw) + dora_message::decode(&raw) .wrap_err("failed to deserialize DaemonRequest") .map(Some) } diff --git a/binaries/daemon/src/node_communication/tcp.rs b/binaries/daemon/src/node_communication/tcp.rs index 54502e0272..e324305d7f 100644 --- a/binaries/daemon/src/node_communication/tcp.rs +++ b/binaries/daemon/src/node_communication/tcp.rs @@ -105,7 +105,7 @@ impl Connection for TcpConnection { } }, }; - bincode::deserialize(&raw) + dora_message::decode(&raw) .wrap_err("failed to deserialize DaemonRequest") .map(Some) } @@ -115,8 +115,8 @@ impl Connection for TcpConnection { // don't send empty replies return Ok(()); } - let serialized = - bincode::serialize(&message).wrap_err("failed to serialize DaemonReply")?; + let serialized = dora_message::encode_presized(&message, message.encode_size_hint()) + .wrap_err("failed to serialize DaemonReply")?; socket_stream_send(&mut self.0, &serialized) .await .wrap_err("failed to send DaemonReply")?; diff --git a/binaries/record-node/src/main.rs b/binaries/record-node/src/main.rs index b86037bdac..ce97b0c614 100644 --- a/binaries/record-node/src/main.rs +++ b/binaries/record-node/src/main.rs @@ -58,7 +58,7 @@ fn main() -> eyre::Result<()> { .as_nanos() as u64; let header = RecordingHeader { - version: 1, + version: dora_recording::FORMAT_VERSION, start_nanos, dataflow_id: uuid::Uuid::new_v4(), descriptor_yaml: descriptor_yaml.into_bytes(), diff --git a/binaries/replay-node/Cargo.toml b/binaries/replay-node/Cargo.toml index d2f17e74fa..8b502e6ef4 100644 --- a/binaries/replay-node/Cargo.toml +++ b/binaries/replay-node/Cargo.toml @@ -18,4 +18,3 @@ dora-node-api = { workspace = true, default-features = false } dora-recording = { workspace = true } dora-message = { workspace = true } eyre = { workspace = true } -bincode = { workspace = true } diff --git a/binaries/replay-node/src/main.rs b/binaries/replay-node/src/main.rs index a62404e665..d869b8fb6b 100644 --- a/binaries/replay-node/src/main.rs +++ b/binaries/replay-node/src/main.rs @@ -89,9 +89,8 @@ fn main() -> eyre::Result<()> { } prev_offset = entry.timestamp_offset_nanos; - // Deserialize the InterDaemonEvent from raw bincode let timestamped: Timestamped = - match bincode::deserialize(&entry.event_bytes) { + match Timestamped::deserialize_inter_daemon_event(&entry.event_bytes) { Ok(event) => event, Err(e) => { eprintln!( @@ -119,7 +118,7 @@ fn main() -> eyre::Result<()> { Err(e) => { // A single corrupt-but-complete payload must not // abort the whole replay. Skip it, matching the - // bincode branch above and the recording layer's + // event-decode branch above and the recording layer's // torn-record resilience (dropping a bad record // rather than failing the run). eprintln!( diff --git a/deny.toml b/deny.toml index 7a4d6f77e3..db3f53d048 100644 --- a/deny.toml +++ b/deny.toml @@ -18,7 +18,7 @@ yanked = "deny" ignore = [ # --- Unmaintained transitive dependencies --- # All waived pending upstream migration by parent crates. Review quarterly. - "RUSTSEC-2025-0141", # bincode 1.x + 2.x unmaintained. Workspace dep; migration to postcard/bincode2 planned post-1.0. (bincode 3.0.0 is a compile_error! protest stub, not an upgrade target.) Review: 2026-09. + "RUSTSEC-2025-0141", # bincode unmaintained. No longer a dora dep (migrated to postcard); now transitive via zenoh-ext 1.9 only. Waiting for zenoh to migrate. Review: 2026-09. "RUSTSEC-2024-0436", # paste: transitive (safer_ffi / macro_rules_attribute). Low risk. Review: 2026-07. "RUSTSEC-2025-0134", # rustls-pemfile unmaintained (thin wrapper over rustls-pki-types). Transitive via zenoh-link-commons 1.9; waiting for zenoh to migrate to rustls-pki-types PemObject. Review: 2026-09. # --- Temporary vulnerability waivers --- diff --git a/docs/architecture.md b/docs/architecture.md index c3e98c563f..cc90372b6b 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -69,7 +69,7 @@ independently. |------|-------|------| | `libraries/message` | dora-message | All inter-component message types, protocol definitions, Arrow metadata | | `libraries/core` | dora-core | Dataflow descriptor parsing, build utilities, Zenoh config | -| `libraries/recording` | dora-recording | Recording format (.drec): bincode header + entries + footer | +| `libraries/recording` | dora-recording | Recording format (.drec): binary header + entries + footer | | `libraries/arrow-convert` | dora-arrow-convert | Arrow type conversions (numeric, datetime) | | `libraries/coordinator-store` | dora-coordinator-store | State persistence for coordinator (in-memory or redb backend) | @@ -231,7 +231,7 @@ Nodes are standalone processes that communicate with the daemon. | Default port | 6013 | | Auth | Bearer token in `Authorization` header | | Control messages | JSON text frames (request/response/event) | -| Topic data | Binary frames: `[16-byte UUID][bincode payload]` | +| Topic data | Binary frames: `[16-byte UUID][postcard payload]` | | Rate limit | 20 connections per IP per 60s | | Max connections | 256 | @@ -304,7 +304,7 @@ Three transport options, configured via `LocalCommunicationConfig`: | Router port | 7447 | | Peer port | 5456 | | Routing | linkstate | -| Serialization | bincode | +| Serialization | postcard | **Topic pattern:** ``` @@ -727,7 +727,7 @@ _unstable_deploy: ├─ output_id: [u8; output_id_len] ├─ timestamp_offset_nanos: u64 LE ├─ event_bytes_len: u32 LE -└─ event_bytes: [u8; event_bytes_len] (bincode InterDaemonEvent) +└─ event_bytes: [u8; event_bytes_len] (postcard InterDaemonEvent) [FOOTER] (optional, written on clean finish) ├─ FOOTER_MAGIC: 8 bytes ("DORAEND") @@ -814,7 +814,7 @@ File download utility for fetching operator/node binaries from HTTP URLs. Saniti | `DORA_COORDINATOR_PORT_WS_DEFAULT` | 6013 | Coordinator WebSocket port | | `DORA_DAEMON_LOCAL_LISTEN_PORT_DEFAULT` | 53291 | Daemon TCP listener port | | `ZERO_COPY_THRESHOLD` | 4096 bytes | Shared memory activation | -| `MAX_MESSAGE_BYTES` | 64 MiB | Max TCP/bincode message | +| `MAX_MESSAGE_BYTES` | 64 MiB | Max TCP/postcard message | | `MAX_CONTROL_MESSAGE_BYTES` | 1 MiB | Max control plane JSON message | | `TCP_READ_TIMEOUT` | 30 seconds | Socket read timeout | | `MAX_WS_CONNECTIONS` | 256 | Concurrent WebSocket limit | @@ -873,7 +873,7 @@ pub enum NodeStatus { |---------|--------|-------| | CLI ↔ Coordinator | JSON text frames | Preserves u128 for HLC timestamps | | Coordinator ↔ Daemon | JSON text frames | Direct string serialization | -| Daemon ↔ Node (TCP) | bincode over length-prefixed frames | 8-byte LE length prefix | -| Daemon ↔ Node (shmem) | bincode via shared memory | Atomic synchronization | -| Daemon ↔ Daemon | bincode over Zenoh | Apache Arrow data format | -| Recording | bincode entries in .drec | Custom binary container | +| Daemon ↔ Node (TCP) | postcard over length-prefixed frames | 8-byte LE length prefix | +| Daemon ↔ Node (shmem) | postcard via shared memory | Atomic synchronization | +| Daemon ↔ Daemon | postcard over Zenoh | Apache Arrow data format | +| Recording | postcard entries in .drec | Custom binary container | diff --git a/docs/debugging.md b/docs/debugging.md index 0e179bd35a..e55a104880 100644 --- a/docs/debugging.md +++ b/docs/debugging.md @@ -262,13 +262,13 @@ The `.drec` format is a simple binary file: ``` ┌──────────────────────────────────┐ -│ Header (bincode) │ +│ Header (binary) │ │ version: u32 │ │ start_nanos: u64 │ │ dataflow_id: Uuid │ │ descriptor_yaml: Vec │ ├──────────────────────────────────┤ -│ Entry 1 (bincode) │ +│ Entry 1 (binary) │ │ node_id: String │ │ output_id: String │ │ timestamp_offset_nanos: u64 │ @@ -278,13 +278,13 @@ The `.drec` format is a simple binary file: ├──────────────────────────────────┤ │ ... │ ├──────────────────────────────────┤ -│ Footer (bincode) │ +│ Footer (binary) │ │ total_messages: u64 │ │ total_bytes: u64 │ └──────────────────────────────────┘ ``` -The `event_bytes` field contains the raw `Timestamped` bincode payload -- the same format used on the wire between daemons. The `descriptor_yaml` in the header stores the original dataflow descriptor so replay can reconstruct the dataflow. +The container framing (header, per-entry lengths, footer) is hand-written little-endian fields; the `event_bytes` field contains the raw `Timestamped` postcard payload -- the same format used on the wire between daemons. Recordings written before dora 1.0 used bincode for `event_bytes` and carry format version 1; this release refuses to open them. The `descriptor_yaml` in the header stores the original dataflow descriptor so replay can reconstruct the dataflow. --- diff --git a/docs/fault-tolerance.md b/docs/fault-tolerance.md index 71ef29b116..0cb33a535a 100644 --- a/docs/fault-tolerance.md +++ b/docs/fault-tolerance.md @@ -434,9 +434,9 @@ The store tracks three record types: |--------|-----|-----------------| | `DataflowRecord` | UUID (16 bytes) | uuid, name, descriptor (JSON), status, daemon IDs, generation counter, created/updated timestamps | | `BuildRecord` | UUID (16 bytes) | build ID, status, errors, created/updated timestamps | -| `DaemonInfo` | DaemonId (bincode) | daemon ID, machine ID | +| `DaemonInfo` | DaemonId (postcard) | daemon ID, machine ID | -Records are serialized with [bincode](https://docs.rs/bincode/2) for compact, fast encoding. +Records are serialized with [postcard](https://docs.rs/postcard) for compact, fast encoding. ### Dataflow Status Lifecycle @@ -518,9 +518,9 @@ pub trait CoordinatorStore: Send + Sync { } ``` -The `RedbStore` implementation uses three redb tables (`daemons`, `dataflows`, `builds`) with UUID-based binary keys and bincode-serialized values. All operations are synchronous (redb is a synchronous library); the coordinator calls them directly from the async event loop since they are fast in-process operations. +The `RedbStore` implementation uses three redb tables (`daemons`, `dataflows`, `builds`) with UUID-based binary keys and postcard-serialized values. All operations are synchronous (redb is a synchronous library); the coordinator calls them directly from the async event loop since they are fast in-process operations. -A bincode deserialization limit of 64 MiB guards against corrupted data that could encode huge allocation sizes in length prefixes. +A 64 MiB record-size limit is enforced symmetrically on both encode and decode, so a record can never be written that the reader would later refuse. postcard reads only from the slice it is handed and does not pre-allocate from a length prefix, so a corrupt row cannot drive a large allocation. --- diff --git a/docs/performance.md b/docs/performance.md index c5f6987213..6735bbab53 100644 --- a/docs/performance.md +++ b/docs/performance.md @@ -8,7 +8,7 @@ Dora achieves 10-17x lower latency than ROS2 Python through zero-copy shared mem |-------|-------|---------------| | Runtime | Rust async (tokio) | Python + C++ middleware | | IPC (>4KB) | Zenoh SHM zero-copy | DDS serialization + copy | -| IPC (<4KB) | TCP with bincode | DDS serialization + copy | +| IPC (<4KB) | TCP with postcard | DDS serialization + copy | | Data format | Apache Arrow (zero-serde) + optional IPC framing | CDR serialization | | Threading | Lock-free channels (flume) | GIL-bound callbacks | | Fan-out | Arc-wrapped (O(1) per receiver) | Per-receiver copy | diff --git a/docs/websocket-topic-data-channel.md b/docs/websocket-topic-data-channel.md index 078b8a1d5b..e5437599ea 100644 --- a/docs/websocket-topic-data-channel.md +++ b/docs/websocket-topic-data-channel.md @@ -99,14 +99,14 @@ After the handshake, the coordinator pushes binary WS frames. Each frame has a f 0 16 N ├───────────────────┼──────────────────────────────┤ │ subscription_id │ Timestamped│ - │ (16 bytes UUID) │ (bincode serialized) │ + │ (16 bytes UUID) │ (postcard serialized) │ └───────────────────┴──────────────────────────────┘ ``` | Field | Size | Description | |-------|------|-------------| | `subscription_id` | 16 bytes | UUID matching the `TopicSubscribed` ack, for multiplexing | -| payload | variable | Raw `Timestamped` bincode bytes from Zenoh | +| payload | variable | Raw `Timestamped` postcard bytes from Zenoh | The 16-byte UUID prefix allows multiplexing multiple subscriptions on a single WS connection without additional framing overhead. @@ -248,7 +248,9 @@ The coordinator subscribes to Zenoh topics using the format from `dora_core::top dora/{dataflow_id}/{node_id}/{data_id} ``` -Each topic carries `Timestamped` as its payload, serialized with bincode. The coordinator forwards these bytes as-is (prepended with subscription UUID) -- no re-serialization. +Each topic carries `Timestamped` as its payload, serialized with [postcard](https://docs.rs/postcard) -- a compact, non-self-describing binary format with a documented, stable wire spec. The coordinator forwards these bytes as-is (prepended with subscription UUID) -- no re-serialization. + +> **Wire break (dora 1.0):** this payload was bincode-encoded before 1.0. The encoding is positional in both formats, so a bincode-era subscriber does not fail cleanly against postcard frames -- it misparses them. There is no version handshake on this channel; match your subscriber to the dora release. --- @@ -258,7 +260,7 @@ Each topic carries `Timestamped` as its payload, serialized wi |-----------|-------|-----------| | Binary frame channel capacity | 64 | Balance between latency and memory | | Drop policy | Drop on full | Prefer freshness over completeness | -| Binary format | Raw bincode (no base64) | Avoid 33% overhead for large payloads | +| Binary format | Raw postcard (no base64) | Avoid 33% overhead for large payloads | For high-throughput topics (camera images, point clouds), the binary frame channel may fill up if the WS connection is slow. Dropped samples are silent -- the CLI will show reduced frequency in `topic hz` but won't stall. diff --git a/guide/src/advanced/ws-topic.md b/guide/src/advanced/ws-topic.md index 078b8a1d5b..e5437599ea 100644 --- a/guide/src/advanced/ws-topic.md +++ b/guide/src/advanced/ws-topic.md @@ -99,14 +99,14 @@ After the handshake, the coordinator pushes binary WS frames. Each frame has a f 0 16 N ├───────────────────┼──────────────────────────────┤ │ subscription_id │ Timestamped│ - │ (16 bytes UUID) │ (bincode serialized) │ + │ (16 bytes UUID) │ (postcard serialized) │ └───────────────────┴──────────────────────────────┘ ``` | Field | Size | Description | |-------|------|-------------| | `subscription_id` | 16 bytes | UUID matching the `TopicSubscribed` ack, for multiplexing | -| payload | variable | Raw `Timestamped` bincode bytes from Zenoh | +| payload | variable | Raw `Timestamped` postcard bytes from Zenoh | The 16-byte UUID prefix allows multiplexing multiple subscriptions on a single WS connection without additional framing overhead. @@ -248,7 +248,9 @@ The coordinator subscribes to Zenoh topics using the format from `dora_core::top dora/{dataflow_id}/{node_id}/{data_id} ``` -Each topic carries `Timestamped` as its payload, serialized with bincode. The coordinator forwards these bytes as-is (prepended with subscription UUID) -- no re-serialization. +Each topic carries `Timestamped` as its payload, serialized with [postcard](https://docs.rs/postcard) -- a compact, non-self-describing binary format with a documented, stable wire spec. The coordinator forwards these bytes as-is (prepended with subscription UUID) -- no re-serialization. + +> **Wire break (dora 1.0):** this payload was bincode-encoded before 1.0. The encoding is positional in both formats, so a bincode-era subscriber does not fail cleanly against postcard frames -- it misparses them. There is no version handshake on this channel; match your subscriber to the dora release. --- @@ -258,7 +260,7 @@ Each topic carries `Timestamped` as its payload, serialized wi |-----------|-------|-----------| | Binary frame channel capacity | 64 | Balance between latency and memory | | Drop policy | Drop on full | Prefer freshness over completeness | -| Binary format | Raw bincode (no base64) | Avoid 33% overhead for large payloads | +| Binary format | Raw postcard (no base64) | Avoid 33% overhead for large payloads | For high-throughput topics (camera images, point clouds), the binary frame channel may fill up if the WS connection is slow. Dropped samples are silent -- the CLI will show reduced frequency in `topic hz` but won't stall. diff --git a/guide/src/concepts/architecture.md b/guide/src/concepts/architecture.md index aad9586307..1232c1e00c 100644 --- a/guide/src/concepts/architecture.md +++ b/guide/src/concepts/architecture.md @@ -50,7 +50,7 @@ All crates share the workspace version. |------|-------|------| | `libraries/message` | dora-message | All inter-component message types, protocol definitions, Arrow metadata | | `libraries/core` | dora-core | Dataflow descriptor parsing, build utilities, Zenoh config | -| `libraries/recording` | dora-recording | Recording format (.drec): bincode header + entries + footer | +| `libraries/recording` | dora-recording | Recording format (.drec): binary header + entries + footer | | `libraries/arrow-convert` | dora-arrow-convert | Arrow type conversions (numeric, datetime) | | `libraries/coordinator-store` | dora-coordinator-store | State persistence for coordinator (in-memory or redb backend) | @@ -212,7 +212,7 @@ Nodes are standalone processes that communicate with the daemon. | Default port | 6013 | | Auth | Bearer token in `Authorization` header | | Control messages | JSON text frames (request/response/event) | -| Topic data | Binary frames: `[16-byte UUID][bincode payload]` | +| Topic data | Binary frames: `[16-byte UUID][postcard payload]` | | Rate limit | 20 connections per IP per 60s | | Max connections | 256 | @@ -252,7 +252,7 @@ Three transport options, configured via `LocalCommunicationConfig`: **TCP** (default): - Binds `127.0.0.1:0` (ephemeral port), `TCP_NODELAY` enabled -- Frame format: `[8-byte u64 LE length][bincode payload]` +- Frame format: `[8-byte u64 LE length][postcard payload]` - Max message: 64 MiB, read timeout: 30s **Shared Memory** (zero-copy): @@ -274,7 +274,7 @@ Three transport options, configured via `LocalCommunicationConfig`: | Router port | 7447 | | Peer port | 5456 | | Routing | linkstate | -| Serialization | bincode | +| Serialization | postcard | **Topic pattern:** ``` @@ -697,7 +697,7 @@ _unstable_deploy: ├─ output_id: [u8; output_id_len] ├─ timestamp_offset_nanos: u64 LE ├─ event_bytes_len: u32 LE -└─ event_bytes: [u8; event_bytes_len] (bincode InterDaemonEvent) +└─ event_bytes: [u8; event_bytes_len] (postcard InterDaemonEvent) [FOOTER] (optional, written on clean finish) ├─ FOOTER_MAGIC: 8 bytes ("DORAEND") @@ -784,7 +784,7 @@ File download utility for fetching operator/node binaries from HTTP URLs. Saniti | `DORA_COORDINATOR_PORT_WS_DEFAULT` | 6013 | Coordinator WebSocket port | | `DORA_DAEMON_LOCAL_LISTEN_PORT_DEFAULT` | 53291 | Daemon TCP listener port | | `ZERO_COPY_THRESHOLD` | 4096 bytes | Shared memory activation | -| `MAX_MESSAGE_BYTES` | 64 MiB | Max TCP/bincode message | +| `MAX_MESSAGE_BYTES` | 64 MiB | Max TCP/postcard message | | `MAX_CONTROL_MESSAGE_BYTES` | 1 MiB | Max control plane JSON message | | `TCP_READ_TIMEOUT` | 30 seconds | Socket read timeout | | `MAX_WS_CONNECTIONS` | 256 | Concurrent WebSocket limit | @@ -843,7 +843,7 @@ pub enum NodeStatus { |---------|--------|-------| | CLI ↔ Coordinator | JSON text frames | Preserves u128 for HLC timestamps | | Coordinator ↔ Daemon | JSON text frames | Direct string serialization | -| Daemon ↔ Node (TCP) | bincode over length-prefixed frames | 8-byte LE length prefix | -| Daemon ↔ Node (shmem) | bincode via shared memory | Atomic synchronization | -| Daemon ↔ Daemon | bincode over Zenoh | Apache Arrow data format | -| Recording | bincode entries in .drec | Custom binary container | +| Daemon ↔ Node (TCP) | postcard over length-prefixed frames | 8-byte LE length prefix | +| Daemon ↔ Node (shmem) | postcard via shared memory | Atomic synchronization | +| Daemon ↔ Daemon | postcard over Zenoh | Apache Arrow data format | +| Recording | postcard entries in .drec | Custom binary container | diff --git a/guide/src/operations/debugging.md b/guide/src/operations/debugging.md index 7669829846..33b4fa2aef 100644 --- a/guide/src/operations/debugging.md +++ b/guide/src/operations/debugging.md @@ -236,13 +236,13 @@ The `.drec` format is a simple binary file: ``` ┌──────────────────────────────────┐ -│ Header (bincode) │ +│ Header (binary) │ │ version: u32 │ │ start_nanos: u64 │ │ dataflow_id: Uuid │ │ descriptor_yaml: Vec │ ├──────────────────────────────────┤ -│ Entry 1 (bincode) │ +│ Entry 1 (binary) │ │ node_id: String │ │ output_id: String │ │ timestamp_offset_nanos: u64 │ @@ -252,13 +252,13 @@ The `.drec` format is a simple binary file: ├──────────────────────────────────┤ │ ... │ ├──────────────────────────────────┤ -│ Footer (bincode) │ +│ Footer (binary) │ │ total_messages: u64 │ │ total_bytes: u64 │ └──────────────────────────────────┘ ``` -The `event_bytes` field contains the raw `Timestamped` bincode payload -- the same format used on the wire between daemons. The `descriptor_yaml` in the header stores the original dataflow descriptor so replay can reconstruct the dataflow. +The container framing (header, per-entry lengths, footer) is hand-written little-endian fields; the `event_bytes` field contains the raw `Timestamped` postcard payload -- the same format used on the wire between daemons. Recordings written before dora 1.0 used bincode for `event_bytes` and carry format version 1; this release refuses to open them. The `descriptor_yaml` in the header stores the original dataflow descriptor so replay can reconstruct the dataflow. --- diff --git a/guide/src/operations/fault-tolerance.md b/guide/src/operations/fault-tolerance.md index 94bd0ae306..8ec7c0ae61 100644 --- a/guide/src/operations/fault-tolerance.md +++ b/guide/src/operations/fault-tolerance.md @@ -430,9 +430,9 @@ The store tracks three record types: |--------|-----|-----------------| | `DataflowRecord` | UUID (16 bytes) | uuid, name, descriptor (JSON), status, daemon IDs, generation counter, created/updated timestamps | | `BuildRecord` | UUID (16 bytes) | build ID, status, errors, created/updated timestamps | -| `DaemonInfo` | DaemonId (bincode) | daemon ID, machine ID | +| `DaemonInfo` | DaemonId (postcard) | daemon ID, machine ID | -Records are serialized with [bincode](https://docs.rs/bincode/2) for compact, fast encoding. +Records are serialized with [postcard](https://docs.rs/postcard) for compact, fast encoding. ### Dataflow Status Lifecycle @@ -514,9 +514,9 @@ pub trait CoordinatorStore: Send + Sync { } ``` -The `RedbStore` implementation uses three redb tables (`daemons`, `dataflows`, `builds`) with UUID-based binary keys and bincode-serialized values. All operations are synchronous (redb is a synchronous library); the coordinator calls them directly from the async event loop since they are fast in-process operations. +The `RedbStore` implementation uses three redb tables (`daemons`, `dataflows`, `builds`) with UUID-based binary keys and postcard-serialized values. All operations are synchronous (redb is a synchronous library); the coordinator calls them directly from the async event loop since they are fast in-process operations. -A bincode deserialization limit of 64 MiB guards against corrupted data that could encode huge allocation sizes in length prefixes. +A 64 MiB record-size limit is enforced symmetrically on both encode and decode, so a record can never be written that the reader would later refuse. postcard reads only from the slice it is handed and does not pre-allocate from a length prefix, so a corrupt row cannot drive a large allocation. --- diff --git a/guide/src/operations/performance.md b/guide/src/operations/performance.md index f5003ddf9e..ed26aed7a2 100644 --- a/guide/src/operations/performance.md +++ b/guide/src/operations/performance.md @@ -8,7 +8,7 @@ Dora achieves 10-17x lower latency than ROS2 Python through zero-copy shared mem |-------|-------|---------------| | Runtime | Rust async (tokio) | Python + C++ middleware | | IPC (>4KB) | Zenoh SHM zero-copy | DDS serialization + copy | -| IPC (<4KB) | TCP with bincode | DDS serialization + copy | +| IPC (<4KB) | TCP with postcard | DDS serialization + copy | | Data format | Apache Arrow (zero-serde) | CDR serialization | | Threading | Lock-free channels (flume) | GIL-bound callbacks | | Fan-out | Arc-wrapped (O(1) per receiver) | Per-receiver copy | diff --git a/libraries/coordinator-store/Cargo.toml b/libraries/coordinator-store/Cargo.toml index 273b6b5a26..98ac219ab4 100644 --- a/libraries/coordinator-store/Cargo.toml +++ b/libraries/coordinator-store/Cargo.toml @@ -10,11 +10,10 @@ repository.workspace = true [features] default = [] -redb-backend = ["dep:redb", "dep:bincode", "dep:libc", "dep:tracing"] +redb-backend = ["dep:redb", "dep:libc", "dep:tracing"] [dependencies] dora-message = { workspace = true } -bincode = { version = "2.0", features = ["serde"], optional = true } eyre = { workspace = true } libc = { version = "0.2", optional = true } redb = { workspace = true, optional = true } diff --git a/libraries/coordinator-store/src/lib.rs b/libraries/coordinator-store/src/lib.rs index 3d2d49da1e..ced7f090d5 100644 --- a/libraries/coordinator-store/src/lib.rs +++ b/libraries/coordinator-store/src/lib.rs @@ -143,9 +143,9 @@ pub enum DataflowStatus { /// store-vs-CLI view across coordinator restarts. /// /// `#[serde(default)]` documents the intended semantics for a - /// missing value, but with the bincode encoding used by + /// missing value, but with the postcard encoding used by /// `RedbStore` it does NOT make old bytes decodable on its own - /// (bincode is not self-describing, so a missing trailing field + /// (postcard is not self-describing, so a missing trailing field /// fails to decode rather than falling back to `Default`). Records /// written before this field existed require a `SCHEMA_VERSION` /// bump (see `redb_store.rs`) so old databases are rejected at diff --git a/libraries/coordinator-store/src/redb_store.rs b/libraries/coordinator-store/src/redb_store.rs index e197fbc011..4bdf9fdc47 100644 --- a/libraries/coordinator-store/src/redb_store.rs +++ b/libraries/coordinator-store/src/redb_store.rs @@ -18,16 +18,16 @@ const NODE_PARAMS: TableDefinition<&str, &[u8]> = TableDefinition::new("node_par /// Bump this when the serialization format of any stored record changes. /// -/// bincode is not self-describing, so `#[serde(default)]` on a struct/enum +/// postcard is not self-describing, so `#[serde(default)]` on a struct/enum /// field does NOT make old persisted bytes decodable -- it only fills in the /// field when the deserializer already knows the field is absent, which for -/// bincode's positional encoding never happens (a missing trailing field +/// postcard's positional encoding never happens (a missing trailing field /// reads past the end of the buffer and fails). Any change to a persisted /// type's shape -- including adding a `#[serde(default)]` field -- MUST bump /// this constant, so `open()` rejects old-format databases up front instead /// of decoding old rows into errors that get silently dropped by the /// `list_*` methods below. -const SCHEMA_VERSION: u32 = 4; // v4: added ready-barrier release + verdict (#2998) +const SCHEMA_VERSION: u32 = 5; // v5: record encoding moved from bincode to postcard const SCHEMA_VERSION_KEY: &str = "schema_version"; /// Run `f` with umask set to `0o077` (owner-only) on Unix, restoring afterwards. @@ -140,32 +140,35 @@ impl RedbStore { // --------------------------------------------------------------------------- /// Maximum size of a single serialized record. `decode` refuses to *read* -/// beyond this (capping allocation from a malformed varint length prefix), and -/// `encode` refuses to *write* beyond it. The two limits must match: a record -/// that encodes larger than `decode` will accept is silently unreadable — -/// `put_dataflow` would persist it, but every later `get_dataflow` returns a -/// decode error and `list_dataflows` skips it as "corrupt" (dora-rs/dora#2027), -/// dropping a perfectly valid dataflow from startup recovery. Guarding at write -/// time turns that silent data loss into an explicit error at the source. +/// beyond this, and `encode` refuses to *write* beyond it. The two limits must +/// match: a record that encodes larger than `decode` will accept is silently +/// unreadable — `put_dataflow` would persist it, but every later +/// `get_dataflow` returns a decode error and `list_dataflows` skips it as +/// "corrupt" (dora-rs/dora#2027), dropping a perfectly valid dataflow from +/// startup recovery. Guarding at write time turns that silent data loss into an +/// explicit error at the source. const MAX_RECORD_BYTES: usize = 64 * 1024 * 1024; -fn encode(value: &T) -> Result> { - let bytes = bincode::serde::encode_to_vec(value, bincode::config::standard()) - .map_err(|e| eyre!("encode error: {e}"))?; - if bytes.len() > MAX_RECORD_BYTES { +/// The one place the limit is applied, so `encode` and `decode` cannot drift. +fn check_record_size(len: usize, direction: &str) -> Result<()> { + if len > MAX_RECORD_BYTES { eyre::bail!( - "record too large to store: {} bytes exceeds the {MAX_RECORD_BYTES}-byte limit", - bytes.len() + "record too large to {direction}: {len} bytes exceeds the {MAX_RECORD_BYTES}-byte limit" ); } + Ok(()) +} + +fn encode(value: &T) -> Result> { + let bytes = dora_message::encode(value).map_err(|e| eyre!("encode error: {e}"))?; + check_record_size(bytes.len(), "store")?; Ok(bytes) } fn decode(bytes: &[u8]) -> Result { - let config = bincode::config::standard().with_limit::<{ MAX_RECORD_BYTES }>(); - let (val, _) = - bincode::serde::decode_from_slice(bytes, config).map_err(|e| eyre!("decode error: {e}"))?; - Ok(val) + // The codec has no built-in decode cap, so mirror `encode`'s limit here. + check_record_size(bytes.len(), "read").map_err(|e| eyre!("decode error: {e}"))?; + dora_message::decode(bytes).map_err(|e| eyre!("decode error: {e:#}")) } // --------------------------------------------------------------------------- @@ -609,18 +612,18 @@ mod tests { } /// #2471: `#[serde(default)]` on `DataflowStatus::Failed::terminal` does NOT - /// make bincode-encoded bytes from before the field existed decodable -- - /// bincode is positional, not self-describing. `status` is not the last + /// make encoded bytes from before the field existed decodable -- + /// postcard is positional, not self-describing. `status` is not the last /// field of `DataflowRecord`, so a missing `terminal` byte doesn't reliably /// fail to decode the way a standalone `DataflowStatus` would: the decoder /// instead reads the next record field's first byte as `terminal` and keeps /// going, shifted by one byte, into `daemon_ids`/`node_to_daemon`/`uv`/ /// `generation`/timestamps. Depending on the field values that land on /// each shifted position, this either surfaces much later as a confusing, - /// unrelated decode error (e.g. `InvalidBooleanValue` on `uv`, as observed - /// with the field values below -- nothing about that error points at - /// `terminal` or `status`) or -- for other field-value combinations -- - /// could silently succeed with a structurally valid but corrupted record. + /// unrelated decode error (a bad-bool or bad-length failure on one of the + /// shifted trailing fields -- nothing about it points at `terminal` or + /// `status`) or -- for other field-value combinations -- could silently + /// succeed with a structurally valid but corrupted record. /// Neither outcome is the clean, predictable "missing field" failure the /// `terminal` doc comment implies, and both confirm the actual safety net /// is the `SCHEMA_VERSION` bump: a database written before `terminal` @@ -674,12 +677,11 @@ mod tests { created_at: 1000, updated_at: 2000, }; - let old_bytes = - bincode::serde::encode_to_vec(&old_record, bincode::config::standard()).unwrap(); + let old_bytes = dora_message::encode(&old_record).unwrap(); // Decoding as the current `DataflowRecord` must NOT reproduce the - // encoded values: it either errors out (empirically: `InvalidBooleanValue` - // on the shifted `uv` field, not anything mentioning `terminal`), or -- + // encoded values: it either errors out on a shifted trailing field + // (with nothing in the message mentioning `terminal`), or -- // for other field-value combinations -- silently misdecodes into a // structurally valid but wrong record. Either outcome is unsafe to // serve from `list_dataflows`/`get_dataflow`, which is exactly why old @@ -694,7 +696,7 @@ mod tests { || decoded.updated_at != 2000, "decode succeeded AND reproduced the original trailing fields -- \ the byte-shift this test exists to demonstrate did not occur; \ - re-check bincode's encoding of an empty Vec/BTreeMap length prefix" + re-check postcard's encoding of an empty Vec/BTreeMap length prefix" ); } } @@ -832,7 +834,7 @@ mod tests { /// redb 3 dropped support for the v2 on-disk format, so a store written by /// an older dora (redb 2.x) is unreadable after the redb 4 upgrade (#2449). - /// That failure must carry [`crate::SCHEMA_MISMATCH_MARKER`] like a bincode + /// That failure must carry [`crate::SCHEMA_MISMATCH_MARKER`] like a record /// schema bump does, so `dora up` offers the same `--recreate-store` /// recovery instead of surfacing a bare redb error the user cannot act on. #[test] @@ -1323,7 +1325,7 @@ mod tests { } } - /// A corrupt bincode blob in the dataflows table surfaces as an error + /// A corrupt record blob in the dataflows table surfaces as an error /// on the read path, not a panic. This is the file-corruption scenario /// from plan-fault-injection.md section 3.3 at the API level. #[test] @@ -1338,7 +1340,7 @@ mod tests { let txn = store.db.begin_write().unwrap(); { let mut table = txn.open_table(DATAFLOWS).unwrap(); - let garbage: &[u8] = b"not a valid bincode record XXXXXXXX"; + let garbage: &[u8] = b"not a valid postcard record XXXXXXXX"; let key: &[u8] = uuid.as_bytes(); table.insert(key, garbage).unwrap(); } diff --git a/libraries/core/src/topics.rs b/libraries/core/src/topics.rs index 30be74902f..81c4cc92da 100644 --- a/libraries/core/src/topics.rs +++ b/libraries/core/src/topics.rs @@ -655,7 +655,7 @@ fn local_address_toward(target: SocketAddr) -> Option { /// Zenoh key for node output data. /// -/// Payload format: raw Arrow bytes with bincode `Metadata` in the Zenoh +/// Payload format: raw Arrow bytes with postcard `Metadata` in the Zenoh /// attachment. This topic is published by nodes and consumed directly by /// downstream nodes (plus debug-inspection subscribers). Daemon control frames /// must not be published here; use [`zenoh_daemon_control_topic`] instead. @@ -741,7 +741,7 @@ pub fn zenoh_output_ack_topic( /// Zenoh key for control frames associated with a node output. /// -/// Payload format: bincode `Timestamped` with no Zenoh +/// Payload format: postcard `Timestamped` with no Zenoh /// attachment. Published by daemons for inter-daemon control (for example /// `OutputClosed`) and by the coordinator for explicit topic injection. Keeping /// this separate from [`zenoh_output_publish_topic`] avoids mixing control frames @@ -931,7 +931,7 @@ mod tests { // Node raw output and daemon control frames MUST live on distinct Zenoh // keys: they share neither format nor consumer, and merging them caused the - // #1992 crossover (daemon bincode-decoding node output). Guard the split. + // #1992 crossover (daemon postcard-decoding node output). Guard the split. #[cfg(feature = "zenoh")] #[test] fn output_and_control_topics_are_distinct() { diff --git a/libraries/message/Cargo.toml b/libraries/message/Cargo.toml index 3a5f5a7f07..f02853b52e 100644 --- a/libraries/message/Cargo.toml +++ b/libraries/message/Cargo.toml @@ -42,7 +42,7 @@ schemars = "1.2.1" uhlc = "0.9.0" serde_yaml = { workspace = true } serde-with-expand-env = "1.1.0" -bincode = { workspace = true } +postcard = { workspace = true } serde_json = { workspace = true } chrono = { version = "0.4", features = ["serde"] } getrandom = "0.4" diff --git a/libraries/message/benches/message_serde.rs b/libraries/message/benches/message_serde.rs index 5071d5445c..0d17fbcf72 100644 --- a/libraries/message/benches/message_serde.rs +++ b/libraries/message/benches/message_serde.rs @@ -19,28 +19,34 @@ fn make_send_message(payload_size: usize) -> DaemonRequest { } } -fn bench_bincode_serialize(c: &mut Criterion) { - let mut group = c.benchmark_group("bincode_serialize"); +fn bench_encode(c: &mut Criterion) { + let mut group = c.benchmark_group("encode"); for &size in &[64, 4096, 65536, 1_048_576] { let msg = make_send_message(size); - group.bench_with_input(BenchmarkId::new("SendMessage", size), &msg, |b, msg| { + // The production path (`daemon_connection::tcp`): pre-sized buffer. + group.bench_with_input(BenchmarkId::new("presized", size), &msg, |b, msg| { b.iter(|| { - let bytes = bincode::serialize(black_box(msg)).unwrap(); - black_box(bytes); + let msg = black_box(msg); + black_box(dora_message::encode_presized(msg, msg.encode_size_hint()).unwrap()); }); }); + // The same encoding into a buffer that grows from empty. Keeps the + // pre-sizing win measurable in-tree: if this arm stops being slower, + // `encode_presized` has stopped earning its complexity. + group.bench_with_input(BenchmarkId::new("grow_from_empty", size), &msg, |b, msg| { + b.iter(|| black_box(postcard::to_stdvec(black_box(msg)).unwrap())); + }); } group.finish(); } -fn bench_bincode_deserialize(c: &mut Criterion) { - let mut group = c.benchmark_group("bincode_deserialize"); +fn bench_decode(c: &mut Criterion) { + let mut group = c.benchmark_group("decode"); for &size in &[64, 4096, 65536, 1_048_576] { - let msg = make_send_message(size); - let bytes = bincode::serialize(&msg).unwrap(); + let bytes = dora_message::encode(&make_send_message(size)).unwrap(); group.bench_with_input(BenchmarkId::new("SendMessage", size), &bytes, |b, bytes| { b.iter(|| { - let msg: DaemonRequest = bincode::deserialize(black_box(bytes)).unwrap(); + let msg: DaemonRequest = dora_message::decode(black_box(bytes)).unwrap(); black_box(msg); }); }); @@ -62,10 +68,5 @@ fn bench_metadata_clone(c: &mut Criterion) { group.finish(); } -criterion_group!( - benches, - bench_bincode_serialize, - bench_bincode_deserialize, - bench_metadata_clone, -); +criterion_group!(benches, bench_encode, bench_decode, bench_metadata_clone,); criterion_main!(benches); diff --git a/libraries/message/src/common.rs b/libraries/message/src/common.rs index e019c2a4d0..8c64167e92 100644 --- a/libraries/message/src/common.rs +++ b/libraries/message/src/common.rs @@ -241,18 +241,15 @@ pub struct Timestamped { pub timestamp: uhlc::Timestamp, } -impl Timestamped -where - T: serde::Serialize, -{ +impl Timestamped { + /// Encode this event for the zenoh data plane and `dora record` files. pub fn serialize(&self) -> eyre::Result> { - bincode::serialize(self).wrap_err("failed to serialize timestamped message") + crate::encode_presized(self, self.inner.encode_size_hint()) + .wrap_err("failed to serialize timestamped message") } -} -impl Timestamped { pub fn deserialize_inter_daemon_event(bytes: &[u8]) -> eyre::Result { - bincode::deserialize(bytes).wrap_err("failed to deserialize InterDaemonEvent") + crate::decode(bytes).wrap_err("failed to deserialize InterDaemonEvent") } } @@ -263,12 +260,25 @@ pub enum DataMessage { Vec(AVec>), } +impl DataMessage { + /// Byte length of the carried payload. + pub fn len(&self) -> usize { + match self { + Self::Vec(v) => v.len(), + } + } + + pub fn is_empty(&self) -> bool { + self.len() == 0 + } +} + impl fmt::Debug for DataMessage { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { - Self::Vec(v) => f + Self::Vec(_) => f .debug_struct("Vec") - .field("len", &v.len()) + .field("len", &self.len()) .finish_non_exhaustive(), } } diff --git a/libraries/message/src/daemon_to_daemon.rs b/libraries/message/src/daemon_to_daemon.rs index cb4b70e640..fe402aa042 100644 --- a/libraries/message/src/daemon_to_daemon.rs +++ b/libraries/message/src/daemon_to_daemon.rs @@ -22,3 +22,19 @@ pub enum InterDaemonEvent { output_id: DataId, }, } + +impl InterDaemonEvent { + /// Bulk bytes this event will contribute to its encoding, for + /// [`crate::encode_presized`]. + /// + /// Not to be confused with [`crate::metadata::debug_frame_wire_size`], which + /// answers "how big was this on the wire" and deliberately prefers the + /// daemon-stamped `WIRE_SIZE` parameter over the buffer length (#2584). This + /// one must be the actual buffer length, since it sizes an allocation. + pub fn encode_size_hint(&self) -> usize { + match self { + Self::Output { data, .. } => data.as_ref().map_or(0, |d| d.len()), + Self::OutputClosed { .. } => 0, + } + } +} diff --git a/libraries/message/src/daemon_to_node.rs b/libraries/message/src/daemon_to_node.rs index e3641dcb8f..d7547ee365 100644 --- a/libraries/message/src/daemon_to_node.rs +++ b/libraries/message/src/daemon_to_node.rs @@ -96,6 +96,28 @@ pub enum DaemonReply { Empty, } +impl DaemonReply { + /// Bulk bytes this reply will contribute to its encoding, for + /// [`crate::encode_presized`]. + /// + /// `NextEvents` is a batch, so this sums the per-event hints rather than + /// relying on the single flat envelope allowance `encode_presized` adds: + /// the daemon drains up to `NODE_EVENT_CHANNEL_CAPACITY` events into one + /// reply, and a batch of a few dozen would otherwise realloc several times + /// on envelopes alone. + pub fn encode_size_hint(&self) -> usize { + match self { + DaemonReply::NextEvents(events) => { + events.iter().map(|e| e.inner.encode_size_hint()).sum() + } + DaemonReply::Result(_) + | DaemonReply::NodeConfig { .. } + | DaemonReply::PinnedMemoryMetadata { .. } + | DaemonReply::Empty => 0, + } + } +} + #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] #[non_exhaustive] #[allow(clippy::large_enum_variant)] @@ -132,9 +154,9 @@ pub enum NodeEvent { /// Sent when `dora param set` changes a parameter for this node. /// /// `value_json` carries JSON-encoded bytes rather than `serde_json::Value`: - /// this message is serialized with bincode on the daemon↔node TCP channel, + /// this message is serialized with postcard on the daemon↔node TCP channel, /// and `serde_json::Value::deserialize` uses `deserialize_any`, which - /// bincode does not support. + /// postcard (like any non-self-describing format) does not support. ParamUpdate { key: String, value_json: Vec, @@ -164,3 +186,33 @@ pub enum NodeEvent { shared_memory_id: String, }, } + +impl NodeEvent { + /// Bulk bytes this event will contribute to the encoding of the + /// [`DaemonReply`] that wraps it. + /// + /// Includes a flat per-event allowance for the `Timestamped` wrapper, + /// `Metadata` and ids, because these are batched: see + /// [`DaemonReply::encode_size_hint`]. + pub fn encode_size_hint(&self) -> usize { + /// Measured at ~72 bytes for a typical `Input` (timestamp 25 + + /// `Metadata` 27 + tags and ids); rounded up so a batch of small events + /// still lands in one allocation. + const PER_EVENT_ENVELOPE: usize = 128; + + let payload = match self { + NodeEvent::Input { data, .. } => data.as_ref().map_or(0, |d| d.len()), + NodeEvent::Stop + | NodeEvent::Reload { .. } + | NodeEvent::InputClosed { .. } + | NodeEvent::InputRecovered { .. } + | NodeEvent::NodeRestarted { .. } + | NodeEvent::AllInputsClosed + | NodeEvent::ParamUpdate { .. } + | NodeEvent::ParamDeleted { .. } + | NodeEvent::NodeFailed { .. } + | NodeEvent::FreeMemoryPool { .. } => 0, + }; + payload.saturating_add(PER_EVENT_ENVELOPE) + } +} diff --git a/libraries/message/src/lib.rs b/libraries/message/src/lib.rs index 4ec15e531c..dfdac9db94 100644 --- a/libraries/message/src/lib.rs +++ b/libraries/message/src/lib.rs @@ -1,4 +1,10 @@ -//! Enable serialisation and deserialisation of capnproto messages +//! Message types and wire protocol shared by all dora components. +//! +//! Binary encoding is [postcard](https://docs.rs/postcard): a compact, +//! non-self-describing serde format with a stable, documented wire spec. The +//! encoding is positional, so field order and enum variant order are part of +//! the protocol — see [`metadata::Metadata::CURRENT_VERSION`] for how an +//! incompatible peer is detected. //! /// The version of the dora-message crate @@ -6,6 +12,65 @@ pub const VERSION: &str = env!("CARGO_PKG_VERSION"); pub use uhlc; +/// Per-message slack when pre-sizing an encode buffer, covering the envelope +/// around the bulk payload: enum tags, ids, [`metadata::Metadata`] and varint +/// length prefixes. Overshooting is free (one slightly larger allocation); +/// undershooting costs a realloc plus a memcpy of the payload. +const ENVELOPE_SIZE_HINT: usize = 512; + +/// Encode `value` in dora's binary wire format. +/// +/// This and [`encode_presized`] are the only encode entry points; call sites do +/// not name the underlying codec, so it stays swappable from one file. +pub fn encode(value: &T) -> postcard::Result> { + encode_presized(value, 0) +} + +/// [`encode`], for a message carrying `bulk_bytes` bytes of bulk payload. +/// +/// The encoder writes into a `Vec` that would otherwise start empty and +/// reallocate as it grows — 10–80% of the encode cost on dora's messages, worst +/// on the small control messages that dominate the daemon↔node TCP path. Pass +/// `encode_size_hint()`, which the message types that carry a payload provide; +/// the envelope slack is added here so that policy lives in one place. +pub fn encode_presized( + value: &T, + bulk_bytes: usize, +) -> postcard::Result> { + postcard::to_extend( + value, + Vec::with_capacity(bulk_bytes.saturating_add(ENVELOPE_SIZE_HINT)), + ) +} + +/// Decode `bytes` in dora's binary wire format, requiring the value to consume +/// the **entire** slice. +/// +/// This is the only decode entry point, and the trailing-byte check is why it +/// must stay that way: postcard's own `from_bytes` ignores trailing bytes where +/// bincode (this protocol's previous codec) rejected them, and two call sites +/// depend on the strict behaviour. +/// +/// - The zenoh attachment filters read a decode failure as "not a dora message, +/// ignore it". A foreign publisher whose attachment merely *starts* with +/// something shaped like a [`metadata::Metadata`] would otherwise be accepted +/// as genuine. +/// - The length-prefixed daemon↔node frames read a decode failure as a desynced +/// or incompatible peer. A frame longer than the value it carries is evidence +/// of exactly that, and must not be silently truncated. +pub fn decode<'a, T: serde::Deserialize<'a>>(bytes: &'a [u8]) -> eyre::Result { + let (value, rest) = postcard::take_from_bytes(bytes)?; + if !rest.is_empty() { + eyre::bail!( + "trailing bytes after decoded value ({} of {} unconsumed) — \ + likely a desynced or incompatible peer", + rest.len(), + bytes.len() + ); + } + Ok(value) +} + /// Maximum allowed message size over TCP (64 MiB). /// /// Large payloads should use the shared-memory transport instead, @@ -131,3 +196,23 @@ pub(crate) fn versions_compatible( let matches = req.matches(specified_version) || specified_dora_req.matches(crate_version); Ok(matches) } + +#[cfg(test)] +mod encoding_tests { + use crate::metadata::Metadata; + + /// The property the zenoh attachment filters rely on: a buffer that merely + /// *starts* with a valid value is not a valid message. + #[test] + fn decode_rejects_trailing_bytes() { + let mut bytes = + crate::encode(&Metadata::new(uhlc::HLC::default().new_timestamp())).expect("serialize"); + bytes.extend_from_slice(b"foreign publisher trailer"); + + let err = crate::decode::(&bytes).expect_err("trailing bytes must be rejected"); + assert!( + format!("{err:#}").contains("trailing bytes"), + "error should name the cause, got: {err:#}" + ); + } +} diff --git a/libraries/message/src/metadata.rs b/libraries/message/src/metadata.rs index 65fa583df0..ddd9f3e02c 100644 --- a/libraries/message/src/metadata.rs +++ b/libraries/message/src/metadata.rs @@ -19,11 +19,14 @@ impl Metadata { /// Current metadata wire-format version, stamped on every outgoing message. /// /// Bumped from 0 to 1 when the `ArrowTypeInfo` sidecar was dropped and the - /// wire format became Arrow-IPC-only. A receiver can compare + /// wire format became Arrow-IPC-only, and from 1 to 2 when the binary + /// encoding moved from bincode to postcard (varint integers and length + /// prefixes, so the byte layout differs even though the field order does + /// not). A receiver can compare /// [`metadata_version`](Self::metadata_version) against this to detect a peer /// speaking an incompatible format and report it clearly instead of failing /// with a cryptic positional-deserialization error. - pub const CURRENT_VERSION: u16 = 1; + pub const CURRENT_VERSION: u16 = 2; pub fn new(timestamp: uhlc::Timestamp) -> Self { Self::from_parameters(timestamp, Default::default()) @@ -347,14 +350,14 @@ mod tests { #[test] fn startup_marker_is_detected_and_survives_the_wire() { // A marker is recognized only via the reserved parameter, and the flag - // must survive bincode round-tripping — it travels as the zenoh + // must survive postcard round-tripping — it travels as the zenoh // attachment, and a receiver that failed to recognize it would decode // the marker as node data and surface it to user code. let marker = Metadata::startup_marker(test_timestamp()); assert!(marker.is_startup_marker()); - let bytes = bincode::serialize(&marker).expect("serialize"); - let decoded: Metadata = bincode::deserialize(&bytes).expect("deserialize"); + let bytes = crate::encode(&marker).expect("serialize"); + let decoded: Metadata = crate::decode(&bytes).expect("deserialize"); assert!(decoded.is_startup_marker()); } @@ -404,7 +407,7 @@ mod tests { #[test] fn startup_ack_round_trips_and_extracts_identity() { - // The ack travels as a bincode attachment on the `@ack` topic; the + // The ack travels as a postcard attachment on the `@ack` topic; the // producer must recover exactly the (consumer, input) identity it needs // to tick off a required acker. let ack = Metadata::startup_ack(test_timestamp(), "camera-consumer", "image/depth"); @@ -416,8 +419,8 @@ mod tests { // travel on different topics but share the filtering code path. assert!(!ack.is_startup_marker()); - let bytes = bincode::serialize(&ack).expect("serialize"); - let decoded: Metadata = bincode::deserialize(&bytes).expect("deserialize"); + let bytes = crate::encode(&ack).expect("serialize"); + let decoded: Metadata = crate::decode(&bytes).expect("deserialize"); assert_eq!( decoded.startup_ack_identity(), Some(("camera-consumer", "image/depth")) @@ -540,10 +543,10 @@ mod tests { #[test] fn outgoing_metadata_is_stamped_with_current_version() { - // The wire format is positional (bincode) and carries no separate type + // The wire format is positional (postcard) and carries no separate type // descriptor, so `metadata_version` is the only in-band signal of an // incompatible layout. Every constructor must stamp `CURRENT_VERSION`. - assert_eq!(Metadata::CURRENT_VERSION, 1); + assert_eq!(Metadata::CURRENT_VERSION, 2); let ts = uhlc::HLC::default().new_timestamp(); assert_eq!( Metadata::new(ts).metadata_version(), diff --git a/libraries/message/src/node_to_daemon.rs b/libraries/message/src/node_to_daemon.rs index 96004e15a2..958b757499 100644 --- a/libraries/message/src/node_to_daemon.rs +++ b/libraries/message/src/node_to_daemon.rs @@ -42,7 +42,31 @@ pub enum DaemonRequest { } impl DaemonRequest { - pub fn expects_tcp_bincode_reply(&self) -> bool { + /// Bulk bytes this request will contribute to its encoding, for + /// [`crate::encode_presized`]. + /// + /// Matched exhaustively on purpose: a new payload-carrying variant that + /// forgets to report its size would silently fall back to growing the + /// buffer from empty, which is the cost `encode_presized` exists to avoid + /// and which no test would catch. + pub fn encode_size_hint(&self) -> usize { + match self { + DaemonRequest::SendMessage { data, .. } => data.as_ref().map_or(0, DataMessage::len), + DaemonRequest::Register(_) + | DaemonRequest::Subscribe + | DaemonRequest::OutputSent { .. } + | DaemonRequest::CloseOutputs(_) + | DaemonRequest::OutputsDone + | DaemonRequest::NextEvent + | DaemonRequest::EventStreamDropped + | DaemonRequest::NodeConfig { .. } + | DaemonRequest::RegisterPinnedMemory { .. } + | DaemonRequest::ReadPinnedMemory { .. } + | DaemonRequest::FreePinnedMemory { .. } => 0, + } + } + + pub fn expects_tcp_binary_reply(&self) -> bool { #[allow(clippy::match_like_matches_macro)] match self { DaemonRequest::SendMessage { .. } @@ -93,7 +117,13 @@ pub struct NodeRegisterRequest { /// layout differs (as it did across #2366). Carrying the format version /// explicitly lets the daemon reject an incompatible peer at register with /// a clear message, instead of the node desyncing mid-stream into a - /// cryptic `tag for enum is not valid` bincode error (#2742). + /// cryptic bad-enum-discriminant deserialization error (#2742). + /// + /// This covers *layout* drift within one encoding. It cannot cover a change + /// of the encoding itself: the register frame is encoded the same way as + /// everything else, so a peer from before the bincode→postcard move fails + /// while decoding the frame that carries this field, never reaching the + /// check. Such changes are gated by the release notes instead. metadata_version: u16, } @@ -168,7 +198,7 @@ mod register_version_tests { // A peer built from a dora revision with a different metadata wire // format — same semver, incompatible bytes. This is the #2366 / #2742 // shape: it must be rejected at register with a legible message rather - // than desyncing mid-stream into a cryptic bincode error. + // than desyncing mid-stream into a cryptic deserialization error. let mut req = request(); req.metadata_version = Metadata::CURRENT_VERSION.wrapping_add(1); let err = req diff --git a/libraries/message/tests/uhlc_wire_format.rs b/libraries/message/tests/uhlc_wire_format.rs index 5adf0491b2..f03ceb4574 100644 --- a/libraries/message/tests/uhlc_wire_format.rs +++ b/libraries/message/tests/uhlc_wire_format.rs @@ -1,6 +1,6 @@ //! Wire-format contract between `dora-message` and its `uhlc` dependency. //! -//! `uhlc::Timestamp` is embedded in every bincode-encoded daemon↔node, +//! `uhlc::Timestamp` is embedded in every postcard-encoded daemon↔node, //! daemon↔daemon and record-file message (`Metadata`, `Timestamped`), and in //! every JSON-encoded CLI↔coordinator and coordinator↔daemon WebSocket frame. //! Its encoding is therefore part of dora's on-the-wire protocol even though the @@ -9,13 +9,18 @@ //! loudly. //! //! The golden vectors below pin that encoding so a future `uhlc` bump cannot -//! move it unnoticed. They were captured under `uhlc 0.5.2` and still hold -//! under `uhlc 0.9.0` (dora-rs/dora#2446): 0.9 changed the in-memory -//! representation of `ID` from `NonZeroU128` to `[u8; 16]`, but both encode to -//! the same 16 little-endian bytes under bincode, so the binary plane is -//! untouched. The JSON plane *did* change shape (see -//! `timestamp_json_encoding_is_pinned`) — that link is version-gated and now -//! survives `serde_json::Value`, which the `NonZeroU128` form could not. +//! move it unnoticed. The binary vectors were re-captured when the binary plane +//! moved from bincode to postcard (`Metadata::CURRENT_VERSION` 1 → 2): postcard +//! varint-encodes the `NTP64` instead of writing a fixed little-endian `u64`, +//! while the HLC id stays a fixed 16-byte array. That was the last deliberate +//! change to these bytes — a test failing against them now means a protocol +//! migration is genuinely required, not that the vector needs updating. +//! +//! The JSON plane is unaffected by the postcard move. It last changed shape at +//! `uhlc 0.9.0` (dora-rs/dora#2446), which switched `ID`'s in-memory form from +//! `NonZeroU128` to `[u8; 16]` (see `timestamp_json_encoding_is_pinned`) — that +//! link is version-gated and now survives `serde_json::Value`, which the +//! `NonZeroU128` form could not. use dora_message::{ common::Timestamped, @@ -23,15 +28,15 @@ use dora_message::{ uhlc::{ID, NTP64, Timestamp}, }; -/// The pinned bincode encoding of [`golden_timestamp`]: the NTP64 as a -/// little-endian `u64`, then the HLC id as its 16 little-endian bytes. +/// The pinned postcard encoding of [`golden_timestamp`]: the NTP64 as a +/// 9-byte varint, then the HLC id as its 16 little-endian bytes (a fixed-size +/// array, so postcard writes no length prefix for it). /// -/// This exact vector is what a uhlc-0.5 dora produced and what a uhlc-0.9 dora -/// produces. If a bump changes it, every daemon↔node message, every inter-daemon +/// If a uhlc bump changes this, every daemon↔node message, every inter-daemon /// zenoh sample and every `dora record` file written by a different dora version /// becomes unreadable — so a test failing against it means a protocol migration /// is genuinely required, not that the vector needs updating. -const TIMESTAMP_HEX: &str = "88776655443322110102030405060708090a0b0c0d0e0f10"; +const TIMESTAMP_HEX: &str = "88ef99abc5e88c91110102030405060708090a0b0c0d0e0f10"; /// A fixed timestamp: an arbitrary but non-round NTP64 and a 16-byte HLC id /// whose bytes are all distinct, so any reordering (the failure mode #2446 @@ -52,20 +57,20 @@ fn hex(bytes: &[u8]) -> String { } #[test] -fn timestamp_bincode_encoding_is_pinned() { - let encoded = bincode::serialize(&golden_timestamp()).expect("serialize timestamp"); +fn timestamp_postcard_encoding_is_pinned() { + let encoded = postcard::to_stdvec(&golden_timestamp()).expect("serialize timestamp"); assert_eq!( hex(&encoded), TIMESTAMP_HEX, - "uhlc::Timestamp bincode encoding changed — this is a protocol break" + "uhlc::Timestamp postcard encoding changed — this is a protocol break" ); } #[test] -fn timestamp_bincode_round_trips() { +fn timestamp_postcard_round_trips() { let original = golden_timestamp(); - let encoded = bincode::serialize(&original).expect("serialize"); - let decoded: Timestamp = bincode::deserialize(&encoded).expect("deserialize"); + let encoded = postcard::to_stdvec(&original).expect("serialize"); + let decoded: Timestamp = postcard::from_bytes(&encoded).expect("deserialize"); assert_eq!(decoded, original); // Compare the id through its byte form too: `PartialEq` alone would still @@ -78,31 +83,31 @@ fn timestamp_bincode_round_trips() { } #[test] -fn metadata_bincode_encoding_is_pinned() { +fn metadata_postcard_encoding_is_pinned() { // `Metadata` is the envelope that actually rides on every output message, - // so pin the composite too: `metadata_version` as a little-endian u16, the - // 24-byte timestamp, then the parameter map length as a little-endian u64. - let encoded = bincode::serialize(&Metadata::new(golden_timestamp())).expect("serialize"); + // so pin the composite too: `metadata_version` as a varint u16, the 25-byte + // timestamp, then the parameter map length as a varint. + let encoded = postcard::to_stdvec(&Metadata::new(golden_timestamp())).expect("serialize"); assert_eq!( hex(&encoded), format!( "{version}{TIMESTAMP_HEX}{empty_parameter_map}", - version = "0100", - empty_parameter_map = "0000000000000000", + version = "02", + empty_parameter_map = "00", ), - "Metadata bincode encoding changed — bump Metadata::CURRENT_VERSION" + "Metadata postcard encoding changed — bump Metadata::CURRENT_VERSION" ); } #[test] -fn timestamped_bincode_prefix_is_the_inner_value() { +fn timestamped_postcard_prefix_is_the_inner_value() { // `Timestamped` puts `inner` first and the timestamp last, so the - // timestamp's 24 bytes are the tail of every daemon↔coordinator frame. + // timestamp's 25 bytes are the tail of every daemon↔coordinator frame. let timestamped = Timestamped { inner: 0xABu8, timestamp: golden_timestamp(), }; - let encoded = bincode::serialize(×tamped).expect("serialize"); + let encoded = postcard::to_stdvec(×tamped).expect("serialize"); assert_eq!(hex(&encoded), format!("ab{TIMESTAMP_HEX}")); } diff --git a/libraries/recording/src/lib.rs b/libraries/recording/src/lib.rs index 5f283912c6..983585f35d 100644 --- a/libraries/recording/src/lib.rs +++ b/libraries/recording/src/lib.rs @@ -7,12 +7,12 @@ //! files as well as in-memory buffers. //! //! ``` -//! use dora_recording::{RecordEntry, RecordingHeader, RecordingReader, RecordingWriter}; +//! use dora_recording::{FORMAT_VERSION, RecordEntry, RecordingHeader, RecordingReader, RecordingWriter}; //! use std::io::Cursor; //! //! # fn main() -> eyre::Result<()> { //! let header = RecordingHeader { -//! version: 1, +//! version: FORMAT_VERSION, //! start_nanos: 0, //! dataflow_id: uuid::Uuid::nil(), //! descriptor_yaml: b"nodes: []".to_vec(), @@ -50,7 +50,26 @@ use uuid::Uuid; const MAGIC: &[u8; 8] = b"DORAREC\x00"; const FOOTER_MAGIC: &[u8; 8] = b"DORAEND\x00"; -const FORMAT_VERSION: u16 = 1; +/// Version stamped into the header of newly written `.drec` files. +/// +/// Bumped from 1 to 2 when `event_bytes` moved from bincode to postcard: the +/// container framing is unchanged, but every entry's payload is a +/// `Timestamped` in the new encoding. +/// +/// Writers must stamp [`RecordingHeader::version`] with this rather than a +/// literal, or they produce files their own reader rejects. +pub const FORMAT_VERSION: u16 = 2; +/// Oldest `.drec` version this build can read. +/// +/// The entry payloads are opaque to the container, so a version that only the +/// *payload* encoding changed still has to be rejected here — otherwise the +/// header check passes and every entry fails to decode further downstream, +/// surfacing as "corrupt or format-drifted recording" instead of a clear +/// version error naming the dora release that wrote the file. +const MIN_SUPPORTED_FORMAT_VERSION: u16 = 2; +/// A bump that leaves the writers stamping a version below the reader's floor +/// would make every freshly written recording unreadable by the same build. +const _: () = assert!(FORMAT_VERSION >= MIN_SUPPORTED_FORMAT_VERSION); /// Maximum size for a single record or YAML descriptor in a `.drec` file. /// Guards against OOM from crafted files with `u32::MAX` length fields. const MAX_RECORD_BYTES: usize = 64 * 1024 * 1024; // 64 MB @@ -310,6 +329,13 @@ fn read_header(r: &mut R) -> eyre::Result { "unsupported recording format version {version} (max supported: {FORMAT_VERSION})" ); } + if version < MIN_SUPPORTED_FORMAT_VERSION { + eyre::bail!( + "recording format version {version} is no longer supported (min supported: \ + {MIN_SUPPORTED_FORMAT_VERSION}); it was written by a dora release that encoded \ + events with bincode. Re-record with this version of dora." + ); + } let mut nanos_buf = [0u8; 8]; r.read_exact(&mut nanos_buf)?; @@ -370,6 +396,31 @@ mod tests { assert_eq!(header, read_back); } + /// A v1 `.drec` holds bincode-encoded `event_bytes`, which this build + /// cannot decode. The container framing is identical, so nothing downstream + /// would notice until every entry failed to deserialize and got skipped as + /// "corrupt" — the reader must reject the file up front and say why. + #[test] + fn pre_postcard_recordings_are_rejected_with_a_version_error() { + let mut buf = Vec::new(); + write_header( + &mut buf, + &RecordingHeader { + version: 1, + ..sample_header() + }, + ) + .unwrap(); + + let err = read_header(&mut std::io::Cursor::new(&buf)) + .expect_err("a bincode-era recording must be rejected"); + let msg = format!("{err:#}"); + assert!( + msg.contains("no longer supported"), + "error must name the version problem, got: {msg}" + ); + } + #[test] fn single_record_roundtrip() { let header = sample_header();