From 382ca0f1465660c10fdf1e4ee3343a734258ddfb Mon Sep 17 00:00:00 2001 From: Amidamaru Date: Sun, 12 Jul 2026 15:10:56 +0700 Subject: [PATCH 1/2] feat(linkable): add allocation-free postcard `encode_into` Add a caller-provided buffer API to `Linkable` while retaining the existing from_bytes and to_bytes signatures for compatibility. Introduce `LinkCodecError` and an optional into-slice serializer path in core. Each outbound route owns one bounded reusable scratch buffer, validates the returned length, and falls back to the existing `Vec` serializer when the buffer is too small. Legacy and custom codecs keep their previous behavior. Generate a true postcard override using `postcard::to_slice` with a 256-byte preferred capacity. Extend codegen drift coverage for generated connector wiring, exact and undersized buffers, Cortex-M compilation and the JSON-free dependency graph. Add an allocation-counting benchmark covering 10000 `encode_into` calls with zero allocation calls and zero allocated bytes. This removes the codec allocation only, connector-internal payload copies remain outside the claim. --- CHANGELOG.md | 12 + Cargo.lock | 30 +- aimdb-bench/Cargo.toml | 6 + aimdb-bench/README.md | 13 +- aimdb-bench/benches/b0_alloc_linkable.rs | 86 +++++ aimdb-bench/src/lib.rs | 1 + aimdb-codegen/CHANGELOG.md | 7 + aimdb-codegen/src/rust.rs | 98 +++++- aimdb-core/CHANGELOG.md | 10 + aimdb-core/src/connector.rs | 87 ++++- aimdb-core/src/lib.rs | 7 +- aimdb-core/src/session/pump.rs | 29 +- aimdb-core/src/typed_api.rs | 303 +++++++++++++++++- aimdb-data-contracts/CHANGELOG.md | 8 + aimdb-data-contracts/src/lib.rs | 38 +++ aimdb-data-contracts/src/linkable.rs | 74 ++++- docs/design/024-M11-codegen-common-crate.md | 10 +- docs/design/041-data-contracts-integration.md | 45 +++ tools/codegen-drift/state-postcard.toml | 5 + tools/scripts/codegen-drift-check.sh | 58 +++- 20 files changed, 897 insertions(+), 30 deletions(-) create mode 100644 aimdb-bench/benches/b0_alloc_linkable.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 59b2bf38..79fd09e1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -105,6 +105,18 @@ multi-step migration round-trip suite and a trybuild compile-fail harness. ### Added +- **Zero-allocation generated Postcard encoding (Issue #177).** `Linkable` + gains a source-compatible `encode_into(&mut [u8])` fast path and generated + Postcard records implement it with `postcard::to_slice`. Core reuses one + bounded 256-byte scratch allocation per generated Postcard outbound route + (raw builders choose their capacity) and falls back to the existing owned + serializer when a value does not fit. The codec seam is gated + by a 10,000-iteration allocation-counting bench (0 allocations, 0 bytes) plus + host, `no_std` Cortex-M, and codegen-drift tests. The claim intentionally + excludes connector-internal payload copies and the existing boxed connector + reader future. ([aimdb-core](aimdb-core/CHANGELOG.md), + [aimdb-data-contracts](aimdb-data-contracts/CHANGELOG.md), + [aimdb-codegen](aimdb-codegen/CHANGELOG.md)) - **Embassy MQTT client TLS — `mqtts://` for embedded, WP7 ([design 044](docs/design/044-embassy-mqtt-tls.md)).** The Embassy MQTT connector can now dial a broker over TLS 1.3 (`embedded-tls`, pure-Rust `rustpki` certificate verification with `rsa`/`p384` so public CA chains verify out of the box), with SNI/hostname verification, DNS resolution, and a connector-internal SNTP time source that gates the first handshake so certificate validity is always checked against real time. Behind the new `embassy-tls` feature; `MqttConnectorBuilder::new` picks the transport from the URL scheme (`mqtt://` plain, `mqtts://` TLS) and gains `with_tls(TlsOptions)` (entropy, record buffers, SNTP server) and `with_credentials(username, password)` (MQTT CONNECT auth, both transports). The plain `mqtt://` path is untouched. The `embassy-mqtt-connector-demo` example gains a `tls` feature demonstrating the full flow against the repo's `dev/mosquitto` bench broker. ([aimdb-mqtt-connector](aimdb-mqtt-connector/CHANGELOG.md)) - **Design 034 Phase 3 — sans-io KNX/IP tunneling engine shared by both transports (Issue #135, [review doc §3.7](docs/design/034-technical-debt-review.md)).** The entire tunneling lifecycle — CONNECT_REQUEST/RESPONSE handshake, TUNNELING_REQUEST/ACK sequence + pending-ACK bookkeeping, keepalive (CONNECTIONSTATE_REQUEST) scheduling, ACK-timeout sweeps, and reconnect-with-backoff — now lives **once**, in the new runtime-neutral `aimdb_knx_connector::tunnel` module (`no_std + alloc`, no tokio/embassy imports), driven as a poll-based state machine (events in, `Action`s out, `next_deadline()` for timer arming). `tokio_client.rs` (988 → ~530 lines incl. a new fake-gateway integration test) and `embassy_client.rs` (1,055 → ~450 lines) are reduced to socket shims; the previously untestable handshake/ACK/keepalive/reconnect paths now have 15 host-run unit tests plus a scripted localhost-UDP roundtrip test. Behavioral unifications: Embassy gains the 5 s CONNECT_RESPONSE timeout (previously waited forever), both shims reconnect on fatal socket errors, and the dead tokio-only per-publish ACK oneshot is dropped (it was always `None`); the CONNECT_REQUEST HPAI stays per-transport (`LocalEndpoint`: tokio = real bound address, Embassy = NAT mode). ([aimdb-knx-connector](aimdb-knx-connector/CHANGELOG.md)) diff --git a/Cargo.lock b/Cargo.lock index 6c16ee62..edef104b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -72,6 +72,7 @@ dependencies = [ "embassy-time-driver", "futures", "portable-atomic", + "postcard", "serde", "serde_json", "tokio", @@ -303,7 +304,7 @@ dependencies = [ "aimdb-core", "aimdb-embassy-adapter", "aimdb-tokio-adapter", - "cobs", + "cobs 0.5.1", "defmt 1.0.1", "embassy-time-driver", "embedded-io-async 0.7.0", @@ -837,6 +838,15 @@ version = "0.7.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a1d728cc89cf3aee9ff92b05e62b19ee65a02b5702cff7d5a377e32c6ae29d8d" +[[package]] +name = "cobs" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0fa961b519f0b462e3a3b4a34b64d119eeaca1d59af726fe450bbba07a9fc0a1" +dependencies = [ + "thiserror 2.0.17", +] + [[package]] name = "cobs" version = "0.5.1" @@ -1651,6 +1661,12 @@ dependencies = [ "nb 1.1.0", ] +[[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" @@ -3112,6 +3128,18 @@ dependencies = [ "critical-section", ] +[[package]] +name = "postcard" +version = "1.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6764c3b5dd454e283a30e6dfe78e9b31096d9e32036b5d1eaac7a6119ccb9a24" +dependencies = [ + "cobs 0.3.0", + "embedded-io 0.4.0", + "embedded-io 0.6.1", + "serde", +] + [[package]] name = "potential_utf" version = "0.1.4" diff --git a/aimdb-bench/Cargo.toml b/aimdb-bench/Cargo.toml index b8d47661..19b08f27 100644 --- a/aimdb-bench/Cargo.toml +++ b/aimdb-bench/Cargo.toml @@ -40,6 +40,10 @@ harness = false name = "b0_alloc_migration" harness = false +[[bench]] +name = "b0_alloc_linkable" +harness = false + [features] default = ["std"] # Gates `profiles`/`reports`/`harness` and their criterion/serde_json/ @@ -89,6 +93,7 @@ serde_json = { workspace = true, optional = true } # criterion/serde_json above, since that's the only bench that needs it. aimdb-data-contracts = { path = "../aimdb-data-contracts", default-features = false, features = [ "alloc", + "linkable", "migratable", ], optional = true } @@ -124,3 +129,4 @@ critical-section = { version = "1.1", features = ["std"] } # defmt logger / panic-handler + embassy-time driver stubs for the host bench defmt = { workspace = true } embassy-time-driver = { path = "../_external/embassy/embassy-time-driver" } +postcard = { version = "1.0", default-features = false, features = ["alloc"] } diff --git a/aimdb-bench/README.md b/aimdb-bench/README.md index 327764f6..3e0bd994 100644 --- a/aimdb-bench/README.md +++ b/aimdb-bench/README.md @@ -20,6 +20,8 @@ Plus two informational benches that exercise the full runner-driven pipeline. **Adapters covered:** - **Tokio** — `b0_alloc_tokio`, `b1_b2_tokio` (host). +- **Postcard `Linkable` codec** — `b0_alloc_linkable` (host). This isolates the + issue #177 `encode_into` seam; it is not a whole-connector allocation claim. - **Embassy** — `b0_alloc_embassy`, `b1_b2_embassy` (host). These drive the real [`EmbassyBuffer`] backend via `futures::executor::block_on` over embassy-sync's poll methods — no @@ -53,6 +55,9 @@ Always run from the workspace root (`/aimdb_ws/aimdb`). # B0 — allocation gate (buffer layer) cargo bench -p aimdb-bench --bench b0_alloc_tokio +# B0 — allocation gate (Postcard Linkable::encode_into codec seam) +cargo bench -p aimdb-bench --bench b0_alloc_linkable + # B1 + B2 — latency (time/iter) and throughput (msgs/sec), one Criterion suite cargo bench -p aimdb-bench --bench b1_b2_tokio @@ -103,6 +108,13 @@ The committed baseline lives in `data/baselines/b0_alloc_tokio.json`. When a cha `b0_alloc_embassy` mirrors this against the Embassy buffer backend and writes `data/baselines/b0_alloc_embassy.json` — also **0 allocs/msg** across all three profiles, confirming the Embassy `poll_recv` path is allocation-free on the host. The on-target B3 bench (`examples/embassy-bench-stm32h5`) re-checks the same 0-alloc claim against the real embedded allocator. +`b0_alloc_linkable` warms up for 1,000 iterations, then measures 10,000 +generated-shape Postcard `Linkable::encode_into` calls into one stack buffer. +The required result is **0 allocation calls and 0 allocated bytes**. It isolates +the codec seam: `SerializedReader` still returns a boxed future, dynamic topics +may allocate, and connector implementations may copy payload ownership after +the core pump lends them the scratch slice. + > **Embassy eager registration (design 039 F8/F9).** An Embassy `SpmcRing` reader registers its embassy `Subscriber` eagerly, at `subscribe()` time — matching Tokio's `broadcast` — so no separate priming step is needed before the first `push`. **Noise reduction:** a `new_current_thread()` Tokio executor is used so there are no work-stealing threads and Tokio's scheduler does not allocate per-poll in the hot path. @@ -128,4 +140,3 @@ Use them as a comparison point, not a regression gate. If they regress, `b0_allo - Always specify `--release` or debug build consistently when comparing runs; optimizations differ by 5–50×. - `b_alloc_pipeline` uses a paced source: per-message pace tokens and notification channels. The coordination overhead is included in the measured window. - diff --git a/aimdb-bench/benches/b0_alloc_linkable.rs b/aimdb-bench/benches/b0_alloc_linkable.rs new file mode 100644 index 00000000..38dfdb8f --- /dev/null +++ b/aimdb-bench/benches/b0_alloc_linkable.rs @@ -0,0 +1,86 @@ +//! B0-Linkable — allocation gate for the Postcard `Linkable::encode_into` path. +//! +//! This deliberately measures the codec seam, not a complete connector. The +//! core pump still uses a boxed `SerializedReader` future and connector adapters +//! may copy payload ownership; issue #177 only claims that a generated-shape +//! Postcard codec writes into caller-owned storage with zero heap allocations. + +use std::hint::black_box; + +use aimdb_bench::alloc::{reset, snapshot}; +use aimdb_core::connector::LinkCodecError; +use aimdb_data_contracts::{Linkable, SchemaType}; +use serde::{Deserialize, Serialize}; + +const WARMUP_ITERS: usize = 1_000; +const MEASURE_ITERS: usize = 10_000; + +#[global_allocator] +static GLOBAL: aimdb_bench::alloc::CountingAllocator = + aimdb_bench::alloc::CountingAllocator(std::alloc::System); + +#[derive(Clone, Debug, Serialize, Deserialize)] +struct PostcardReading { + value: f32, + sequence: u32, +} + +impl SchemaType for PostcardReading { + const NAME: &'static str = "postcard_reading"; +} + +impl Linkable for PostcardReading { + const ENCODE_BUFFER_CAPACITY: Option = Some(256); + + fn from_bytes(data: &[u8]) -> Result { + postcard::from_bytes(data).map_err(|e| e.to_string()) + } + + fn to_bytes(&self) -> Result, String> { + postcard::to_allocvec(self).map_err(|e| e.to_string()) + } + + fn encode_into(&self, out: &mut [u8]) -> Result { + match postcard::to_slice(self, out) { + Ok(used) => Ok(used.len()), + Err(postcard::Error::SerializeBufferFull) => Err(LinkCodecError::BufferTooSmall), + Err(_) => Err(LinkCodecError::InvalidData), + } + } +} + +fn encode_batch(reading: &PostcardReading, out: &mut [u8; 256], iterations: usize) { + for _ in 0..iterations { + let written = black_box(reading) + .encode_into(black_box(out.as_mut_slice())) + .expect("Postcard encode_into should fit the fixed scratch buffer"); + black_box(&out[..written]); + } +} + +fn main() { + let reading = PostcardReading { + value: 23.75, + sequence: 42, + }; + let mut out = [0_u8; 256]; + + encode_batch(&reading, &mut out, WARMUP_ITERS); + reset(); + encode_batch(&reading, &mut out, MEASURE_ITERS); + let (allocations, bytes) = snapshot(); + + println!("=== B0 Linkable Postcard encode_into ==="); + println!(" Iterations : {MEASURE_ITERS}"); + println!(" Allocations : {allocations}"); + println!(" Bytes : {bytes}"); + + assert_eq!( + allocations, 0, + "encode_into allocated in the measured window" + ); + assert_eq!( + bytes, 0, + "encode_into allocated bytes in the measured window" + ); +} diff --git a/aimdb-bench/src/lib.rs b/aimdb-bench/src/lib.rs index 7c37dbae..46ccbd6e 100644 --- a/aimdb-bench/src/lib.rs +++ b/aimdb-bench/src/lib.rs @@ -21,6 +21,7 @@ //! | `benches/b0_alloc_tokio.rs` | B0 | Per-message allocation (Tokio buffer) | //! | `benches/b1_b2_tokio.rs` | B1+B2 | Latency (time/iter) + throughput (Tokio) | //! | `benches/b0_alloc_embassy.rs` | B0 | Per-message allocation (Embassy buffer) | +//! | `benches/b0_alloc_linkable.rs` | B0 | Postcard `Linkable::encode_into` allocs | //! | `benches/b1_b2_embassy.rs` | B1+B2 | Latency (time/iter) + throughput (Embassy)| //! | `benches/b_alloc_pipeline.rs` | info | Per-message allocation (runner pipeline) | //! | `benches/b_runner_pipeline.rs` | info | Runner pipeline throughput (Criterion) | diff --git a/aimdb-codegen/CHANGELOG.md b/aimdb-codegen/CHANGELOG.md index 26553dfc..6fc6253a 100644 --- a/aimdb-codegen/CHANGELOG.md +++ b/aimdb-codegen/CHANGELOG.md @@ -9,6 +9,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- **Issue #177 — generated Postcard records now have a true into-slice codec.** + Their `Linkable` implementation advertises a 256-byte reusable scratch + capacity and overrides `encode_into` with `postcard::to_slice`; generated + outbound connector chains install `with_serializer_into`. Values that exceed + the scratch capacity retain the existing `postcard::to_allocvec` fallback. + Host behavior, generated connector wiring, `no_std` Cortex-M compilation, and + the JSON-free dependency graph are covered by `make codegen-drift`. - Postcard codegen is now exercised end to end by `make codegen-drift`: a generated Postcard-only common crate roundtrips on the host, cross-compiles for `thumbv7em-none-eabihf`, and proves its normal dependency graph is JSON-free. A generated mixed JSON + Postcard crate is also compiled so codec feature/dependency drift fails in CI (#155). ### Fixed diff --git a/aimdb-codegen/src/rust.rs b/aimdb-codegen/src/rust.rs index ee9ea6bd..b3b7436d 100644 --- a/aimdb-codegen/src/rust.rs +++ b/aimdb-codegen/src/rust.rs @@ -15,6 +15,13 @@ use crate::state::{ TaskType, }; +/// Small per-route scratch buffer for generated Postcard link serializers. +/// +/// Generated records with a larger encoded value fall back to their existing +/// owned `to_allocvec` serializer for that value, so this is a memory/performance +/// trade-off rather than a wire-size limit. +const POSTCARD_ENCODE_BUFFER_CAPACITY: usize = 256; + // ── Public API ──────────────────────────────────────────────────────────────── /// Generate a complete Rust source file from architecture state. @@ -814,8 +821,17 @@ fn emit_record_configure_block(rec: &RecordDef) -> TokenStream { // everything must be a single fluent chain starting from `reg.buffer(...)`. // We build two branches: one with connectors wired (when all addresses // resolve), one plain buffer fallback. - let linked_chain = - emit_connector_chain(&rec.connectors, &value_type, &buffer_tokens, is_custom); + let serialization = rec + .serialization + .as_ref() + .unwrap_or(&SerializationType::Json); + let linked_chain = emit_connector_chain( + &rec.connectors, + &value_type, + &buffer_tokens, + is_custom, + serialization, + ); let addr_conditions: Vec = (0..rec.connectors.len()) .map(|i| { let addr_var = format_ident!("addr_{}", i); @@ -878,6 +894,7 @@ fn emit_connector_chain( value_type: &syn::Ident, buffer_tokens: &TokenStream, is_custom: bool, + serialization: &SerializationType, ) -> TokenStream { // Start the chain with reg.buffer(...) let mut chain = quote! { reg.buffer(#buffer_tokens) }; @@ -911,6 +928,17 @@ fn emit_connector_chain( }; } ConnectorDirection::Outbound => { + let into_slice = if matches!(serialization, SerializationType::Postcard) { + quote! { + .with_serializer_into( + <#value_type as Linkable>::ENCODE_BUFFER_CAPACITY + .unwrap_or(0), + |_ctx, v: &#value_type, out| v.encode_into(out), + ) + } + } else { + quote! {} + }; chain = quote! { #chain .link_to(#addr_var) @@ -918,6 +946,7 @@ fn emit_connector_chain( v.to_bytes() .map_err(|_| aimdb_core::connector::SerializeError::InvalidData) }) + #into_slice .finish() }; } @@ -995,13 +1024,29 @@ fn emit_linkable_json(rec: &RecordDef) -> TokenStream { fn emit_linkable_postcard(rec: &RecordDef) -> TokenStream { let struct_name = format_ident!("{}Value", rec.name); + let scratch_capacity = proc_macro2::Literal::usize_unsuffixed(POSTCARD_ENCODE_BUFFER_CAPACITY); quote! { impl Linkable for #struct_name { + const ENCODE_BUFFER_CAPACITY: Option = Some(#scratch_capacity); + fn to_bytes(&self) -> Result, alloc::string::String> { postcard::to_allocvec(self) .map_err(|e| alloc::format!("serialize {}: {e}", Self::NAME)) } + fn encode_into( + &self, + buf: &mut [u8], + ) -> Result { + match postcard::to_slice(self, buf) { + Ok(used) => Ok(used.len()), + Err(postcard::Error::SerializeBufferFull) => { + Err(aimdb_core::connector::LinkCodecError::BufferTooSmall) + } + Err(_) => Err(aimdb_core::connector::LinkCodecError::InvalidData), + } + } + fn from_bytes(data: &[u8]) -> Result { postcard::from_bytes(data) .map_err(|e| alloc::format!("deserialize {}: {e}", Self::NAME)) @@ -1501,12 +1546,23 @@ fn emit_transform_configure_block( .iter() .any(|c| matches!(c.direction, ConnectorDirection::Outbound)); let outbound_chain = if has_outbound { + let into_slice = if matches!(rec.serialization, Some(SerializationType::Postcard)) { + quote! { + .with_serializer_into( + <#value_type as Linkable>::ENCODE_BUFFER_CAPACITY.unwrap_or(0), + |_ctx, v: &#value_type, out| v.encode_into(out), + ) + } + } else { + quote! {} + }; quote! { .link_to(addr) .with_serializer(|_ctx, v: &#value_type| { v.to_bytes() .map_err(|_| aimdb_core::connector::SerializeError::InvalidData) }) + #into_slice .finish() } } else { @@ -2256,6 +2312,11 @@ serialization = "postcard" name = "value" type = "f32" description = "Reading value" + +[[records.connectors]] +protocol = "mqtt" +direction = "outbound" +url = "mqtt://readings/postcard/{variant}" "#; #[test] @@ -2296,6 +2357,7 @@ description = "Reading value" let mut state = ArchitectureState::from_toml(MIXED_CODEC_TOML).unwrap(); state.records.retain(|r| r.name == "PostcardReading"); let rust = generate_schema_rs(&state); + let configured = generate_rust(&state); assert!( rust.contains("impl Linkable for PostcardReadingValue"), "Missing Postcard Linkable impl:\n{rust}" @@ -2304,6 +2366,22 @@ description = "Reading value" rust.contains("postcard::to_allocvec(self)"), "Missing postcard serializer:\n{rust}" ); + assert!( + rust.contains("const ENCODE_BUFFER_CAPACITY: Option = Some(256)"), + "Missing Postcard scratch capacity:\n{rust}" + ); + assert!( + rust.contains("postcard::to_slice(self, buf)"), + "Missing zero-allocation Postcard serializer:\n{rust}" + ); + assert!( + rust.contains("LinkCodecError::BufferTooSmall"), + "Missing Postcard small-buffer mapping:\n{rust}" + ); + assert!( + configured.contains("with_serializer_into"), + "Missing Postcard connector fast-path wiring:\n{configured}" + ); assert!( rust.contains("postcard::from_bytes(data)"), "Missing postcard deserializer:\n{rust}" @@ -2355,12 +2433,18 @@ key_prefix = "sensors.temp." key_variants = ["indoor"] producers = ["sensor_task"] consumers = ["dashboard_task"] +serialization = "postcard" [[records.fields]] name = "celsius" type = "f64" description = "Temperature in degrees Celsius" +[[records.connectors]] +protocol = "mqtt" +direction = "outbound" +url = "mqtt://readings/postcard/{variant}" + [[tasks]] name = "sensor_task" task_type = "source" @@ -2380,6 +2464,11 @@ record = "TemperatureReading" [[binaries]] name = "drift-check-hub" tasks = ["sensor_task", "dashboard_task"] + +[[binaries.external_connectors]] +protocol = "mqtt" +env_var = "MQTT_BROKER" +default = "localhost" "#; fn hub_state() -> ArchitectureState { @@ -2397,6 +2486,11 @@ tasks = ["sensor_task", "dashboard_task"] out.contains(".source(sensor_task)"), "Producing task should still be wired via .source():\n{out}" ); + assert!( + out.contains(".with_serializer_into(") + && out.contains("Linkable>::ENCODE_BUFFER_CAPACITY"), + "Postcard-producing task should wire the into-slice serializer:\n{out}" + ); } #[test] diff --git a/aimdb-core/CHANGELOG.md b/aimdb-core/CHANGELOG.md index 9e503924..9ec98a74 100644 --- a/aimdb-core/CHANGELOG.md +++ b/aimdb-core/CHANGELOG.md @@ -9,6 +9,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- **Issue #177 — bounded into-slice serialization for outbound links.** New + `LinkCodecError` (with `SerializeError` retained as a compatibility alias), + `OutboundConnectorBuilder::with_serializer_into`, and the additive + `SerializedReader::recv_into`/`SerializedPayload` SPI let a fused typed reader + encode into caller-owned storage. `pump_sink` allocates one bounded scratch + buffer per route and reuses it across messages; an undersized buffer falls + back to the existing owned serializer for that value. Existing serializers, + third-party `SerializedReader` implementations, and `pump_client` keep their + owned-`Vec` behavior. This removes the codec allocation when an into-slice + encoder is installed; connector adapters may still copy the borrowed payload. - **`RecordRegistrar::signal_gauge(name, unit) -> SignalGaugeHandle`** (design 041 §3.2) — the core hook behind `aimdb-data-contracts`'s `Observable::observe()`. Feeding values via `SignalGaugeHandle::update(f64)` folds them into per-record `SignalStats` (last/min/max/mean), surfaced through `RecordMetadata::signal_stats` on `record.list`/`record.get`. Mirrors the `with_name` precedent: always callable, an inert no-op handle when the `observability` feature is off, so callers never `#[cfg]` on core's features. New public types: `SignalGaugeHandle` (always available), `SignalStats`/`SignalGauge`/`SignalStatsInfo` (feature `observability`). ### Fixed diff --git a/aimdb-core/src/connector.rs b/aimdb-core/src/connector.rs index abf8094a..7ca3c663 100644 --- a/aimdb-core/src/connector.rs +++ b/aimdb-core/src/connector.rs @@ -41,12 +41,17 @@ use alloc::format; use crate::{builder::AimDb, DbResult}; -/// Error that can occur during serialization +/// Allocation-free error shared by record/link codec operations. /// -/// Uses an enum instead of String for better performance in `no_std` environments -/// and to enable defmt logging support in Embassy. +/// This is deliberately separate from [`crate::CodecError`], which describes +/// failures in a session envelope codec (AimX, WebSocket, and similar framed +/// protocols). A link codec sits one layer higher: it turns a typed record into +/// the opaque payload passed to a connector. +/// +/// Uses an enum instead of `String` for predictable `no_std` behavior and +/// `defmt` logging support on Embassy targets. #[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum SerializeError { +pub enum LinkCodecError { /// Output buffer is too small for the serialized data BufferTooSmall, @@ -54,8 +59,14 @@ pub enum SerializeError { InvalidData, } +/// Backwards-compatible name for the outbound-only serializer error. +/// +/// New APIs should prefer [`LinkCodecError`], which also names the record/link +/// layer without implying that the error can only occur while encoding. +pub use LinkCodecError as SerializeError; + #[cfg(feature = "defmt")] -impl defmt::Format for SerializeError { +impl defmt::Format for LinkCodecError { fn format(&self, f: defmt::Formatter) { match self { Self::BufferTooSmall => defmt::write!(f, "BufferTooSmall"), @@ -65,7 +76,7 @@ impl defmt::Format for SerializeError { } #[cfg(feature = "std")] -impl std::fmt::Display for SerializeError { +impl std::fmt::Display for LinkCodecError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { Self::BufferTooSmall => write!(f, "Output buffer too small"), @@ -75,7 +86,7 @@ impl std::fmt::Display for SerializeError { } #[cfg(feature = "std")] -impl std::error::Error for SerializeError {} +impl std::error::Error for LinkCodecError {} /// One serialized record update, produced by a fused [`SerializedReader`] /// @@ -94,6 +105,35 @@ pub struct SerializedValue { pub payload: Vec, } +/// Location of a payload produced by [`SerializedReader::recv_into`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum SerializedPayload { + /// The initialized prefix of the caller-provided scratch buffer. + /// + /// A custom reader returning this variant must write the prefix during the + /// same `recv_into` call and its source must advertise a sufficient + /// [`SerializedSource::serializer_scratch_capacity`]. The pump validates + /// the length before publishing. + Scratch { + /// Number of initialized payload bytes in the scratch buffer. + len: usize, + }, + /// Compatibility fallback from the existing `Vec` serializer. + Owned(Vec), +} + +/// One serialized record produced into caller-owned scratch storage or an owned fallback. +/// +/// No reference escapes the async reader call. The pump validates `len`, then +/// borrows its own scratch buffer only for the subsequent `publish().await`. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct SerializedValueInto { + /// Dynamic destination resolved while the typed record was available. + pub dest: Option, + /// Scratch prefix metadata or an owned compatibility payload. + pub payload: SerializedPayload, +} + /// Type alias for the future returned by [`SerializedReader::recv`] /// /// Manual boxed future for object safety — same pattern as the rest of this @@ -101,6 +141,10 @@ pub struct SerializedValue { pub type RecvSerializedFuture<'a> = Pin> + Send + 'a>>; +/// Future returned by [`SerializedReader::recv_into`]. +pub type RecvSerializedIntoFuture<'a> = + Pin> + Send + 'a>>; + /// A subscription to one record, fused with destination resolution and /// serialization at registration time — no `dyn Any` crosses this boundary ///. @@ -113,6 +157,26 @@ pub trait SerializedReader: Send { /// recovered; any other error means the buffer is gone. Serialization /// failures are logged and skipped inside the reader. fn recv<'a>(&'a mut self, ctx: &'a crate::RuntimeContext) -> RecvSerializedFuture<'a>; + + /// Yield the next serialized value into caller-owned scratch storage. + /// + /// Third-party readers remain source-compatible: the default adapter calls + /// [`recv`](Self::recv) and returns its `Vec` as + /// [`SerializedPayload::Owned`]. AimDB's fused reader overrides this method + /// when an into-slice serializer was registered. + fn recv_into<'a>( + &'a mut self, + ctx: &'a crate::RuntimeContext, + _scratch: &'a mut [u8], + ) -> RecvSerializedIntoFuture<'a> { + Box::pin(async move { + let value = self.recv(ctx).await?; + Ok(SerializedValueInto { + dest: value.dest, + payload: SerializedPayload::Owned(value.payload), + }) + }) + } } /// A record's outbound wire interface, built where the record type `T` is @@ -120,6 +184,15 @@ pub trait SerializedReader: Send { /// bytes. Replaces the erased `ConsumerTrait` + serializer + topic-provider /// triple. pub trait SerializedSource: Send + Sync { + /// Scratch capacity requested by this source's into-slice serializer. + /// + /// `None` means the source only supports the existing owned serializer. A + /// source whose reader can return [`SerializedPayload::Scratch`] must + /// return `Some(capacity)` here so the route pump provides that storage. + fn serializer_scratch_capacity(&self) -> Option { + None + } + /// Subscribe to the record's updates. /// /// Synchronous and infallible — the buffer handle is pre-resolved at diff --git a/aimdb-core/src/lib.rs b/aimdb-core/src/lib.rs index 1074f011..fa700647 100644 --- a/aimdb-core/src/lib.rs +++ b/aimdb-core/src/lib.rs @@ -97,9 +97,12 @@ pub use profiling::{ // Connector Infrastructure exports pub use connector::TopicProvider; pub use connector::TopicResolverFn; -pub use connector::{ConnectorLink, ConnectorUrl, LinkAddress, SerializeError}; +pub use connector::{ConnectorLink, ConnectorUrl, LinkAddress, LinkCodecError, SerializeError}; pub use connector::{IngestFactoryFn, IngestFn}; -pub use connector::{SerializedReader, SerializedSource, SerializedValue, SourceFactoryFn}; +pub use connector::{ + SerializedPayload, SerializedReader, SerializedSource, SerializedValue, SerializedValueInto, + SourceFactoryFn, +}; // Router exports for connector implementations pub use router::{Route, Router, RouterBuilder}; diff --git a/aimdb-core/src/session/pump.rs b/aimdb-core/src/session/pump.rs index 0b01afb7..54c710f7 100644 --- a/aimdb-core/src/session/pump.rs +++ b/aimdb-core/src/session/pump.rs @@ -50,11 +50,15 @@ pub fn pump_sink(db: &AimDb, scheme: &str, sink: Arc) -> Vec) -> Vec m, // SPMC-ring overflow: messages were missed, but the reader // recovers (cursor resets to the oldest live value). Skip the @@ -82,11 +89,27 @@ pub fn pump_sink(db: &AimDb, scheme: &str, sink: Arc) -> Vec { + let Some(payload) = scratch.get(..*len) else { + log_error!( + "pump_sink: serializer returned invalid length {} for {}-byte scratch buffer", + len, + scratch.len() + ); + continue; + }; + payload + } + crate::connector::SerializedPayload::Owned(payload) => payload.as_slice(), + }; // Publish through the connector's pure I/O adapter. - if let Err(_e) = sink.publish(&dest, &cfg, &msg.payload).await { + if let Err(_e) = sink.publish(dest, &cfg, payload).await { log_error!("pump_sink: failed to publish to '{}': {:?}", dest, _e); } else { log_debug!("pump_sink: published to: {}", dest); diff --git a/aimdb-core/src/typed_api.rs b/aimdb-core/src/typed_api.rs index 136f0dc3..88102489 100644 --- a/aimdb-core/src/typed_api.rs +++ b/aimdb-core/src/typed_api.rs @@ -297,6 +297,17 @@ type FusedSerializeFn = Arc< + Sync, >; +/// Optional allocation-free serializer captured beside [`FusedSerializeFn`]. +/// +/// The callback writes into one pump-owned bounded scratch buffer. Returning +/// `BufferTooSmall` selects the owned serializer for that value; other failures +/// retain the existing skip-and-log behavior. +type FusedSerializeIntoFn = Arc< + dyn Fn(&crate::RuntimeContext, &T, &mut [u8]) -> Result + + Send + + Sync, +>; + /// The [`SerializedSource`](crate::connector::SerializedSource) built by /// `OutboundConnectorBuilder::finish()` — holds the typed consumer, /// serializer, and optional topic provider, so every per-message step stays @@ -304,6 +315,7 @@ type FusedSerializeFn = Arc< struct FusedSource { consumer: Consumer, serialize: FusedSerializeFn, + serialize_into: Option<(usize, FusedSerializeIntoFn)>, topic: Option>>, } @@ -311,10 +323,18 @@ impl crate::connector::SerializedSource for FusedSource where T: Send + Sync + 'static + Debug + Clone, { + fn serializer_scratch_capacity(&self) -> Option { + self.serialize_into.as_ref().map(|(capacity, _)| *capacity) + } + fn subscribe(&self) -> Box { Box::new(FusedReader { inner: self.consumer.subscribe(), serialize: self.serialize.clone(), + serialize_into: self + .serialize_into + .as_ref() + .map(|(_, serialize_into)| serialize_into.clone()), topic: self.topic.clone(), }) } @@ -329,6 +349,7 @@ where struct FusedReader { inner: crate::buffer::Reader, serialize: FusedSerializeFn, + serialize_into: Option>, topic: Option>>, } @@ -361,6 +382,85 @@ impl crate::connector::SerializedReader for FusedRead } }) } + + fn recv_into<'a>( + &'a mut self, + ctx: &'a crate::RuntimeContext, + scratch: &'a mut [u8], + ) -> crate::connector::RecvSerializedIntoFuture<'a> { + Box::pin(async move { + loop { + let value = self.inner.recv().await?; + let dest = self.topic.as_ref().and_then(|p| p.topic(&value)); + + let Some(serialize_into) = &self.serialize_into else { + match (self.serialize)(ctx, &value) { + Ok(payload) => { + return Ok(crate::connector::SerializedValueInto { + dest, + payload: crate::connector::SerializedPayload::Owned(payload), + }); + } + Err(_e) => { + log_error!( + "outbound link: failed to serialize {} (dest {:?}): {:?}", + core::any::type_name::(), + dest, + _e + ); + continue; + } + } + }; + + match serialize_into(ctx, &value, scratch) { + Ok(len) => { + if scratch.get(..len).is_none() { + log_error!( + "outbound link: serializer for {} returned invalid length {} for {}-byte scratch buffer", + core::any::type_name::(), + len, + scratch.len() + ); + continue; + } + return Ok(crate::connector::SerializedValueInto { + dest, + payload: crate::connector::SerializedPayload::Scratch { len }, + }); + } + Err(crate::connector::LinkCodecError::BufferTooSmall) => { + match (self.serialize)(ctx, &value) { + Ok(payload) => { + return Ok(crate::connector::SerializedValueInto { + dest, + payload: crate::connector::SerializedPayload::Owned(payload), + }); + } + Err(_e) => { + log_error!( + "outbound link: fallback serialization failed for {} (dest {:?}): {:?}", + core::any::type_name::(), + dest, + _e + ); + continue; + } + } + } + Err(_e) => { + log_error!( + "outbound link: failed to serialize {} into scratch buffer (dest {:?}): {:?}", + core::any::type_name::(), + dest, + _e + ); + continue; + } + } + } + }) + } } // ============================================================================ @@ -378,6 +478,14 @@ type TypedContextSerializerFn = Arc< + 'static, >; +/// Typed into-slice serializer stored by [`OutboundConnectorBuilder`]. +type TypedContextSerializerIntoFn = Arc< + dyn Fn(crate::RuntimeContext, &T, &mut [u8]) -> Result + + Send + + Sync + + 'static, +>; + /// Kind of execution stage, used to address per-stage profiling metrics and to /// remember which stage `RecordRegistrar::with_name` should rename. #[doc(hidden)] @@ -633,6 +741,7 @@ where url: url.to_string(), config: Vec::new(), context_serializer: None, + context_serializer_into: None, topic_provider: None, } } @@ -664,6 +773,7 @@ pub struct OutboundConnectorBuilder<'r, 'a, T: Send + Sync + 'static + Debug + C url: String, config: Vec<(String, String)>, context_serializer: Option>, + context_serializer_into: Option<(usize, TypedContextSerializerIntoFn)>, topic_provider: Option>>, } @@ -694,6 +804,32 @@ where self } + /// Adds an into-slice fast path beside the required owned serializer. + /// + /// One zero-initialized scratch buffer of `capacity` bytes is allocated when + /// each route pump starts, then reused for every message. A successful + /// callback returns the initialized prefix length. The framework validates + /// that length before borrowing the prefix. `BufferTooSmall` falls back to + /// [`with_serializer`](Self::with_serializer) for that value; any other + /// codec error is logged and the value is skipped. + /// + /// This method is an optimization only: callers must still install the + /// owned serializer so oversized and legacy values retain their old behavior. + pub fn with_serializer_into(mut self, capacity: usize, f: F) -> Self + where + F: Fn( + crate::RuntimeContext, + &T, + &mut [u8], + ) -> Result + + Send + + Sync + + 'static, + { + self.context_serializer_into = Some((capacity, Arc::new(f))); + self + } + /// Sets the operation timeout in milliseconds (the connector interprets /// it; passed as the `timeout_ms` option — see /// `ConnectorConfig::from_query`) @@ -770,6 +906,16 @@ where return self.registrar; }; + let serialize_into: Option<(usize, FusedSerializeIntoFn)> = + self.context_serializer_into.map(|(capacity, ser)| { + let adapted: FusedSerializeIntoFn = Arc::new( + move |ctx: &crate::RuntimeContext, value: &T, out: &mut [u8]| { + ser(ctx.clone(), value, out) + }, + ); + (capacity, adapted) + }); + // Validation: Check that connector builder is registered let has_connector = self .registrar @@ -840,6 +986,7 @@ where Box::new(FusedSource { consumer, serialize: serialize.clone(), + serialize_into: serialize_into.clone(), topic: topic_provider.clone(), }) as Box }) @@ -1693,10 +1840,24 @@ mod tests { FusedReader { inner: crate::buffer::Reader::new(Box::new(ScriptedReader { script })), serialize, + serialize_into: None, topic, } } + fn fused_reader_into( + script: Vec>, + serialize: FusedSerializeFn, + serialize_into: FusedSerializeIntoFn, + ) -> FusedReader { + FusedReader { + inner: crate::buffer::Reader::new(Box::new(ScriptedReader { script })), + serialize, + serialize_into: Some(serialize_into), + topic: None, + } + } + fn test_ctx() -> crate::RuntimeContext { crate::RuntimeContext::new(Arc::new(MockRuntime)) } @@ -1751,6 +1912,125 @@ mod tests { assert_eq!(msg.payload, 42i32.to_le_bytes().to_vec()); } + #[tokio::test] + async fn fused_reader_into_uses_scratch_without_owned_fallback() { + let owned_calls = Arc::new(AtomicUsize::new(0)); + let into_calls = Arc::new(AtomicUsize::new(0)); + let owned_counter = owned_calls.clone(); + let into_counter = into_calls.clone(); + let mut reader = fused_reader_into( + vec![Ok(TestRecord { value: 7 })], + Arc::new(move |_ctx, r| { + owned_counter.fetch_add(1, Ordering::SeqCst); + Ok(r.value.to_le_bytes().to_vec()) + }), + Arc::new(move |_ctx, r, out| { + into_counter.fetch_add(1, Ordering::SeqCst); + let bytes = r.value.to_le_bytes(); + out.get_mut(..bytes.len()) + .ok_or(crate::connector::LinkCodecError::BufferTooSmall)? + .copy_from_slice(&bytes); + Ok(bytes.len()) + }), + ); + let mut scratch = [0_u8; 8]; + + let msg = reader + .recv_into(&test_ctx(), &mut scratch) + .await + .expect("value"); + + assert_eq!( + msg.payload, + crate::connector::SerializedPayload::Scratch { len: 4 } + ); + assert_eq!(&scratch[..4], 7i32.to_le_bytes().as_slice()); + assert_eq!(into_calls.load(Ordering::SeqCst), 1); + assert_eq!(owned_calls.load(Ordering::SeqCst), 0); + } + + #[tokio::test] + async fn fused_reader_into_falls_back_once_when_scratch_is_small() { + let owned_calls = Arc::new(AtomicUsize::new(0)); + let owned_counter = owned_calls.clone(); + let mut reader = fused_reader_into( + vec![Ok(TestRecord { value: 9 })], + Arc::new(move |_ctx, r| { + owned_counter.fetch_add(1, Ordering::SeqCst); + Ok(r.value.to_le_bytes().to_vec()) + }), + Arc::new(|_ctx, _r, _out| Err(crate::connector::LinkCodecError::BufferTooSmall)), + ); + let mut scratch = [0_u8; 2]; + + let msg = reader + .recv_into(&test_ctx(), &mut scratch) + .await + .expect("fallback value"); + + assert_eq!( + msg.payload, + crate::connector::SerializedPayload::Owned(9i32.to_le_bytes().to_vec()) + ); + assert_eq!(owned_calls.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn fused_reader_into_rejects_invalid_length_and_skips_value() { + let mut reader = fused_reader_into( + vec![Ok(TestRecord { value: 1 }), Ok(TestRecord { value: 2 })], + Arc::new(|_ctx, r| Ok(r.value.to_le_bytes().to_vec())), + Arc::new(|_ctx, r, out| { + if r.value == 1 { + return Ok(out.len() + 1); + } + let bytes = r.value.to_le_bytes(); + out[..bytes.len()].copy_from_slice(&bytes); + Ok(bytes.len()) + }), + ); + let mut scratch = [0_u8; 8]; + + let msg = reader + .recv_into(&test_ctx(), &mut scratch) + .await + .expect("second value"); + + assert_eq!( + msg.payload, + crate::connector::SerializedPayload::Scratch { len: 4 } + ); + assert_eq!(&scratch[..4], 2i32.to_le_bytes().as_slice()); + } + + #[tokio::test] + async fn fused_reader_into_skips_invalid_data() { + let mut reader = fused_reader_into( + vec![Ok(TestRecord { value: 1 }), Ok(TestRecord { value: 2 })], + Arc::new(|_ctx, r| Ok(r.value.to_le_bytes().to_vec())), + Arc::new(|_ctx, r, out| { + if r.value == 1 { + return Err(crate::connector::LinkCodecError::InvalidData); + } + let bytes = r.value.to_le_bytes(); + out[..bytes.len()].copy_from_slice(&bytes); + Ok(bytes.len()) + }), + ); + let mut scratch = [0_u8; 8]; + + let msg = reader + .recv_into(&test_ctx(), &mut scratch) + .await + .expect("second value"); + + assert_eq!( + msg.payload, + crate::connector::SerializedPayload::Scratch { len: 4 } + ); + assert_eq!(&scratch[..4], 2i32.to_le_bytes().as_slice()); + } + /// The destination is resolved from the typed value while it is in hand. #[tokio::test] async fn fused_reader_resolves_dynamic_topic() { @@ -1813,6 +2093,14 @@ mod tests { .with_serializer(|_ctx: crate::RuntimeContext, r: &TestRecord| { Ok(r.value.to_le_bytes().to_vec()) }) + .with_serializer_into(4, |_ctx, r: &TestRecord, out| { + let encoded = r.value.to_le_bytes(); + let dest = out + .get_mut(..encoded.len()) + .ok_or(crate::connector::LinkCodecError::BufferTooSmall)?; + dest.copy_from_slice(&encoded); + Ok(encoded.len()) + }) .finish(); }); let (db, _runner) = builder.build().await.expect("build must succeed"); @@ -1820,14 +2108,23 @@ mod tests { let routes = db.collect_outbound_routes("mqtt"); assert_eq!(routes.len(), 1); assert_eq!(routes[0].topic, "tele/out"); + assert_eq!(routes[0].source.serializer_scratch_capacity(), Some(4)); let mut reader = routes[0].source.subscribe(); let ctx = db.runtime_ctx(); - let msg = reader.recv(&ctx).await.expect("value"); + let mut scratch = [0u8; 4]; + let msg = reader.recv_into(&ctx, &mut scratch).await.expect("value"); assert_eq!(msg.dest.as_deref(), Some("dyn/5")); - assert_eq!(msg.payload, 5i32.to_le_bytes().to_vec()); + assert_eq!( + msg.payload, + crate::connector::SerializedPayload::Scratch { len: 4 } + ); + assert_eq!(scratch, 5i32.to_le_bytes()); - let closed = reader.recv(&ctx).await.expect_err("buffer closed"); + let closed = reader + .recv_into(&ctx, &mut scratch) + .await + .expect_err("buffer closed"); assert!(matches!(closed, crate::DbError::BufferClosed { .. })); } } diff --git a/aimdb-data-contracts/CHANGELOG.md b/aimdb-data-contracts/CHANGELOG.md index ca7d8474..cb9fc189 100644 --- a/aimdb-data-contracts/CHANGELOG.md +++ b/aimdb-data-contracts/CHANGELOG.md @@ -16,6 +16,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- **Issue #177 — `Linkable::encode_into(&mut [u8])`.** Existing implementations + remain source-compatible through a checked `to_bytes`-and-copy default. A new + `ENCODE_BUFFER_CAPACITY` associated constant lets implementations advertise a + real allocation-free override; `linked_to` then installs the core's reusable + scratch-buffer fast path while retaining `to_bytes` for oversized values. + The new method uses `aimdb_core::connector::LinkCodecError`; the older + `from_bytes`/`to_bytes` `String` errors remain unchanged pending a separately + agreed breaking migration. - `SimulatableRegistrarExt::simulate(profile, rng)` — installs a `.source()` loop emitting `T::simulate(...)` on a timer. - `ObservableRegistrarExt::observe()` — feeds `T::signal()` into a core signal gauge (last/min/max/mean), surfaced on `record.list`/`record.get`/stage profiling; `ObservableRegistrarExt::log(node_id)` for the console-logging path (replaces the old `format_log`). - `LinkableRegistrarExt::linked_from(url)` / `linked_to(url)` — one-line `.link_from()`/`.link_to()` wiring defaulted to `T::from_bytes`/`T::to_bytes`. `linkable` now also requires `aimdb-core`. diff --git a/aimdb-data-contracts/src/lib.rs b/aimdb-data-contracts/src/lib.rs index 688cb1c8..64aacc1f 100644 --- a/aimdb-data-contracts/src/lib.rs +++ b/aimdb-data-contracts/src/lib.rs @@ -250,6 +250,16 @@ pub trait Observable: SchemaType { /// ``` #[cfg(feature = "linkable")] pub trait Linkable: SchemaType + Sized { + /// Preferred reusable scratch capacity for [`encode_into`](Self::encode_into). + /// + /// `None` keeps outbound links on the existing owned-`Vec` serializer. + /// Implementations that override `encode_into` with an allocation-free path + /// can advertise `Some(bytes)` so [`LinkableRegistrarExt::linked_to`] installs + /// the bounded scratch-buffer fast path. An undersized scratch buffer is a + /// performance fallback, not a correctness failure: the connector pipeline + /// retries that value through [`to_bytes`](Self::to_bytes). + const ENCODE_BUFFER_CAPACITY: Option = None; + /// Deserialize from bytes (e.g., MQTT payload). /// /// Returns `Err` with error message on parse failure. @@ -259,4 +269,32 @@ pub trait Linkable: SchemaType + Sized { /// /// Returns `Err` with error message on serialization failure. fn to_bytes(&self) -> Result, alloc::string::String>; + + /// Serialize into caller-owned storage and return the number of bytes used. + /// + /// The default implementation preserves source compatibility for existing + /// `Linkable` implementations by calling [`to_bytes`](Self::to_bytes) and + /// copying the result after a checked bounds test. It is therefore not + /// allocation-free. Implementations such as generated Postcard records can + /// override this method and set [`ENCODE_BUFFER_CAPACITY`](Self::ENCODE_BUFFER_CAPACITY) + /// to enable the allocation-free outbound fast path. + /// + /// On success, bytes after the returned length are left untouched. On any + /// error, callers must treat the contents of `buf` as unspecified: a true + /// streaming encoder may have written a prefix before discovering that the + /// buffer is too small. The compatibility implementation below performs a + /// checked copy and therefore leaves `buf` untouched on + /// [`LinkCodecError::BufferTooSmall`]. + /// + /// [`LinkCodecError::BufferTooSmall`]: aimdb_core::connector::LinkCodecError::BufferTooSmall + fn encode_into(&self, buf: &mut [u8]) -> Result { + use aimdb_core::connector::LinkCodecError; + + let bytes = self.to_bytes().map_err(|_| LinkCodecError::InvalidData)?; + let out = buf + .get_mut(..bytes.len()) + .ok_or(LinkCodecError::BufferTooSmall)?; + out.copy_from_slice(&bytes); + Ok(bytes.len()) + } } diff --git a/aimdb-data-contracts/src/linkable.rs b/aimdb-data-contracts/src/linkable.rs index 758701b1..ae3c919e 100644 --- a/aimdb-data-contracts/src/linkable.rs +++ b/aimdb-data-contracts/src/linkable.rs @@ -20,7 +20,8 @@ where /// `.link_from(url)` with the codec defaulted to `T::from_bytes`. fn linked_from(&mut self, url: &str) -> &mut RecordRegistrar<'a, T>; - /// `.link_to(url)` with the codec defaulted to `T::to_bytes`. + /// `.link_to(url)` with the codec defaulted to `T::to_bytes` and an + /// optional bounded [`Linkable::encode_into`] fast path. /// /// `Linkable::to_bytes`'s `String` error is mapped to /// `SerializeError::InvalidData` — the connector layer's serializer error @@ -39,17 +40,23 @@ where } fn linked_to(&mut self, url: &str) -> &mut RecordRegistrar<'a, T> { - self.link_to(url) - .with_serializer(|_ctx, value: &T| { - value.to_bytes().map_err(|_| SerializeError::InvalidData) - }) - .finish() + let builder = self.link_to(url).with_serializer(|_ctx, value: &T| { + value.to_bytes().map_err(|_| SerializeError::InvalidData) + }); + + match T::ENCODE_BUFFER_CAPACITY { + Some(capacity) => builder + .with_serializer_into(capacity, |_ctx, value: &T, out| value.encode_into(out)) + .finish(), + None => builder.finish(), + } } } #[cfg(test)] mod tests { use alloc::string::{String, ToString}; + use alloc::vec; use alloc::vec::Vec; use crate::{Linkable, SchemaType}; @@ -213,4 +220,59 @@ mod tests { let err = TestTemp::from_bytes(invalid).unwrap_err(); assert!(!err.is_empty(), "Error message should be descriptive"); } + + #[test] + fn default_encode_into_copies_exact_payload_and_preserves_tail() { + let temp = TestTemp { + celsius: 22.5, + timestamp: 1_704_326_400_000, + }; + let expected = temp.to_bytes().expect("serialization should succeed"); + let mut out = vec![0xA5; expected.len() + 4]; + + let written = temp + .encode_into(&mut out) + .expect("compatibility encode should succeed"); + + assert_eq!(written, expected.len()); + assert_eq!(&out[..written], expected.as_slice()); + assert_eq!(&out[written..], &[0xA5; 4]); + } + + #[test] + fn default_encode_into_accepts_exact_size_buffer() { + let temp = TestTemp { + celsius: 22.5, + timestamp: 1_704_326_400_000, + }; + let expected = temp.to_bytes().expect("serialization should succeed"); + let mut out = vec![0_u8; expected.len()]; + + let written = temp + .encode_into(&mut out) + .expect("exact-size output should succeed"); + + assert_eq!(written, expected.len()); + assert_eq!(out, expected); + } + + #[test] + fn default_encode_into_rejects_small_buffer_without_mutating_it() { + use aimdb_core::connector::LinkCodecError; + + let temp = TestTemp { + celsius: 22.5, + timestamp: 1_704_326_400_000, + }; + let required = temp.to_bytes().expect("serialization should succeed").len(); + let mut out = vec![0xA5; required - 1]; + let before = out.clone(); + + let err = temp + .encode_into(&mut out) + .expect_err("undersized output must fail"); + + assert_eq!(err, LinkCodecError::BufferTooSmall); + assert_eq!(out, before); + } } diff --git a/docs/design/024-M11-codegen-common-crate.md b/docs/design/024-M11-codegen-common-crate.md index bac0c171..09c2fee6 100644 --- a/docs/design/024-M11-codegen-common-crate.md +++ b/docs/design/024-M11-codegen-common-crate.md @@ -269,7 +269,12 @@ impl Linkable for WeatherObservationValue { ``` **For `serialization = "postcard"`:** generates `postcard::to_allocvec` / -`postcard::from_bytes` calls. Adds `postcard` to Cargo.toml dependencies. +`postcard::from_bytes` for the compatibility path and a true +`Linkable::encode_into` override backed by `postcard::to_slice`. The generated +type advertises a 256-byte preferred scratch capacity; generated outbound +connector chains install `with_serializer_into`, while values that do not fit +fall back to `to_allocvec`. The 256 bytes are a performance budget, not a wire +limit. Adds `postcard` to Cargo.toml dependencies. **For `serialization = "custom"`:** no deterministic `Linkable` impl is generated. Instead, the agent proposes a `Linkable` impl as an inline @@ -714,7 +719,8 @@ impl for connector wiring. 1. **Should `postcard` be a first-class serialisation target?** *Resolved — yes.* `serialization = "postcard"` emits - `postcard::to_allocvec` / `postcard::from_bytes`. Issue #155 split the + `postcard::to_allocvec` / `postcard::from_bytes`, plus an allocation-free + `postcard::to_slice` override for `Linkable::encode_into`. Issue #155 split the format-neutral `linkable` feature from the `linkable-json` convenience and added host roundtrip, mixed-codec compile, JSON-free dependency, and `thumbv7em-none-eabihf` drift checks for the generated Postcard path. diff --git a/docs/design/041-data-contracts-integration.md b/docs/design/041-data-contracts-integration.md index 77392455..0ff46e2e 100644 --- a/docs/design/041-data-contracts-integration.md +++ b/docs/design/041-data-contracts-integration.md @@ -285,6 +285,51 @@ Notes: | Per-URL opaque payloads (MQTT/KNX/serial/UDS) | fused per-link codec | `Linkable` | | AimX `record.get`/`record.set`/`subscribe` JSON | blanket `RemoteSerialize` ([`codec.rs:33`](../../aimdb-core/src/codec.rs)), automatic for serde types | none needed | +#### 3.3.1 Issue #177 follow-up: bounded into-slice encoding + +Issue #177 implements the non-allocating outbound half without forcing the +deferred breaking error migration. `Linkable` now has a source-compatible +`encode_into(&mut [u8]) -> Result` default and an +`ENCODE_BUFFER_CAPACITY` opt-in. Existing implementations stay on `to_bytes`; +generated Postcard implementations opt in and call `postcard::to_slice`. + +The ownership boundary is deliberately in the route pump: + +```text +route pump starts once + scratch: Vec with fixed length N + +for each T + FusedReader::recv_into(&mut scratch) + success(len <= N) -> SerializedPayload::Scratch { len } + BufferTooSmall -> old to_bytes() -> SerializedPayload::Owned(Vec) + InvalidData -> log and skip + pump validates scratch.get(..len) + Connector::publish(&scratch[..len]).await + only then may the next record overwrite scratch +``` + +Returning a slice borrowed from reader-owned storage was rejected: a boxed +async retry loop cannot safely lend that slice and then mutably reuse the same +storage on a later iteration. Returning only `{ len }` metadata lets the pump +borrow its own storage across `publish().await`, needs no `unsafe`, and keeps the +additive `SerializedReader::recv_into` default source-compatible for connector +authors. + +Error ownership is now explicit but only partially aligned: + +- `LinkCodecError::{BufferTooSmall, InvalidData}` names the record/per-link + codec layer; `SerializeError` remains its compatibility alias. +- Root `aimdb_core::CodecError` still belongs to session-envelope codecs and is + intentionally not reused. +- `Linkable::from_bytes`/`to_bytes` still return `String`. Migrating those + signatures remains the separately agreed breaking part of the §7 follow-up. + +The allocation claim is correspondingly narrow: successful generated Postcard +encoding performs zero heap allocations, measured over 10,000 calls. The +existing boxed `SerializedReader` future, dynamic topic `String`, and +connector-internal ownership copies are outside that claim. + ### 3.4 `Settable` → consumed by `aimdb-sync` `Settable` was built for the sync bridge, but `aimdb-sync` never referenced it: `SyncProducer::set(value: T)` took a fully constructed record, so every outside-the-thread caller hand-assembled the struct. The verb the trait was named for is a small inherent impl — and note it is distinct from AimX's existing `record.set {name, value}` (full JSON value through `JsonCodec`); `Settable` is **set-by-primitive**: diff --git a/tools/codegen-drift/state-postcard.toml b/tools/codegen-drift/state-postcard.toml index a51bfd5d..8b4c0068 100644 --- a/tools/codegen-drift/state-postcard.toml +++ b/tools/codegen-drift/state-postcard.toml @@ -29,3 +29,8 @@ description = "Sensor value" name = "sequence" type = "u32" description = "Monotonic sample sequence" + +[[records.connectors]] +protocol = "mqtt" +direction = "outbound" +url = "mqtt://readings/postcard/{variant}" diff --git a/tools/scripts/codegen-drift-check.sh b/tools/scripts/codegen-drift-check.sh index 63d5dd98..83743888 100755 --- a/tools/scripts/codegen-drift-check.sh +++ b/tools/scripts/codegen-drift-check.sh @@ -4,8 +4,9 @@ # The codegen templates print API-shaped strings the compiler never checks, # so they can silently rot against the real AimDB API. This script makes the # drift loud: it generates the default common/hub/flat outputs plus Postcard- -# only and mixed-codec common crates, then compiles and exercises them against -# the workspace at HEAD (design 038 §3.10 decision, issue #155). +# only and mixed-codec common crates plus a Postcard flat schema, then compiles +# and exercises them against the workspace at HEAD (design 038 §3.10 decision, +# issues #155 and #177). set -euo pipefail REPO="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)" @@ -15,7 +16,10 @@ MIXED_CODEC_STATE="$REPO/tools/codegen-drift/state-mixed-codec.toml" OUT="${CODEGEN_DRIFT_DIR:-$REPO/target/codegen-drift}" rm -rf "$OUT" -mkdir -p "$OUT/.aimdb" "$OUT/drift-check-flat/src" +mkdir -p \ + "$OUT/.aimdb" \ + "$OUT/drift-check-flat/src" \ + "$OUT/postcard-drift-flat/src" run_generate() { local state="$1" @@ -32,11 +36,13 @@ run_generate "$STATE" --rust "$OUT/drift-check-flat/src/generated_schema.rs" echo "── Generating Postcard-only and mixed-codec common crates" run_generate "$POSTCARD_STATE" --common-crate +run_generate "$POSTCARD_STATE" --rust "$OUT/postcard-drift-flat/src/generated_schema.rs" run_generate "$MIXED_CODEC_STATE" --common-crate # Exercise the generated Postcard codec as behavior, not just compilable tokens. mkdir -p "$OUT/postcard-drift-common/tests" cat > "$OUT/postcard-drift-common/tests/postcard_roundtrip.rs" <<'EOF' +use aimdb_core::connector::LinkCodecError; use aimdb_data_contracts::Linkable; use postcard_drift_common::PostcardReadingValue; @@ -50,6 +56,24 @@ fn generated_postcard_linkable_roundtrips() { let bytes = expected.to_bytes().expect("serialize with postcard"); assert!(!bytes.is_empty()); + let mut out = vec![0xA5; bytes.len() + 4]; + let written = expected + .encode_into(&mut out) + .expect("encode postcard into caller buffer"); + assert_eq!(written, bytes.len()); + assert_eq!(&out[..written], bytes.as_slice()); + assert_eq!(&out[written..], &[0xA5; 4]); + + let mut exact = vec![0_u8; bytes.len()]; + assert_eq!(expected.encode_into(&mut exact), Ok(bytes.len())); + assert_eq!(exact, bytes); + + let mut too_small = vec![0_u8; bytes.len() - 1]; + assert_eq!( + expected.encode_into(&mut too_small), + Err(LinkCodecError::BufferTooSmall) + ); + let actual = PostcardReadingValue::from_bytes(&bytes).expect("deserialize with postcard"); assert_eq!(actual.value.to_bits(), expected.value.to_bits()); assert_eq!(actual.sequence, expected.sequence); @@ -88,6 +112,33 @@ serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" EOF +# The Postcard flat output includes configure_schema(), so compiling this +# wrapper checks that generated outbound chains call with_serializer_into with +# the real core API. The common-crate output above checks the no_std codec impl. +cat > "$OUT/postcard-drift-flat/src/lib.rs" <<'EOF' +extern crate alloc; + +mod generated_schema; +pub use generated_schema::*; +EOF +cat > "$OUT/postcard-drift-flat/Cargo.toml" < Date: Sun, 12 Jul 2026 22:21:39 +0700 Subject: [PATCH 2/2] refactor(linkable): reuse `SerializeError` for `encode_into` Remove the newly introduced `LinkCodecError` and reuse the existing `aimdb_core::connector::SerializeError` across the into-slice API, postcard codegen, tests, benchmarks and documentation. Keep the legacy Linkable `from_bytes`/`to_bytes` String errors unchanged under the compatibility-first approach agreed in #177. --- CHANGELOG.md | 37 +++++++++---- aimdb-bench/README.md | 12 ++--- aimdb-bench/benches/b0_alloc_linkable.rs | 8 +-- aimdb-codegen/CHANGELOG.md | 11 ++-- aimdb-codegen/src/rust.rs | 8 +-- aimdb-core/CHANGELOG.md | 16 +++--- aimdb-core/src/connector.rs | 16 ++---- aimdb-core/src/lib.rs | 2 +- aimdb-core/src/typed_api.rs | 53 ++++++++----------- aimdb-data-contracts/CHANGELOG.md | 6 +-- aimdb-data-contracts/src/lib.rs | 12 ++--- aimdb-data-contracts/src/linkable.rs | 4 +- docs/design/041-data-contracts-integration.md | 17 +++--- tools/scripts/codegen-drift-check.sh | 4 +- 14 files changed, 99 insertions(+), 107 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 79fd09e1..5357efa6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). > **Note**: This is the global changelog for the AimDB project. For detailed changes to individual crates, see their respective CHANGELOG.md files: +> > - [aimdb-core/CHANGELOG.md](aimdb-core/CHANGELOG.md) > - [aimdb-codegen/CHANGELOG.md](aimdb-codegen/CHANGELOG.md) > - [aimdb-data-contracts/CHANGELOG.md](aimdb-data-contracts/CHANGELOG.md) @@ -105,16 +106,8 @@ multi-step migration round-trip suite and a trybuild compile-fail harness. ### Added -- **Zero-allocation generated Postcard encoding (Issue #177).** `Linkable` - gains a source-compatible `encode_into(&mut [u8])` fast path and generated - Postcard records implement it with `postcard::to_slice`. Core reuses one - bounded 256-byte scratch allocation per generated Postcard outbound route - (raw builders choose their capacity) and falls back to the existing owned - serializer when a value does not fit. The codec seam is gated - by a 10,000-iteration allocation-counting bench (0 allocations, 0 bytes) plus - host, `no_std` Cortex-M, and codegen-drift tests. The claim intentionally - excludes connector-internal payload copies and the existing boxed connector - reader future. ([aimdb-core](aimdb-core/CHANGELOG.md), +- **Zero-allocation generated Postcard encoding (Issue #177).** `Linkable` gains a source-compatible `encode_into(&mut [u8])` fast path and generated Postcard records implement it with `postcard::to_slice`. Core reuses onebounded 256-byte scratch allocation per generated Postcard outbound route (raw builders choose their capacity) and falls back to the existing owned serializer when a value does not fit. The codec seam is gated by a 10,000-iteration allocation-counting bench (0 allocations, 0 bytes) plus + host, `no_std` Cortex-M, and codegen-drift tests. The claim intentionally excludes connector-internal payload copies and the existing boxed connector reader future. ([aimdb-core](aimdb-core/CHANGELOG.md), [aimdb-data-contracts](aimdb-data-contracts/CHANGELOG.md), [aimdb-codegen](aimdb-codegen/CHANGELOG.md)) - **Embassy MQTT client TLS — `mqtts://` for embedded, WP7 ([design 044](docs/design/044-embassy-mqtt-tls.md)).** The Embassy MQTT connector can now dial a broker over TLS 1.3 (`embedded-tls`, pure-Rust `rustpki` certificate verification with `rsa`/`p384` so public CA chains verify out of the box), with SNI/hostname verification, DNS resolution, and a connector-internal SNTP time source that gates the first handshake so certificate validity is always checked against real time. Behind the new `embassy-tls` feature; `MqttConnectorBuilder::new` picks the transport from the URL scheme (`mqtt://` plain, `mqtts://` TLS) and gains `with_tls(TlsOptions)` (entropy, record buffers, SNTP server) and `with_credentials(username, password)` (MQTT CONNECT auth, both transports). The plain `mqtt://` path is untouched. The `embassy-mqtt-connector-demo` example gains a `tls` feature demonstrating the full flow against the repo's `dev/mosquitto` bench broker. ([aimdb-mqtt-connector](aimdb-mqtt-connector/CHANGELOG.md)) @@ -430,6 +423,7 @@ builder.configure::("sensor.temperature", |reg| { ``` **Key naming conventions:** + - Use dot-separated hierarchical names: `"sensors.indoor"`, `"config.app"` - Keys must be unique across all records (duplicate keys panic at registration) - For single-instance records, any descriptive string works (e.g., `"sensor.temperature"`, `"app.config"`) @@ -525,6 +519,7 @@ This release introduces **bidirectional connector support**, enabling true two-w ### Modified Crates See individual changelogs for detailed changes: + - **[aimdb-core](aimdb-core/CHANGELOG.md)**: Core connector architecture, router system, bidirectional APIs - **[aimdb-tokio-adapter](aimdb-tokio-adapter/CHANGELOG.md)**: Connector builder integration - **[aimdb-embassy-adapter](aimdb-embassy-adapter/CHANGELOG.md)**: Network stack access, connector support @@ -534,6 +529,7 @@ See individual changelogs for detailed changes: ### Migration Guide **1. Update connector registration:** + ```rust // Old (v0.1.0) let mqtt = MqttConnector::new("mqtt://broker:1883").await?; @@ -544,6 +540,7 @@ builder.with_connector(MqttConnectorBuilder::new("mqtt://broker:1883")) ``` **2. Make build() async:** + ```rust // Old (v0.1.0) let db = builder.build()?; @@ -553,6 +550,7 @@ let db = builder.build().await?; ``` **3. Use directional link methods:** + ```rust // Old (v0.1.0) .link("mqtt://sensors/temp") @@ -563,6 +561,7 @@ let db = builder.build().await?; ``` **4. Remove manual MQTT task spawning (Embassy):** + ```rust // Old (v0.1.0) - Manual task spawning required let mqtt_result = MqttConnector::create(...).await?; @@ -639,12 +638,14 @@ impl MyConnector { ``` **Why this change?** The new trait-based architecture provides: + - ✅ Symmetry with inbound routing (`ProducerTrait` ↔ `ConsumerTrait`) - ✅ Testability (can mock `ConsumerTrait` without real records) - ✅ Type safety via factory pattern (type capture at configuration time) - ✅ Maintainability (connector logic stays in connector crate) **Migration checklist for custom connectors:** + - [ ] Add `spawn_outbound_publishers()` method to connector implementation - [ ] Call `db.collect_outbound_routes(protocol_name)` in `ConnectorBuilder::build()` - [ ] Call `connector.spawn_outbound_publishers(db, routes)?` before returning @@ -665,6 +666,7 @@ See [Connector Development Guide](docs/design/012-M5-connector-development-guide ### Added #### Core Database (`aimdb-core`) + - Initial release of AimDB async in-memory database engine - Type-safe record system using `TypeId`-based routing - Three buffer types for different data flow patterns: @@ -678,6 +680,7 @@ See [Connector Development Guide](docs/design/012-M5-connector-development-guide - Remote access protocol (AimX v1) for cross-process introspection #### Runtime Adapters + - **Tokio Adapter** (`aimdb-tokio-adapter`): Full-featured std runtime support - Lock-free buffer implementations - Configurable buffer capacities @@ -688,6 +691,7 @@ See [Connector Development Guide](docs/design/012-M5-connector-development-guide - Compatible with ARM Cortex-M targets #### MQTT Connector (`aimdb-mqtt-connector`) + - Dual runtime support for both Tokio and Embassy - Automatic consumer registration via builder pattern - Topic mapping with QoS and retain configuration @@ -697,6 +701,7 @@ See [Connector Development Guide](docs/design/012-M5-connector-development-guide - Uses `mountain-mqtt` for embedded environments #### Developer Tools + - **MCP Server** (`aimdb-mcp`): LLM-powered introspection and debugging - Discover running AimDB instances - Query record values and schemas @@ -714,12 +719,14 @@ See [Connector Development Guide](docs/design/012-M5-connector-development-guide - Clean error handling #### Synchronous API (`aimdb-sync`) + - Blocking wrapper around async AimDB core - Thread-safe synchronous record access - Automatic Tokio runtime management - Ideal for gradual migration from sync to async #### Documentation & Examples + - Comprehensive README with architecture overview - Individual crate documentation with examples - 12 detailed design documents in `/docs/design` @@ -730,6 +737,7 @@ See [Connector Development Guide](docs/design/012-M5-connector-development-guide - `remote-access-demo`: Cross-process introspection server #### Build & CI Infrastructure + - Comprehensive Makefile with color-coded output - GitHub Actions workflows: - Continuous integration (format, lint, test) @@ -741,6 +749,7 @@ See [Connector Development Guide](docs/design/012-M5-connector-development-guide - Dev container setup for consistent development environment ### Design Goals Achieved + - ✅ Sub-50ms latency for data synchronization - ✅ Lock-free buffer operations - ✅ Cross-platform support (MCU → edge → cloud) @@ -748,21 +757,25 @@ See [Connector Development Guide](docs/design/012-M5-connector-development-guide - ✅ Protocol-agnostic connector architecture ### Known Limitations + - Kafka and DDS connectors planned for future releases - CLI tool is currently skeleton implementation - Performance benchmarks not yet included - Limited to Unix domain sockets for remote access (no TCP yet) ### Dependencies + - Rust 1.75+ required - Tokio 1.47+ for std environments - Embassy 0.9+ for embedded environments - See `deny.toml` for approved dependency licenses ### Breaking Changes + None (initial release) ### Migration Guide + Not applicable (initial release) --- @@ -774,6 +787,7 @@ Not applicable (initial release) AimDB v0.1.0 establishes the foundational architecture for async, in-memory data synchronization across MCU → edge → cloud environments. This release focuses on core functionality, dual runtime support, and developer tooling. **Highlights:** + - 🚀 Dual runtime support: Works on both standard library (Tokio) and embedded (Embassy) - 🔒 Type-safe record system eliminates runtime string lookups - 📦 Three buffer types cover most data patterns @@ -782,6 +796,7 @@ AimDB v0.1.0 establishes the foundational architecture for async, in-memory data - ✅ 27+ core tests, comprehensive CI/CD, security auditing **Get Started:** + ```bash cargo add aimdb-core aimdb-tokio-adapter ``` @@ -790,7 +805,7 @@ See [README.md](README.md) for quickstart guide and examples. **Feedback Welcome:** This is an early release. Please report issues, suggest features, or contribute at: -https://github.com/aimdb-dev/aimdb + --- diff --git a/aimdb-bench/README.md b/aimdb-bench/README.md index 3e0bd994..82e061db 100644 --- a/aimdb-bench/README.md +++ b/aimdb-bench/README.md @@ -20,7 +20,7 @@ Plus two informational benches that exercise the full runner-driven pipeline. **Adapters covered:** - **Tokio** — `b0_alloc_tokio`, `b1_b2_tokio` (host). -- **Postcard `Linkable` codec** — `b0_alloc_linkable` (host). This isolates the +- **postcard `Linkable` codec** — `b0_alloc_linkable` (host). This isolates the issue #177 `encode_into` seam; it is not a whole-connector allocation claim. - **Embassy** — `b0_alloc_embassy`, `b1_b2_embassy` (host). These drive the real [`EmbassyBuffer`] backend via @@ -96,6 +96,7 @@ Criterion writes HTML reports to `target/criterion/`. `b0_alloc_tokio` does not use Criterion. It runs a fixed warmup + batch cycle and writes JSON results to `aimdb-bench/target/bench-results/b0_alloc_tokio.json` (the path is anchored to the crate dir, so it is the same regardless of the directory you run from). **Measurement model:** + 1. Create buffer + reader. 2. Warmup ≥ 200 push → recv cycles (excluded from counters). 3. Reset allocation counters. @@ -108,12 +109,8 @@ The committed baseline lives in `data/baselines/b0_alloc_tokio.json`. When a cha `b0_alloc_embassy` mirrors this against the Embassy buffer backend and writes `data/baselines/b0_alloc_embassy.json` — also **0 allocs/msg** across all three profiles, confirming the Embassy `poll_recv` path is allocation-free on the host. The on-target B3 bench (`examples/embassy-bench-stm32h5`) re-checks the same 0-alloc claim against the real embedded allocator. -`b0_alloc_linkable` warms up for 1,000 iterations, then measures 10,000 -generated-shape Postcard `Linkable::encode_into` calls into one stack buffer. -The required result is **0 allocation calls and 0 allocated bytes**. It isolates -the codec seam: `SerializedReader` still returns a boxed future, dynamic topics -may allocate, and connector implementations may copy payload ownership after -the core pump lends them the scratch slice. +`b0_alloc_linkable` warms up for 1,000 iterations, then measures 10,000 generated-shape postcard `Linkable::encode_into` calls into one stack buffer. +The required result is **0 allocation calls and 0 allocated bytes**. It isolates the codec seam: `SerializedReader` still returns a boxed future, dynamic topics may allocate and connector implementations may copy payload ownership after the core pump lends them the scratch slice. > **Embassy eager registration (design 039 F8/F9).** An Embassy `SpmcRing` reader registers its embassy `Subscriber` eagerly, at `subscribe()` time — matching Tokio's `broadcast` — so no separate priming step is needed before the first `push`. @@ -139,4 +136,3 @@ Use them as a comparison point, not a regression gate. If they regress, `b0_allo - Criterion p99 can vary ±5–10% on noisy CI runners. Use p50 medians for trend comparisons. - Always specify `--release` or debug build consistently when comparing runs; optimizations differ by 5–50×. - `b_alloc_pipeline` uses a paced source: per-message pace tokens and notification channels. The coordination overhead is included in the measured window. - diff --git a/aimdb-bench/benches/b0_alloc_linkable.rs b/aimdb-bench/benches/b0_alloc_linkable.rs index 38dfdb8f..7909c697 100644 --- a/aimdb-bench/benches/b0_alloc_linkable.rs +++ b/aimdb-bench/benches/b0_alloc_linkable.rs @@ -8,7 +8,7 @@ use std::hint::black_box; use aimdb_bench::alloc::{reset, snapshot}; -use aimdb_core::connector::LinkCodecError; +use aimdb_core::connector::SerializeError; use aimdb_data_contracts::{Linkable, SchemaType}; use serde::{Deserialize, Serialize}; @@ -40,11 +40,11 @@ impl Linkable for PostcardReading { postcard::to_allocvec(self).map_err(|e| e.to_string()) } - fn encode_into(&self, out: &mut [u8]) -> Result { + fn encode_into(&self, out: &mut [u8]) -> Result { match postcard::to_slice(self, out) { Ok(used) => Ok(used.len()), - Err(postcard::Error::SerializeBufferFull) => Err(LinkCodecError::BufferTooSmall), - Err(_) => Err(LinkCodecError::InvalidData), + Err(postcard::Error::SerializeBufferFull) => Err(SerializeError::BufferTooSmall), + Err(_) => Err(SerializeError::InvalidData), } } } diff --git a/aimdb-codegen/CHANGELOG.md b/aimdb-codegen/CHANGELOG.md index 6fc6253a..66c07d04 100644 --- a/aimdb-codegen/CHANGELOG.md +++ b/aimdb-codegen/CHANGELOG.md @@ -9,13 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added -- **Issue #177 — generated Postcard records now have a true into-slice codec.** - Their `Linkable` implementation advertises a 256-byte reusable scratch - capacity and overrides `encode_into` with `postcard::to_slice`; generated - outbound connector chains install `with_serializer_into`. Values that exceed - the scratch capacity retain the existing `postcard::to_allocvec` fallback. - Host behavior, generated connector wiring, `no_std` Cortex-M compilation, and - the JSON-free dependency graph are covered by `make codegen-drift`. +- **Issue #177 - generated postcard records now have a true into-slice codec.** + Their `Linkable` implementation advertises a 256-byte reusable scratch capacity and overrides `encode_into` with `postcard::to_slice`; generated outbound connector chains install `with_serializer_into`. Values that exceed the scratch capacity retain the existing `postcard::to_allocvec` fallback. Host behavior, generated connector wiring, `no_std` Cortex-M compilation and the JSON-free dependency graph are covered by `make codegen-drift`. - Postcard codegen is now exercised end to end by `make codegen-drift`: a generated Postcard-only common crate roundtrips on the host, cross-compiles for `thumbv7em-none-eabihf`, and proves its normal dependency graph is JSON-free. A generated mixed JSON + Postcard crate is also compiled so codec feature/dependency drift fails in CI (#155). ### Fixed @@ -32,6 +27,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed - **Generated join handler stubs** updated to match the new task-model `on_triggers` API (Design 027). Multi-input task handlers are now generated as: + ```rust pub async fn task_handler( mut _rx: aimdb_core::transform::JoinEventRx, @@ -42,6 +38,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 } } ``` + Previously generated `fn task_handler(JoinTrigger, &mut (), &Producer<...>) -> Pin>` for the callback model. - `build_transform_call` for join tasks now emits `.on_triggers(handler)` instead of `.with_state(()).on_trigger(handler)`. diff --git a/aimdb-codegen/src/rust.rs b/aimdb-codegen/src/rust.rs index b3b7436d..6cd394e6 100644 --- a/aimdb-codegen/src/rust.rs +++ b/aimdb-codegen/src/rust.rs @@ -1037,13 +1037,13 @@ fn emit_linkable_postcard(rec: &RecordDef) -> TokenStream { fn encode_into( &self, buf: &mut [u8], - ) -> Result { + ) -> Result { match postcard::to_slice(self, buf) { Ok(used) => Ok(used.len()), Err(postcard::Error::SerializeBufferFull) => { - Err(aimdb_core::connector::LinkCodecError::BufferTooSmall) + Err(aimdb_core::connector::SerializeError::BufferTooSmall) } - Err(_) => Err(aimdb_core::connector::LinkCodecError::InvalidData), + Err(_) => Err(aimdb_core::connector::SerializeError::InvalidData), } } @@ -2375,7 +2375,7 @@ url = "mqtt://readings/postcard/{variant}" "Missing zero-allocation Postcard serializer:\n{rust}" ); assert!( - rust.contains("LinkCodecError::BufferTooSmall"), + rust.contains("SerializeError::BufferTooSmall"), "Missing Postcard small-buffer mapping:\n{rust}" ); assert!( diff --git a/aimdb-core/CHANGELOG.md b/aimdb-core/CHANGELOG.md index 9ec98a74..a60b0fd0 100644 --- a/aimdb-core/CHANGELOG.md +++ b/aimdb-core/CHANGELOG.md @@ -10,15 +10,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added - **Issue #177 — bounded into-slice serialization for outbound links.** New - `LinkCodecError` (with `SerializeError` retained as a compatibility alias), - `OutboundConnectorBuilder::with_serializer_into`, and the additive + `OutboundConnectorBuilder::with_serializer_into` and the additive `SerializedReader::recv_into`/`SerializedPayload` SPI let a fused typed reader - encode into caller-owned storage. `pump_sink` allocates one bounded scratch - buffer per route and reuses it across messages; an undersized buffer falls - back to the existing owned serializer for that value. Existing serializers, - third-party `SerializedReader` implementations, and `pump_client` keep their - owned-`Vec` behavior. This removes the codec allocation when an into-slice - encoder is installed; connector adapters may still copy the borrowed payload. + encode into caller-owned storage using the existing `SerializeError` contract. + `pump_sink` allocates one bounded scratch buffer per route and reuses it across + messages; an undersized buffer falls back to the existing owned serializer for + that value. Existing serializers, third-party `SerializedReader` + implementations, and `pump_client` keep their owned-`Vec` behavior. This + removes the codec allocation when an into-slice encoder is installed; + connector adapters may still copy the borrowed payload. - **`RecordRegistrar::signal_gauge(name, unit) -> SignalGaugeHandle`** (design 041 §3.2) — the core hook behind `aimdb-data-contracts`'s `Observable::observe()`. Feeding values via `SignalGaugeHandle::update(f64)` folds them into per-record `SignalStats` (last/min/max/mean), surfaced through `RecordMetadata::signal_stats` on `record.list`/`record.get`. Mirrors the `with_name` precedent: always callable, an inert no-op handle when the `observability` feature is off, so callers never `#[cfg]` on core's features. New public types: `SignalGaugeHandle` (always available), `SignalStats`/`SignalGauge`/`SignalStatsInfo` (feature `observability`). ### Fixed diff --git a/aimdb-core/src/connector.rs b/aimdb-core/src/connector.rs index 7ca3c663..6da82cc7 100644 --- a/aimdb-core/src/connector.rs +++ b/aimdb-core/src/connector.rs @@ -41,7 +41,7 @@ use alloc::format; use crate::{builder::AimDb, DbResult}; -/// Allocation-free error shared by record/link codec operations. +/// Error shared by outbound record serialization operations. /// /// This is deliberately separate from [`crate::CodecError`], which describes /// failures in a session envelope codec (AimX, WebSocket, and similar framed @@ -51,7 +51,7 @@ use crate::{builder::AimDb, DbResult}; /// Uses an enum instead of `String` for predictable `no_std` behavior and /// `defmt` logging support on Embassy targets. #[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum LinkCodecError { +pub enum SerializeError { /// Output buffer is too small for the serialized data BufferTooSmall, @@ -59,14 +59,8 @@ pub enum LinkCodecError { InvalidData, } -/// Backwards-compatible name for the outbound-only serializer error. -/// -/// New APIs should prefer [`LinkCodecError`], which also names the record/link -/// layer without implying that the error can only occur while encoding. -pub use LinkCodecError as SerializeError; - #[cfg(feature = "defmt")] -impl defmt::Format for LinkCodecError { +impl defmt::Format for SerializeError { fn format(&self, f: defmt::Formatter) { match self { Self::BufferTooSmall => defmt::write!(f, "BufferTooSmall"), @@ -76,7 +70,7 @@ impl defmt::Format for LinkCodecError { } #[cfg(feature = "std")] -impl std::fmt::Display for LinkCodecError { +impl std::fmt::Display for SerializeError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { Self::BufferTooSmall => write!(f, "Output buffer too small"), @@ -86,7 +80,7 @@ impl std::fmt::Display for LinkCodecError { } #[cfg(feature = "std")] -impl std::error::Error for LinkCodecError {} +impl std::error::Error for SerializeError {} /// One serialized record update, produced by a fused [`SerializedReader`] /// diff --git a/aimdb-core/src/lib.rs b/aimdb-core/src/lib.rs index fa700647..7ec25886 100644 --- a/aimdb-core/src/lib.rs +++ b/aimdb-core/src/lib.rs @@ -97,7 +97,7 @@ pub use profiling::{ // Connector Infrastructure exports pub use connector::TopicProvider; pub use connector::TopicResolverFn; -pub use connector::{ConnectorLink, ConnectorUrl, LinkAddress, LinkCodecError, SerializeError}; +pub use connector::{ConnectorLink, ConnectorUrl, LinkAddress, SerializeError}; pub use connector::{IngestFactoryFn, IngestFn}; pub use connector::{ SerializedPayload, SerializedReader, SerializedSource, SerializedValue, SerializedValueInto, diff --git a/aimdb-core/src/typed_api.rs b/aimdb-core/src/typed_api.rs index 88102489..4cf0f798 100644 --- a/aimdb-core/src/typed_api.rs +++ b/aimdb-core/src/typed_api.rs @@ -303,7 +303,7 @@ type FusedSerializeFn = Arc< /// `BufferTooSmall` selects the owned serializer for that value; other failures /// retain the existing skip-and-log behavior. type FusedSerializeIntoFn = Arc< - dyn Fn(&crate::RuntimeContext, &T, &mut [u8]) -> Result + dyn Fn(&crate::RuntimeContext, &T, &mut [u8]) -> Result + Send + Sync, >; @@ -429,7 +429,7 @@ impl crate::connector::SerializedReader for FusedRead payload: crate::connector::SerializedPayload::Scratch { len }, }); } - Err(crate::connector::LinkCodecError::BufferTooSmall) => { + Err(crate::connector::SerializeError::BufferTooSmall) => { match (self.serialize)(ctx, &value) { Ok(payload) => { return Ok(crate::connector::SerializedValueInto { @@ -480,7 +480,7 @@ type TypedContextSerializerFn = Arc< /// Typed into-slice serializer stored by [`OutboundConnectorBuilder`]. type TypedContextSerializerIntoFn = Arc< - dyn Fn(crate::RuntimeContext, &T, &mut [u8]) -> Result + dyn Fn(crate::RuntimeContext, &T, &mut [u8]) -> Result + Send + Sync + 'static, @@ -821,7 +821,7 @@ where crate::RuntimeContext, &T, &mut [u8], - ) -> Result + ) -> Result + Send + Sync + 'static, @@ -1193,7 +1193,10 @@ where #[cfg(test)] mod tests { use super::*; - use crate::DbResult; + use crate::{ + connector::{SerializeError, SerializedPayload, SerializedReader as _, TopicProvider}, + DbResult, + }; use core::pin::Pin; #[cfg(not(feature = "std"))] @@ -1792,8 +1795,6 @@ mod tests { // Fused outbound reader tests // ==================================================================== - use crate::connector::SerializedReader as _; - /// Buffer reader that replays a fixed script, then reports the buffer /// closed. struct ScriptedReader { @@ -1835,7 +1836,7 @@ mod tests { fn fused_reader( script: Vec>, serialize: FusedSerializeFn, - topic: Option>>, + topic: Option>>, ) -> FusedReader { FusedReader { inner: crate::buffer::Reader::new(Box::new(ScriptedReader { script })), @@ -1899,7 +1900,7 @@ mod tests { vec![Ok(TestRecord { value: 13 }), Ok(TestRecord { value: 42 })], Arc::new(|_ctx, r| { if r.value == 13 { - Err(crate::connector::SerializeError::InvalidData) + Err(SerializeError::InvalidData) } else { Ok(r.value.to_le_bytes().to_vec()) } @@ -1928,7 +1929,7 @@ mod tests { into_counter.fetch_add(1, Ordering::SeqCst); let bytes = r.value.to_le_bytes(); out.get_mut(..bytes.len()) - .ok_or(crate::connector::LinkCodecError::BufferTooSmall)? + .ok_or(SerializeError::BufferTooSmall)? .copy_from_slice(&bytes); Ok(bytes.len()) }), @@ -1940,10 +1941,7 @@ mod tests { .await .expect("value"); - assert_eq!( - msg.payload, - crate::connector::SerializedPayload::Scratch { len: 4 } - ); + assert_eq!(msg.payload, SerializedPayload::Scratch { len: 4 }); assert_eq!(&scratch[..4], 7i32.to_le_bytes().as_slice()); assert_eq!(into_calls.load(Ordering::SeqCst), 1); assert_eq!(owned_calls.load(Ordering::SeqCst), 0); @@ -1959,7 +1957,7 @@ mod tests { owned_counter.fetch_add(1, Ordering::SeqCst); Ok(r.value.to_le_bytes().to_vec()) }), - Arc::new(|_ctx, _r, _out| Err(crate::connector::LinkCodecError::BufferTooSmall)), + Arc::new(|_ctx, _r, _out| Err(SerializeError::BufferTooSmall)), ); let mut scratch = [0_u8; 2]; @@ -1970,7 +1968,7 @@ mod tests { assert_eq!( msg.payload, - crate::connector::SerializedPayload::Owned(9i32.to_le_bytes().to_vec()) + SerializedPayload::Owned(9i32.to_le_bytes().to_vec()) ); assert_eq!(owned_calls.load(Ordering::SeqCst), 1); } @@ -1996,10 +1994,7 @@ mod tests { .await .expect("second value"); - assert_eq!( - msg.payload, - crate::connector::SerializedPayload::Scratch { len: 4 } - ); + assert_eq!(msg.payload, SerializedPayload::Scratch { len: 4 }); assert_eq!(&scratch[..4], 2i32.to_le_bytes().as_slice()); } @@ -2010,7 +2005,7 @@ mod tests { Arc::new(|_ctx, r| Ok(r.value.to_le_bytes().to_vec())), Arc::new(|_ctx, r, out| { if r.value == 1 { - return Err(crate::connector::LinkCodecError::InvalidData); + return Err(SerializeError::InvalidData); } let bytes = r.value.to_le_bytes(); out[..bytes.len()].copy_from_slice(&bytes); @@ -2024,10 +2019,7 @@ mod tests { .await .expect("second value"); - assert_eq!( - msg.payload, - crate::connector::SerializedPayload::Scratch { len: 4 } - ); + assert_eq!(msg.payload, SerializedPayload::Scratch { len: 4 }); assert_eq!(&scratch[..4], 2i32.to_le_bytes().as_slice()); } @@ -2035,7 +2027,7 @@ mod tests { #[tokio::test] async fn fused_reader_resolves_dynamic_topic() { struct PositiveTopic; - impl crate::connector::TopicProvider for PositiveTopic { + impl TopicProvider for PositiveTopic { fn topic(&self, value: &TestRecord) -> Option { (value.value > 0).then(|| alloc::format!("dyn/{}", value.value)) } @@ -2074,7 +2066,7 @@ mod tests { } struct FixedTopic; - impl crate::connector::TopicProvider for FixedTopic { + impl TopicProvider for FixedTopic { fn topic(&self, value: &TestRecord) -> Option { Some(alloc::format!("dyn/{}", value.value)) } @@ -2097,7 +2089,7 @@ mod tests { let encoded = r.value.to_le_bytes(); let dest = out .get_mut(..encoded.len()) - .ok_or(crate::connector::LinkCodecError::BufferTooSmall)?; + .ok_or(SerializeError::BufferTooSmall)?; dest.copy_from_slice(&encoded); Ok(encoded.len()) }) @@ -2115,10 +2107,7 @@ mod tests { let mut scratch = [0u8; 4]; let msg = reader.recv_into(&ctx, &mut scratch).await.expect("value"); assert_eq!(msg.dest.as_deref(), Some("dyn/5")); - assert_eq!( - msg.payload, - crate::connector::SerializedPayload::Scratch { len: 4 } - ); + assert_eq!(msg.payload, SerializedPayload::Scratch { len: 4 }); assert_eq!(scratch, 5i32.to_le_bytes()); let closed = reader diff --git a/aimdb-data-contracts/CHANGELOG.md b/aimdb-data-contracts/CHANGELOG.md index cb9fc189..677f9cc5 100644 --- a/aimdb-data-contracts/CHANGELOG.md +++ b/aimdb-data-contracts/CHANGELOG.md @@ -21,9 +21,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 `ENCODE_BUFFER_CAPACITY` associated constant lets implementations advertise a real allocation-free override; `linked_to` then installs the core's reusable scratch-buffer fast path while retaining `to_bytes` for oversized values. - The new method uses `aimdb_core::connector::LinkCodecError`; the older - `from_bytes`/`to_bytes` `String` errors remain unchanged pending a separately - agreed breaking migration. + The new method reuses `aimdb_core::connector::SerializeError`; no additional + codec error type is introduced. The older `from_bytes`/`to_bytes` `String` + errors remain unchanged under the compatibility-first decision. - `SimulatableRegistrarExt::simulate(profile, rng)` — installs a `.source()` loop emitting `T::simulate(...)` on a timer. - `ObservableRegistrarExt::observe()` — feeds `T::signal()` into a core signal gauge (last/min/max/mean), surfaced on `record.list`/`record.get`/stage profiling; `ObservableRegistrarExt::log(node_id)` for the console-logging path (replaces the old `format_log`). - `LinkableRegistrarExt::linked_from(url)` / `linked_to(url)` — one-line `.link_from()`/`.link_to()` wiring defaulted to `T::from_bytes`/`T::to_bytes`. `linkable` now also requires `aimdb-core`. diff --git a/aimdb-data-contracts/src/lib.rs b/aimdb-data-contracts/src/lib.rs index 64aacc1f..6227f6b6 100644 --- a/aimdb-data-contracts/src/lib.rs +++ b/aimdb-data-contracts/src/lib.rs @@ -284,16 +284,16 @@ pub trait Linkable: SchemaType + Sized { /// streaming encoder may have written a prefix before discovering that the /// buffer is too small. The compatibility implementation below performs a /// checked copy and therefore leaves `buf` untouched on - /// [`LinkCodecError::BufferTooSmall`]. + /// [`SerializeError::BufferTooSmall`]. /// - /// [`LinkCodecError::BufferTooSmall`]: aimdb_core::connector::LinkCodecError::BufferTooSmall - fn encode_into(&self, buf: &mut [u8]) -> Result { - use aimdb_core::connector::LinkCodecError; + /// [`SerializeError::BufferTooSmall`]: aimdb_core::connector::SerializeError::BufferTooSmall + fn encode_into(&self, buf: &mut [u8]) -> Result { + use aimdb_core::connector::SerializeError; - let bytes = self.to_bytes().map_err(|_| LinkCodecError::InvalidData)?; + let bytes = self.to_bytes().map_err(|_| SerializeError::InvalidData)?; let out = buf .get_mut(..bytes.len()) - .ok_or(LinkCodecError::BufferTooSmall)?; + .ok_or(SerializeError::BufferTooSmall)?; out.copy_from_slice(&bytes); Ok(bytes.len()) } diff --git a/aimdb-data-contracts/src/linkable.rs b/aimdb-data-contracts/src/linkable.rs index ae3c919e..e5db2c60 100644 --- a/aimdb-data-contracts/src/linkable.rs +++ b/aimdb-data-contracts/src/linkable.rs @@ -258,7 +258,7 @@ mod tests { #[test] fn default_encode_into_rejects_small_buffer_without_mutating_it() { - use aimdb_core::connector::LinkCodecError; + use aimdb_core::connector::SerializeError; let temp = TestTemp { celsius: 22.5, @@ -272,7 +272,7 @@ mod tests { .encode_into(&mut out) .expect_err("undersized output must fail"); - assert_eq!(err, LinkCodecError::BufferTooSmall); + assert_eq!(err, SerializeError::BufferTooSmall); assert_eq!(out, before); } } diff --git a/docs/design/041-data-contracts-integration.md b/docs/design/041-data-contracts-integration.md index 0ff46e2e..a76b04b1 100644 --- a/docs/design/041-data-contracts-integration.md +++ b/docs/design/041-data-contracts-integration.md @@ -273,7 +273,7 @@ where ``` Notes: -- **Inbound matches exactly** (`with_deserializer` takes `Result`). **Outbound needs one lossy mapping**: `with_serializer` returns `Result, SerializeError>`, so the `String` detail is dropped to `SerializeError::InvalidData`. Aligning `Linkable`'s error type with the connector layer (a shared `CodecError` instead of `String` — which also removes an alloc on embedded) is a recorded follow-up (§7); it touches core connector signatures and doesn't block this verb. +- **Inbound matches exactly** (`with_deserializer` takes `Result`). **Outbound needs one lossy mapping**: `with_serializer` returns `Result, SerializeError>`, so the `String` detail is dropped to `SerializeError::InvalidData`. Issue #177 reuses that existing connector error for its additive `encode_into` method without changing either legacy signature. Replacing `String` across `Linkable` and the connector deserializer remains separate breaking follow-up scope (§7). - **The raw builders remain the escape hatch** for per-link options (`with_config`, QoS ext traits, topic providers/resolvers). `.linked_from`/`.linked_to` are the 80% path. - **JSON boilerplate gets a derive:** `#[derive(Linkable)]` in `aimdb-derive` emitting `serde_json::to_vec`/`from_slice` (JSON is the default format; binary formats implement by hand, as the KNX DPT codecs rightly do). Per the D1 rule, the derive replaces a hand-written JSON impl in the same change. After #155, this convenience lives behind `linkable-json`; base `linkable` stays format-neutral and JSON-free for Postcard/custom codecs. - **Coupling:** the `linkable` feature gains an `aimdb-core` dependency for the ext trait (`default-features = false, features = ["alloc"]`, same wiring as `observable`). `linkable-json` adds the optional `serde_json` + derive dependencies. @@ -289,7 +289,7 @@ Notes: Issue #177 implements the non-allocating outbound half without forcing the deferred breaking error migration. `Linkable` now has a source-compatible -`encode_into(&mut [u8]) -> Result` default and an +`encode_into(&mut [u8]) -> Result` default and an `ENCODE_BUFFER_CAPACITY` opt-in. Existing implementations stay on `to_bytes`; generated Postcard implementations opt in and call `postcard::to_slice`. @@ -316,14 +316,15 @@ borrow its own storage across `publish().await`, needs no `unsafe`, and keeps th additive `SerializedReader::recv_into` default source-compatible for connector authors. -Error ownership is now explicit but only partially aligned: +The compatibility-first error decision is explicit: -- `LinkCodecError::{BufferTooSmall, InvalidData}` names the record/per-link - codec layer; `SerializeError` remains its compatibility alias. +- `SerializeError::{BufferTooSmall, InvalidData}` is reused by both the existing + owned serializer and the new into-slice encoder. No new codec error type is + introduced. - Root `aimdb_core::CodecError` still belongs to session-envelope codecs and is intentionally not reused. - `Linkable::from_bytes`/`to_bytes` still return `String`. Migrating those - signatures remains the separately agreed breaking part of the §7 follow-up. + signatures remains separate breaking follow-up scope. The allocation claim is correspondingly narrow: successful generated Postcard encoding performs zero heap allocations, measured over 10,000 calls. The @@ -397,7 +398,7 @@ aimdb-core 2. **The RNG is caller-supplied**; `rand` loses `std_rng` everywhere. 3. **`Simulatable::Params` is an associated type**; `RandomWalkParams` ships as the migration path for existing scalar walks. 4. **`Observable` claims metrics because it delivers metrics**: kernel-only trait, `.observe()` → core signal gauges surfaced via `record.list`/`record.get`/profiling. -5. **`Linkable` gets its verbs in the same design**: `.linked_from`/`.linked_to` + `#[derive(Linkable)]`; the `String` → `SerializeError::InvalidData` mapping is accepted until the `CodecError` alignment (§7). +5. **`Linkable` gets its verbs in the same design**: `.linked_from`/`.linked_to` + `#[derive(Linkable)]`; the legacy `String` → `SerializeError::InvalidData` mapping remains accepted. Issue #177 reuses `SerializeError` for `encode_into`, while any breaking migration of the legacy `String` signatures stays deferred (§7). 6. **`Settable` is wired to its original purpose**: `SyncProducer::set_value` family in `aimdb-sync`, caller-side clock, `settable` feature gate. 7. **No trait bounds on `.source()`/`.tap()`/`.transform()`; no core→contracts dependency.** @@ -420,4 +421,4 @@ One branch/PR; each work item lands as its own commit carrying its caller: 4. `aimdb-sync` `set_value` family + `settable` feature gate + tests. 5. README verb/tier tables + CHANGELOG sweep. -**Out of scope / recorded for later:** runtime sim↔real failover (a single source future multiplexing on sensor staleness — still single-writer-safe, own doc); `CodecError` alignment across `Linkable` and the connector (de)serializer signatures; AimX set-by-primitive verb (second `Settable` consumer); signal-metrics history/percentiles beyond last/min/max/mean; auto-migration on the `linked_from` path (§3.6). +**Out of scope / recorded for later:** runtime sim↔real failover (a single source future multiplexing on sensor staleness — still single-writer-safe, own doc); breaking replacement of the legacy `String` errors across `Linkable` and connector deserializer signatures; AimX set-by-primitive verb (second `Settable` consumer); signal-metrics history/percentiles beyond last/min/max/mean; auto-migration on the `linked_from` path (§3.6). diff --git a/tools/scripts/codegen-drift-check.sh b/tools/scripts/codegen-drift-check.sh index 83743888..8c0e8156 100755 --- a/tools/scripts/codegen-drift-check.sh +++ b/tools/scripts/codegen-drift-check.sh @@ -42,7 +42,7 @@ run_generate "$MIXED_CODEC_STATE" --common-crate # Exercise the generated Postcard codec as behavior, not just compilable tokens. mkdir -p "$OUT/postcard-drift-common/tests" cat > "$OUT/postcard-drift-common/tests/postcard_roundtrip.rs" <<'EOF' -use aimdb_core::connector::LinkCodecError; +use aimdb_core::connector::SerializeError; use aimdb_data_contracts::Linkable; use postcard_drift_common::PostcardReadingValue; @@ -71,7 +71,7 @@ fn generated_postcard_linkable_roundtrips() { let mut too_small = vec![0_u8; bytes.len() - 1]; assert_eq!( expected.encode_into(&mut too_small), - Err(LinkCodecError::BufferTooSmall) + Err(SerializeError::BufferTooSmall) ); let actual = PostcardReadingValue::from_bytes(&bytes).expect("deserialize with postcard");