From 6917da33d708890f4d43534debb64194d99b85c0 Mon Sep 17 00:00:00 2001 From: Amidamaru Date: Thu, 16 Jul 2026 12:00:15 +0700 Subject: [PATCH 1/2] feat(data-contracts): add per-link codec selection - add `LinkCodec` with builder and registrar overrides - add feature-gated JSON and bounded postcard codecs - preserve reusable scratch encoding with owned fallback - verify MQTT extension composition and zero-allocation fast path - update feature matrix, changelogs, README and design docs --- CHANGELOG.md | 10 + Cargo.lock | 4 + Makefile | 16 +- README.md | 19 +- aimdb-bench/Cargo.toml | 1 + aimdb-bench/benches/b0_alloc_linkable.rs | 77 ++- aimdb-bench/src/alloc.rs | 6 + aimdb-data-contracts/CHANGELOG.md | 10 +- aimdb-data-contracts/Cargo.toml | 8 + aimdb-data-contracts/src/lib.rs | 13 +- aimdb-data-contracts/src/link_codec.rs | 604 ++++++++++++++++++ aimdb-data-contracts/src/linkable.rs | 46 +- aimdb-mqtt-connector/Cargo.toml | 4 + aimdb-mqtt-connector/tests/link_ext_tests.rs | 53 +- docs/design/041-data-contracts-integration.md | 2 + docs/design/045-per-link-codec-selection.md | 330 ++++++++++ 16 files changed, 1174 insertions(+), 29 deletions(-) create mode 100644 aimdb-data-contracts/src/link_codec.rs create mode 100644 docs/design/045-per-link-codec-selection.md diff --git a/CHANGELOG.md b/CHANGELOG.md index 66132e5..d7d71e9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added — per-link record codecs + +- **Issue #178:** one `Linkable` record type can now select JSON, bounded + Postcard, or a custom codec independently on each inbound/outbound connector + route through `linked_*_with(url, codec)` or builder-level + `with_link_codec(codec)`. Selection is fused at registration time; bounded + Postcard routes retain the reusable scratch/owned-overflow behavior from + issue #177 without a per-message registry, lock, or lookup. See + [Design 045](docs/design/045-per-link-codec-selection.md). + ### Changed — Design 038 simplification pass (breaking) Implementation of the accepted items of [design 038](docs/design/038-technical-debt-and-simplification-review.md) diff --git a/Cargo.lock b/Cargo.lock index dd48ca4..12bf40c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -160,6 +160,8 @@ version = "0.4.0" dependencies = [ "aimdb-core", "aimdb-derive", + "futures", + "postcard", "rand 0.10.1", "serde", "serde_json", @@ -248,6 +250,7 @@ name = "aimdb-mqtt-connector" version = "0.6.0" dependencies = [ "aimdb-core", + "aimdb-data-contracts", "aimdb-embassy-adapter", "aimdb-tokio-adapter", "async-stream", @@ -265,6 +268,7 @@ dependencies = [ "mountain-mqtt-embassy", "rand_core 0.6.4", "rumqttc", + "serde", "static_cell", "thiserror 2.0.17", "tokio", diff --git a/Makefile b/Makefile index 859f1d7..054486d 100644 --- a/Makefile +++ b/Makefile @@ -63,11 +63,13 @@ help: build: @printf "$(GREEN)Building AimDB (all valid combinations)...$(NC)\n" @printf "$(YELLOW) → Building aimdb-data-contracts (std)$(NC)\n" - cargo build --package aimdb-data-contracts --features "std,simulatable,migratable,observable,linkable-json" + cargo build --package aimdb-data-contracts --features "std,simulatable,migratable,observable,linkable-json,linkable-postcard" @printf "$(YELLOW) → Building aimdb-data-contracts (no_std)$(NC)\n" cargo build --package aimdb-data-contracts --no-default-features --features alloc @printf "$(YELLOW) → Building aimdb-data-contracts (no_std + format-neutral linkable)$(NC)\n" cargo build --package aimdb-data-contracts --no-default-features --features alloc,linkable + @printf "$(YELLOW) → Building aimdb-data-contracts (no_std + linkable-postcard)$(NC)\n" + cargo build --package aimdb-data-contracts --no-default-features --features alloc,linkable-postcard @printf "$(YELLOW) → Building aimdb-data-contracts (no_std + linkable-json + migratable)$(NC)\n" cargo build --package aimdb-data-contracts --no-default-features --features alloc,linkable-json,migratable @printf "$(YELLOW) → Building aimdb-core (no_std + alloc)$(NC)\n" @@ -114,9 +116,11 @@ build: test: @printf "$(GREEN)Running all tests (valid combinations)...$(NC)\n" @printf "$(YELLOW) → Testing aimdb-data-contracts (std)$(NC)\n" - cargo test --package aimdb-data-contracts --features "std,simulatable,migratable,observable,linkable-json" + cargo test --package aimdb-data-contracts --features "std,simulatable,migratable,observable,linkable-json,linkable-postcard" @printf "$(YELLOW) → Testing aimdb-data-contracts (no_std + alloc + format-neutral linkable)$(NC)\n" cargo test --package aimdb-data-contracts --no-default-features --features alloc,linkable + @printf "$(YELLOW) → Testing aimdb-data-contracts (no_std + alloc + linkable-postcard)$(NC)\n" + cargo test --package aimdb-data-contracts --no-default-features --features alloc,linkable-postcard @printf "$(YELLOW) → Testing aimdb-data-contracts (no_std + alloc + linkable-json + migratable)$(NC)\n" cargo test --package aimdb-data-contracts --no-default-features --features alloc,linkable-json,migratable @printf "$(YELLOW) → Testing aimdb-core (no_std + alloc)$(NC)\n" @@ -211,11 +215,13 @@ clippy: @printf "$(YELLOW) → Clippy on aimdb-derive$(NC)\n" cargo clippy --package aimdb-derive --all-targets -- -D warnings @printf "$(YELLOW) → Clippy on aimdb-data-contracts (std)$(NC)\n" - cargo clippy --package aimdb-data-contracts --features "std,simulatable,migratable,observable,linkable-json" --all-targets -- -D warnings + cargo clippy --package aimdb-data-contracts --features "std,simulatable,migratable,observable,linkable-json,linkable-postcard" --all-targets -- -D warnings @printf "$(YELLOW) → Clippy on aimdb-data-contracts (no_std + alloc)$(NC)\n" cargo clippy --package aimdb-data-contracts --no-default-features --features alloc -- -D warnings @printf "$(YELLOW) → Clippy on aimdb-data-contracts (no_std + alloc + format-neutral linkable)$(NC)\n" cargo clippy --package aimdb-data-contracts --no-default-features --features alloc,linkable --all-targets -- -D warnings + @printf "$(YELLOW) → Clippy on aimdb-data-contracts (no_std + alloc + linkable-postcard)$(NC)\n" + cargo clippy --package aimdb-data-contracts --no-default-features --features alloc,linkable-postcard --all-targets -- -D warnings @printf "$(YELLOW) → Clippy on aimdb-data-contracts (no_std + alloc + linkable-json + migratable)$(NC)\n" cargo clippy --package aimdb-data-contracts --no-default-features --features alloc,linkable-json,migratable --all-targets -- -D warnings @printf "$(YELLOW) → Clippy on aimdb-core (no_std + alloc)$(NC)\n" @@ -295,7 +301,7 @@ doc: @mkdir -p target/doc-final/cloud @mkdir -p target/doc-final/embedded @printf "$(YELLOW) → Building cloud/edge documentation$(NC)\n" - cargo doc --package aimdb-data-contracts --features "std,simulatable,migratable,observable,linkable-json" --no-deps + cargo doc --package aimdb-data-contracts --features "std,simulatable,migratable,observable,linkable-json,linkable-postcard" --no-deps cargo doc --package aimdb-core --features "std,tracing,observability" --no-deps cargo doc --package aimdb-tokio-adapter --features "tokio-runtime,tracing,observability" --no-deps cargo doc --package aimdb-sync --no-deps @@ -344,6 +350,8 @@ test-embedded: cargo check --package aimdb-data-contracts --target thumbv7em-none-eabihf --target-dir $(EMBEDDED_CHECK_TARGET_DIR) --no-default-features --features alloc @printf "$(YELLOW) → Checking aimdb-data-contracts (no_std + alloc + format-neutral linkable) on thumbv7em-none-eabihf target$(NC)\n" cargo check --package aimdb-data-contracts --target thumbv7em-none-eabihf --target-dir $(EMBEDDED_CHECK_TARGET_DIR) --no-default-features --features alloc,linkable + @printf "$(YELLOW) → Checking aimdb-data-contracts (no_std + alloc + linkable-postcard) on thumbv7em-none-eabihf target$(NC)\n" + cargo check --package aimdb-data-contracts --target thumbv7em-none-eabihf --target-dir $(EMBEDDED_CHECK_TARGET_DIR) --no-default-features --features alloc,linkable-postcard @printf "$(YELLOW) → Checking aimdb-data-contracts (no_std + alloc + linkable-json + migratable) on thumbv7em-none-eabihf target$(NC)\n" cargo check --package aimdb-data-contracts --target thumbv7em-none-eabihf --target-dir $(EMBEDDED_CHECK_TARGET_DIR) --no-default-features --features alloc,linkable-json,migratable @printf "$(YELLOW) → Checking weather-mesh-common (no_std migratable, real TemperatureV1ToV2 chain, no direct serde_json dep) on thumbv7em-none-eabihf target$(NC)\n" diff --git a/README.md b/README.md index a037d32..dd67e64 100644 --- a/README.md +++ b/README.md @@ -151,13 +151,30 @@ Every capability is an opt-in trait on your schema type: implement it and **exac | Contract | Implement when… | Verb it unlocks | Tier | | --- | --- | --- | --- | -| [`Linkable`](https://aimdb.dev/blog/connectors-where-aimdb-meets-the-real-world) | the record is mirrored to/from an endpoint (MQTT, KNX, serial, UDS…) | `.linked_from(url)` / `.linked_to(url)` (`linkable-json` provides the JSON derive; codegen supports Postcard) | wire (prod) | +| [`Linkable`](https://aimdb.dev/blog/connectors-where-aimdb-meets-the-real-world) | the record is mirrored to/from an endpoint (MQTT, KNX, serial, UDS…) | `.linked_from(url)` / `.linked_to(url)` by default; `*_with(url, codec)` selects JSON/Postcard/custom per link | wire (prod) | | [`Streamable`](https://aimdb.dev/blog/streamable-crossing-boundaries) | the record streams to browsers as schema-named JSON | ws-connector `.register::()` | wire (prod) | | [`Migratable`](https://aimdb.dev/blog/schema-migration-without-ceremony) | the schema evolved across versions | `migration_chain!` | wire (prod) | | `Settable` | sync code outside AimDB sets the value | `SyncProducer::set_value(v)` | wire (prod) | | `Observable` | the value is worth watching in production | `.observe()` → live last/min/max/mean on `record.list`/`record.get` | introspection (prod, optional) | | `Simulatable` | the type can generate realistic synthetic data | `.simulate(profile, rng)` | **dev-only — never ships in prod** | +One record type can use different wire formats without runtime codec lookup: + +```rust,ignore +use aimdb_data_contracts::{ + link_codecs::{Json, Postcard}, + LinkCodecRegistrarExt, +}; + +reg.linked_to_with("serial://mcu/temperature", Postcard::<128>); +reg.linked_to_with("mqtt://cloud/temperature", Json); +``` + +Enable `linkable-json` and/or `linkable-postcard`; the existing `.linked_*` +methods still use the record's `Linkable` implementation. See +[Design 045](docs/design/045-per-link-codec-selection.md) for the bounded +scratch/fallback model. + `Simulatable` is the odd one out: it lives behind the `simulatable` feature (never a default) and switching from simulated to real data is one `#[cfg]` in your app: ```rust diff --git a/aimdb-bench/Cargo.toml b/aimdb-bench/Cargo.toml index 19b08f2..26079d4 100644 --- a/aimdb-bench/Cargo.toml +++ b/aimdb-bench/Cargo.toml @@ -94,6 +94,7 @@ serde_json = { workspace = true, optional = true } aimdb-data-contracts = { path = "../aimdb-data-contracts", default-features = false, features = [ "alloc", "linkable", + "linkable-postcard", "migratable", ], optional = true } diff --git a/aimdb-bench/benches/b0_alloc_linkable.rs b/aimdb-bench/benches/b0_alloc_linkable.rs index 7909c69..785ad28 100644 --- a/aimdb-bench/benches/b0_alloc_linkable.rs +++ b/aimdb-bench/benches/b0_alloc_linkable.rs @@ -1,4 +1,4 @@ -//! B0-Linkable — allocation gate for the Postcard `Linkable::encode_into` path. +//! B0-Linkable — allocation gate for direct and per-link Postcard encoding. //! //! This deliberately measures the codec seam, not a complete connector. The //! core pump still uses a boxed `SerializedReader` future and connector adapters @@ -9,11 +9,12 @@ use std::hint::black_box; use aimdb_bench::alloc::{reset, snapshot}; use aimdb_core::connector::SerializeError; -use aimdb_data_contracts::{Linkable, SchemaType}; +use aimdb_data_contracts::{link_codecs, LinkCodec, Linkable, SchemaType}; use serde::{Deserialize, Serialize}; const WARMUP_ITERS: usize = 1_000; const MEASURE_ITERS: usize = 10_000; +const ALLOCATOR_PROBE_BYTES: usize = 64; #[global_allocator] static GLOBAL: aimdb_bench::alloc::CountingAllocator = @@ -58,6 +59,40 @@ fn encode_batch(reading: &PostcardReading, out: &mut [u8; 256], iterations: usiz } } +fn encode_per_link_codec_batch(reading: &PostcardReading, out: &mut [u8; 256], iterations: usize) { + let codec = link_codecs::Postcard::<256>; + for _ in 0..iterations { + let written = codec + .encode_into(black_box(reading), black_box(out.as_mut_slice())) + .expect("per-link Postcard encode_into should fit the fixed scratch buffer"); + black_box(&out[..written]); + } +} + +/// Prove that this binary's global allocator is actually wired through the +/// counters. Run after the measured windows so the deliberate allocation +/// cannot contaminate either zero-allocation result. +fn allocation_counter_positive_control() -> (u64, u64) { + reset(); + + let mut probe = Vec::::with_capacity(black_box(ALLOCATOR_PROBE_BYTES)); + probe.push(black_box(0xA5)); + black_box(probe.as_slice()); + + let measured = snapshot(); + assert!( + measured.0 >= 1, + "allocation counter missed the deliberate Vec allocation" + ); + assert!( + measured.1 >= ALLOCATOR_PROBE_BYTES as u64, + "byte counter missed the deliberate Vec capacity" + ); + + drop(probe); + measured +} + fn main() { let reading = PostcardReading { value: 23.75, @@ -68,19 +103,41 @@ fn main() { encode_batch(&reading, &mut out, WARMUP_ITERS); reset(); encode_batch(&reading, &mut out, MEASURE_ITERS); - let (allocations, bytes) = snapshot(); + let (direct_allocations, direct_bytes) = snapshot(); + + encode_per_link_codec_batch(&reading, &mut out, WARMUP_ITERS); + reset(); + encode_per_link_codec_batch(&reading, &mut out, MEASURE_ITERS); + let (codec_allocations, codec_bytes) = snapshot(); - println!("=== B0 Linkable Postcard encode_into ==="); + // This resets the global counters only after both result tuples have been + // captured, then deliberately allocates to reject a disconnected or broken + // counting allocator that would otherwise report a misleading zero. + let (probe_allocations, probe_bytes) = allocation_counter_positive_control(); + + println!("=== B0 Postcard encode_into ==="); println!(" Iterations : {MEASURE_ITERS}"); - println!(" Allocations : {allocations}"); - println!(" Bytes : {bytes}"); + println!(" Allocator probe allocations : {probe_allocations}"); + println!(" Allocator probe bytes : {probe_bytes}"); + println!(" Direct Linkable allocations : {direct_allocations}"); + println!(" Direct Linkable bytes : {direct_bytes}"); + println!(" Per-link codec allocations : {codec_allocations}"); + println!(" Per-link codec bytes : {codec_bytes}"); assert_eq!( - allocations, 0, - "encode_into allocated in the measured window" + direct_allocations, 0, + "direct Linkable::encode_into allocated in the measured window" + ); + assert_eq!( + direct_bytes, 0, + "direct Linkable::encode_into allocated bytes in the measured window" + ); + assert_eq!( + codec_allocations, 0, + "per-link Postcard codec allocated in the measured window" ); assert_eq!( - bytes, 0, - "encode_into allocated bytes in the measured window" + codec_bytes, 0, + "per-link Postcard codec allocated bytes in the measured window" ); } diff --git a/aimdb-bench/src/alloc.rs b/aimdb-bench/src/alloc.rs index 3cc28e2..d78dfe4 100644 --- a/aimdb-bench/src/alloc.rs +++ b/aimdb-bench/src/alloc.rs @@ -33,6 +33,10 @@ pub struct CountingAllocator(pub A); // the only side-effect is the atomic counter updates. unsafe impl GlobalAlloc for CountingAllocator { unsafe fn alloc(&self, layout: Layout) -> *mut u8 { + // These are independent observational counters: they publish no data + // and protect no other memory, so cross-variable ordering is irrelevant + // and Relaxed is sufficient even when allocator calls come from more + // than one thread. ALLOC_COUNT.fetch_add(1, Ordering::Relaxed); ALLOC_BYTES.fetch_add(layout.size() as u64, Ordering::Relaxed); // SAFETY: caller guarantees `layout` is valid; delegated to inner. @@ -48,6 +52,8 @@ unsafe impl GlobalAlloc for CountingAllocator { /// Reset both counters to zero. /// /// Call once after the warmup phase, immediately before the measured window. +/// For deterministic attribution, the caller must ensure unrelated threads are +/// not allocating concurrently with `reset` or the measured window. #[inline] pub fn reset() { ALLOC_COUNT.store(0, Ordering::Relaxed); diff --git a/aimdb-data-contracts/CHANGELOG.md b/aimdb-data-contracts/CHANGELOG.md index 677f9cc..89854d0 100644 --- a/aimdb-data-contracts/CHANGELOG.md +++ b/aimdb-data-contracts/CHANGELOG.md @@ -16,14 +16,8 @@ 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 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. +- **Issue #178 — per-link codec selection.** `LinkCodec`, `LinkCodecBuilderExt::with_link_codec` and `LinkCodecRegistrarExt::{linked_from_with, linked_to_with}` let the same record type use different formats on different connector routes without a runtime registry. Built-in `link_codecs::Json` and bounded `link_codecs::Postcard` markers are feature-gated by `linkable-json` and the new `linkable-postcard`; Postcard reuses the route-local scratch and owned overflow fallback added for issue #177. Existing `.linked_from` and `.linked_to` behavior is unchanged. +- **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 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/Cargo.toml b/aimdb-data-contracts/Cargo.toml index 782e3a0..8c13ed5 100644 --- a/aimdb-data-contracts/Cargo.toml +++ b/aimdb-data-contracts/Cargo.toml @@ -20,6 +20,10 @@ linkable = ["alloc", "aimdb-core"] # JSON convenience layered over the format-neutral contract. The derive emits # serde_json calls through this crate's private re-export. linkable-json = ["linkable", "serde_json", "dep:aimdb-derive"] +# Compact binary codec with a caller-chosen bounded scratch capacity. Postcard +# stays optional and no_std-compatible; `alloc` supplies the correctness +# fallback for values larger than the preferred route buffer. +linkable-postcard = ["linkable", "dep:postcard"] # Dev-tier — never a default feature. The ext trait needs core's registrar, # same coupling `observable` has. `rand` is the tracer CI uses to prove a # production graph is sim-free (`make check-no-sim`). @@ -33,6 +37,9 @@ settable = [] [dependencies] serde = { version = "1.0", default-features = false, features = ["derive"] } serde_json = { workspace = true, optional = true } +postcard = { version = "1.0", default-features = false, features = [ + "alloc", +], optional = true } # migration_chain! + #[derive(Linkable)] proc-macros (build-time only; no # target/runtime/no_std impact — see aimdb-derive). No cycle: aimdb-derive @@ -55,5 +62,6 @@ optional = true default-features = false [dev-dependencies] +futures = { workspace = true } serde_json = "1.0" trybuild = "1.0" diff --git a/aimdb-data-contracts/src/lib.rs b/aimdb-data-contracts/src/lib.rs index 6227f6b..48f17dd 100644 --- a/aimdb-data-contracts/src/lib.rs +++ b/aimdb-data-contracts/src/lib.rs @@ -69,11 +69,17 @@ pub mod __private { mod streamable; pub use streamable::Streamable; +#[cfg(feature = "linkable")] +mod link_codec; + #[cfg(feature = "linkable")] mod linkable; #[cfg(feature = "linkable")] -pub use linkable::LinkableRegistrarExt; +pub use link_codec::{link_codecs, LinkCodec, LinkCodecBuilderExt}; + +#[cfg(feature = "linkable")] +pub use linkable::{LinkCodecRegistrarExt, LinkableRegistrarExt}; /// JSON `#[derive(Linkable)]` (feature `linkable-json`) — see [`Linkable`] and /// `aimdb_derive::Linkable`'s docs. @@ -232,7 +238,10 @@ pub trait Observable: SchemaType { /// /// The `linkable` feature is format-neutral. Enable `linkable-json` for the /// JSON `#[derive(Linkable)]` convenience, or implement these methods directly -/// for Postcard and other shared-contract formats. +/// for another default format. To select a different format on each route, use +/// [`LinkCodecRegistrarExt::linked_from_with`] / +/// [`LinkCodecRegistrarExt::linked_to_with`] with a custom [`LinkCodec`] or the +/// feature-gated `link_codecs::Json` and `link_codecs::Postcard` markers. /// /// # Example /// diff --git a/aimdb-data-contracts/src/link_codec.rs b/aimdb-data-contracts/src/link_codec.rs new file mode 100644 index 0000000..eb33d2f --- /dev/null +++ b/aimdb-data-contracts/src/link_codec.rs @@ -0,0 +1,604 @@ +//! Per-link codec selection for [`Linkable`](crate::Linkable) records. +//! +//! Codecs are selected while a link is registered and captured by AimDB's +//! existing typed serializer/deserializer closures. There is no runtime codec +//! registry or per-message lookup. +//! +//! ```rust,ignore +//! use aimdb_data_contracts::{ +//! link_codecs::{Json, Postcard}, +//! LinkCodecBuilderExt, LinkCodecRegistrarExt, +//! }; +//! +//! registrar.linked_to_with("serial://mcu/reading", Postcard::<128>); +//! registrar.linked_to_with("mqtt://cloud/reading", Json); +//! registrar +//! .link_to("mqtt://cloud/alerts") +//! .with_link_codec(Json) +//! .with_config("qos", "1") +//! .finish(); +//! ``` + +use alloc::string::String; +#[cfg(any(feature = "linkable-json", feature = "linkable-postcard"))] +use alloc::string::ToString; +use alloc::vec::Vec; +use core::fmt::Debug; + +use aimdb_core::connector::SerializeError; +use aimdb_core::typed_api::{InboundConnectorBuilder, OutboundConnectorBuilder}; + +/// A wire codec selected for one inbound or outbound record link. +/// +/// The owned [`encode`](Self::encode) operation is always required so an +/// outbound link can handle values larger than its preferred scratch buffer. +/// Set [`ENCODE_BUFFER_CAPACITY`](Self::ENCODE_BUFFER_CAPACITY) to `Some(n)` +/// only when [`encode_into`](Self::encode_into) has a bounded, +/// allocation-free success path. AimDB allocates that scratch storage once per +/// outbound route and reuses it serially. +/// +/// Transport framing is deliberately outside this trait. [`decode`](Self::decode) +/// receives exactly one payload after the connector has applied its own frame +/// limits and partial-read handling. +pub trait LinkCodec: Clone + Send + Sync + 'static { + /// Preferred reusable outbound scratch capacity, in bytes. + /// + /// `None` keeps the route on owned serialization. `Some(n)` installs the + /// scratch serializer alongside the mandatory owned fallback. + const ENCODE_BUFFER_CAPACITY: Option = None; + + /// Decode one connector payload into a record. + fn decode(&self, bytes: &[u8]) -> Result; + + /// Encode a record into owned bytes. + fn encode(&self, value: &T) -> Result, SerializeError>; + + /// Encode a record into caller-owned storage. + /// + /// A codec that advertises a scratch capacity must not allocate on its + /// steady-state success path. Return [`SerializeError::BufferTooSmall`] if + /// the provided storage cannot hold the value; AimDB then invokes + /// [`encode`](Self::encode) once for that value. + fn encode_into(&self, value: &T, out: &mut [u8]) -> Result; +} + +/// Built-in per-link codec markers. +pub mod link_codecs { + /// Delegates to the record's existing [`Linkable`](crate::Linkable) + /// implementation. + #[derive(Clone, Copy, Debug, Default)] + pub struct Default; + + /// JSON via `serde_json`. + /// + /// JSON has no fixed encoded-size bound, so outbound links use the owned + /// serializer rather than advertising a reusable scratch capacity. + #[cfg(feature = "linkable-json")] + #[derive(Clone, Copy, Debug, Default)] + pub struct Json; + + /// Postcard with a route-local reusable scratch capacity. + /// + /// Keep the number of distinct capacity values small on code-size-sensitive + /// targets because each `N` is a separate monomorphization. + #[cfg(feature = "linkable-postcard")] + #[derive(Clone, Copy, Debug, Default)] + pub struct Postcard; +} + +impl LinkCodec for link_codecs::Default +where + T: crate::Linkable, +{ + const ENCODE_BUFFER_CAPACITY: Option = T::ENCODE_BUFFER_CAPACITY; + + fn decode(&self, bytes: &[u8]) -> Result { + T::from_bytes(bytes) + } + + fn encode(&self, value: &T) -> Result, SerializeError> { + value.to_bytes().map_err(|_| SerializeError::InvalidData) + } + + fn encode_into(&self, value: &T, out: &mut [u8]) -> Result { + value.encode_into(out) + } +} + +#[cfg(feature = "linkable-json")] +impl LinkCodec for link_codecs::Json +where + T: serde::Serialize + serde::de::DeserializeOwned, +{ + fn decode(&self, bytes: &[u8]) -> Result { + serde_json::from_slice(bytes).map_err(|error| error.to_string()) + } + + fn encode(&self, value: &T) -> Result, SerializeError> { + serde_json::to_vec(value).map_err(|_| SerializeError::InvalidData) + } + + fn encode_into(&self, value: &T, out: &mut [u8]) -> Result { + let encoded = self.encode(value)?; + let destination = out + .get_mut(..encoded.len()) + .ok_or(SerializeError::BufferTooSmall)?; + destination.copy_from_slice(&encoded); + Ok(encoded.len()) + } +} + +#[cfg(feature = "linkable-postcard")] +impl LinkCodec for link_codecs::Postcard +where + T: serde::Serialize + serde::de::DeserializeOwned, +{ + const ENCODE_BUFFER_CAPACITY: Option = Some(N); + + fn decode(&self, bytes: &[u8]) -> Result { + postcard::from_bytes(bytes).map_err(|error| error.to_string()) + } + + fn encode(&self, value: &T) -> Result, SerializeError> { + postcard::to_allocvec(value).map_err(|_| SerializeError::InvalidData) + } + + fn encode_into(&self, value: &T, out: &mut [u8]) -> Result { + match postcard::to_slice(value, out) { + Ok(used) => Ok(used.len()), + Err(postcard::Error::SerializeBufferFull) => Err(SerializeError::BufferTooSmall), + Err(_) => Err(SerializeError::InvalidData), + } + } +} + +/// Selects a codec on an individual typed connector builder. +/// +/// The builder type is preserved, so connector-specific extension methods can +/// be called before or after `with_link_codec`. +pub trait LinkCodecBuilderExt: Sized +where + T: Send + Sync + Clone + Debug + 'static, +{ + /// Install this codec on the current inbound or outbound link. + fn with_link_codec(self, codec: C) -> Self + where + C: LinkCodec; +} + +impl<'r, 'a, T> LinkCodecBuilderExt for OutboundConnectorBuilder<'r, 'a, T> +where + T: Send + Sync + Clone + Debug + 'static, +{ + fn with_link_codec(self, codec: C) -> Self + where + C: LinkCodec, + { + let scratch_codec = codec.clone(); + let builder = self.with_serializer(move |_ctx, value| codec.encode(value)); + + match C::ENCODE_BUFFER_CAPACITY { + Some(capacity) => builder.with_serializer_into(capacity, move |_ctx, value, out| { + scratch_codec.encode_into(value, out) + }), + None => builder, + } + } +} + +impl<'r, 'a, T> LinkCodecBuilderExt for InboundConnectorBuilder<'r, 'a, T> +where + T: Send + Sync + Clone + Debug + 'static, +{ + fn with_link_codec(self, codec: C) -> Self + where + C: LinkCodec, + { + self.with_deserializer(move |_ctx, bytes| codec.decode(bytes)) + } +} + +#[cfg(test)] +mod tests { + #[cfg(all(feature = "linkable-json", feature = "linkable-postcard"))] + extern crate std; + + #[cfg(all(feature = "linkable-json", feature = "linkable-postcard"))] + use alloc::boxed::Box; + use alloc::string::{String, ToString}; + #[cfg(all(feature = "linkable-json", feature = "linkable-postcard"))] + use alloc::sync::Arc; + use alloc::vec; + use alloc::vec::Vec; + #[cfg(all(feature = "linkable-json", feature = "linkable-postcard"))] + use core::future::Future; + #[cfg(all(feature = "linkable-json", feature = "linkable-postcard"))] + use core::pin::Pin; + #[cfg(all(feature = "linkable-json", feature = "linkable-postcard"))] + use core::task::{Context, Poll}; + + #[cfg(all(feature = "linkable-json", feature = "linkable-postcard"))] + use aimdb_core::buffer::{BufferReader, DynBuffer}; + use aimdb_core::connector::SerializeError; + #[cfg(all(feature = "linkable-json", feature = "linkable-postcard"))] + use aimdb_core::connector::{ConnectorBuilder, SerializedPayload}; + #[cfg(all(feature = "linkable-json", feature = "linkable-postcard"))] + use aimdb_core::executor::test_support::NoopRuntimeOps; + #[cfg(all(feature = "linkable-json", feature = "linkable-postcard"))] + use aimdb_core::{AimDb, AimDbBuilder, BoxFuture, DbError, DbResult}; + use serde::{Deserialize, Serialize}; + + #[cfg(all(feature = "linkable-json", feature = "linkable-postcard"))] + use super::LinkCodecBuilderExt; + use super::{link_codecs, LinkCodec}; + #[cfg(all(feature = "linkable-json", feature = "linkable-postcard"))] + use crate::LinkCodecRegistrarExt; + use crate::{Linkable, SchemaType}; + + #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] + struct Reading { + value: f32, + sequence: u32, + } + + impl SchemaType for Reading { + const NAME: &'static str = "per_link_codec_reading"; + } + + impl Linkable for Reading { + fn from_bytes(data: &[u8]) -> Result { + serde_json::from_slice(data).map_err(|error| error.to_string()) + } + + fn to_bytes(&self) -> Result, String> { + serde_json::to_vec(self).map_err(|error| error.to_string()) + } + } + + #[test] + fn default_codec_delegates_to_linkable() { + let reading = Reading { + value: 23.5, + sequence: 7, + }; + let codec = link_codecs::Default; + let encoded = codec.encode(&reading).expect("default encode"); + + assert_eq!(encoded, reading.to_bytes().expect("Linkable encode")); + let decoded: Reading = codec.decode(&encoded).expect("default decode"); + assert_eq!(decoded, reading); + + let mut out = vec![0xA5; encoded.len() + 4]; + let written = codec + .encode_into(&reading, &mut out) + .expect("default encode_into"); + assert_eq!(&out[..written], encoded.as_slice()); + assert_eq!(&out[written..], &[0xA5; 4]); + } + + #[derive(Clone, Copy)] + struct OffsetCodec { + offset: u32, + } + + impl LinkCodec for OffsetCodec { + const ENCODE_BUFFER_CAPACITY: Option = Some(4); + + fn decode(&self, bytes: &[u8]) -> Result { + let encoded: [u8; 4] = bytes + .try_into() + .map_err(|_| "offset codec expects four bytes".to_string())?; + Ok(Reading { + value: 0.0, + sequence: u32::from_le_bytes(encoded).wrapping_sub(self.offset), + }) + } + + fn encode(&self, value: &Reading) -> Result, SerializeError> { + Ok(value + .sequence + .wrapping_add(self.offset) + .to_le_bytes() + .to_vec()) + } + + fn encode_into(&self, value: &Reading, out: &mut [u8]) -> Result { + let encoded = value.sequence.wrapping_add(self.offset).to_le_bytes(); + out.get_mut(..encoded.len()) + .ok_or(SerializeError::BufferTooSmall)? + .copy_from_slice(&encoded); + Ok(encoded.len()) + } + } + + #[test] + fn stateful_codec_configuration_is_used() { + let reading = Reading { + value: 0.0, + sequence: 10, + }; + let codec = OffsetCodec { offset: 32 }; + let mut out = [0_u8; 4]; + + let written = codec + .encode_into(&reading, &mut out) + .expect("bounded encode"); + + assert_eq!(written, 4); + assert_eq!(u32::from_le_bytes(out), 42); + assert_eq!(codec.decode(&out).expect("decode").sequence, 10); + } + + #[cfg(feature = "linkable-json")] + #[test] + fn json_codec_round_trips_and_rejects_malformed_input() { + let reading = Reading { + value: 21.25, + sequence: 8, + }; + let codec = link_codecs::Json; + let encoded = codec.encode(&reading).expect("JSON encode"); + + let decoded: Reading = codec.decode(&encoded).expect("JSON decode"); + let malformed: Result = codec.decode(b"{"); + assert_eq!(decoded, reading); + assert!(malformed.is_err()); + assert_eq!( + >::ENCODE_BUFFER_CAPACITY, + None + ); + } + + #[cfg(feature = "linkable-postcard")] + #[test] + fn postcard_codec_handles_exact_and_undersized_buffers() { + let reading = Reading { + value: 19.75, + sequence: 9, + }; + let codec = link_codecs::Postcard::<64>; + let owned = codec.encode(&reading).expect("Postcard encode"); + let mut exact = vec![0_u8; owned.len()]; + + let written = codec + .encode_into(&reading, &mut exact) + .expect("exact Postcard buffer"); + + assert_eq!(written, owned.len()); + assert_eq!(exact, owned); + let decoded: Reading = codec.decode(&exact).expect("Postcard decode"); + assert_eq!(decoded, reading); + + let mut small = vec![0_u8; owned.len() - 1]; + assert_eq!( + codec.encode_into(&reading, &mut small), + Err(SerializeError::BufferTooSmall) + ); + let malformed: Result = codec.decode(&[]); + assert!(malformed.is_err()); + } + + /// A fresh reader replays one value. Separate route subscriptions receive + /// independent copies, matching a fan-out buffer's consumer semantics. + #[cfg(all(feature = "linkable-json", feature = "linkable-postcard"))] + struct CannedBuffer { + value: Reading, + } + + #[cfg(all(feature = "linkable-json", feature = "linkable-postcard"))] + impl DynBuffer for CannedBuffer { + fn push(&self, _value: Reading) {} + + fn subscribe_boxed(&self) -> Box + Send> { + Box::new(OneShotReader { + value: Some(self.value.clone()), + }) + } + + fn as_any(&self) -> &dyn core::any::Any { + self + } + } + + #[cfg(all(feature = "linkable-json", feature = "linkable-postcard"))] + struct OneShotReader { + value: Option, + } + + #[cfg(all(feature = "linkable-json", feature = "linkable-postcard"))] + impl OneShotReader { + fn receive(&mut self) -> Result { + self.value.take().ok_or_else(|| DbError::BufferClosed { + buffer_name: "per-link-codec-test".to_string(), + }) + } + } + + #[cfg(all(feature = "linkable-json", feature = "linkable-postcard"))] + impl BufferReader for OneShotReader { + fn poll_recv(&mut self, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(self.receive()) + } + + fn try_recv(&mut self) -> Result { + self.receive() + } + } + + #[cfg(all(feature = "linkable-json", feature = "linkable-postcard"))] + struct CapturingBuffer { + latest: Arc>>, + } + + #[cfg(all(feature = "linkable-json", feature = "linkable-postcard"))] + impl DynBuffer for CapturingBuffer { + fn push(&self, value: Reading) { + *self.latest.lock().expect("capture lock poisoned") = Some(value); + } + + fn subscribe_boxed(&self) -> Box + Send> { + Box::new(OneShotReader { value: None }) + } + + fn as_any(&self) -> &dyn core::any::Any { + self + } + } + + #[cfg(all(feature = "linkable-json", feature = "linkable-postcard"))] + struct NoopConnector; + + #[cfg(all(feature = "linkable-json", feature = "linkable-postcard"))] + impl ConnectorBuilder for NoopConnector { + fn build<'a>( + &'a self, + _db: &'a AimDb, + ) -> Pin>> + Send + 'a>> { + Box::pin(async { Ok(Vec::new()) }) + } + + fn scheme(&self) -> &str { + "test" + } + } + + #[cfg(all(feature = "linkable-json", feature = "linkable-postcard"))] + #[test] + fn one_record_type_uses_route_local_json_and_postcard_codecs() { + futures::executor::block_on(async { + let reading = Reading { + value: 24.5, + sequence: 11, + }; + let json_in = Arc::new(std::sync::Mutex::new(None)); + let postcard_in = Arc::new(std::sync::Mutex::new(None)); + let mut builder = AimDbBuilder::new() + .runtime(Arc::new(NoopRuntimeOps)) + .with_connector(NoopConnector); + + builder.configure::("reading.out", |registrar| { + registrar.buffer_raw(Box::new(CannedBuffer { + value: reading.clone(), + })); + registrar.linked_to_with("test://json", link_codecs::Json); + registrar + .link_to("test://postcard") + .with_link_codec(link_codecs::Postcard::<64>) + .with_config("wire", "binary") + .finish(); + registrar + .linked_to_with("test://postcard-owned-fallback", link_codecs::Postcard::<1>); + }); + + let json_capture = json_in.clone(); + builder.configure::("reading.json.in", move |registrar| { + registrar.buffer_raw(Box::new(CapturingBuffer { + latest: json_capture, + })); + registrar.linked_from_with("test://json-in", link_codecs::Json); + }); + + let postcard_capture = postcard_in.clone(); + builder.configure::("reading.postcard.in", move |registrar| { + registrar.buffer_raw(Box::new(CapturingBuffer { + latest: postcard_capture, + })); + registrar + .link_from("test://postcard-in") + .with_link_codec(link_codecs::Postcard::<64>) + .with_config("wire", "binary") + .finish(); + }); + + let (db, _runner) = builder.build().await.expect("codec routes build"); + let routes = db.collect_outbound_routes("test"); + assert_eq!(routes.len(), 3); + + let json_route = routes + .iter() + .find(|route| route.topic == "json") + .expect("JSON route"); + assert_eq!(json_route.source.serializer_scratch_capacity(), None); + let mut json_reader = json_route.source.subscribe(); + let mut unused_scratch = []; + let json_message = json_reader + .recv_into(&db.runtime_ctx(), &mut unused_scratch) + .await + .expect("JSON payload"); + let SerializedPayload::Owned(json_bytes) = json_message.payload else { + panic!("JSON must use owned serialization"); + }; + let decoded_json: Reading = link_codecs::Json.decode(&json_bytes).expect("JSON decode"); + assert_eq!(decoded_json, reading); + + let postcard_route = routes + .iter() + .find(|route| route.topic == "postcard") + .expect("Postcard route"); + assert_eq!( + postcard_route.source.serializer_scratch_capacity(), + Some(64) + ); + assert!(postcard_route + .config + .contains(&("wire".to_string(), "binary".to_string()))); + let mut postcard_reader = postcard_route.source.subscribe(); + let mut scratch = [0_u8; 64]; + let postcard_message = postcard_reader + .recv_into(&db.runtime_ctx(), &mut scratch) + .await + .expect("Postcard payload"); + let SerializedPayload::Scratch { len } = postcard_message.payload else { + panic!("Postcard must use route scratch storage"); + }; + let decoded_postcard: Reading = link_codecs::Postcard::<64> + .decode(&scratch[..len]) + .expect("Postcard decode"); + assert_eq!(decoded_postcard, reading); + assert_ne!(json_bytes, scratch[..len]); + + let fallback_route = routes + .iter() + .find(|route| route.topic == "postcard-owned-fallback") + .expect("undersized Postcard route"); + assert_eq!(fallback_route.source.serializer_scratch_capacity(), Some(1)); + let mut fallback_reader = fallback_route.source.subscribe(); + let mut undersized_scratch = [0_u8; 1]; + let fallback_message = fallback_reader + .recv_into(&db.runtime_ctx(), &mut undersized_scratch) + .await + .expect("owned Postcard fallback"); + let SerializedPayload::Owned(fallback_bytes) = fallback_message.payload else { + panic!("undersized Postcard scratch must use owned fallback"); + }; + let decoded_fallback: Reading = link_codecs::Postcard::<1> + .decode(&fallback_bytes) + .expect("fallback Postcard decode"); + assert_eq!(decoded_fallback, reading); + + let inbound = db.collect_inbound_routes("test"); + assert_eq!(inbound.len(), 2); + let json_ingest = inbound + .iter() + .find(|(topic, _)| topic == "json-in") + .map(|(_, ingest)| ingest) + .expect("JSON ingest route"); + json_ingest(&db.runtime_ctx(), &json_bytes).expect("JSON ingest"); + assert_eq!( + json_in.lock().expect("JSON capture lock").as_ref(), + Some(&reading) + ); + + let postcard_ingest = inbound + .iter() + .find(|(topic, _)| topic == "postcard-in") + .map(|(_, ingest)| ingest) + .expect("Postcard ingest route"); + postcard_ingest(&db.runtime_ctx(), &scratch[..len]).expect("Postcard ingest"); + assert_eq!( + postcard_in.lock().expect("Postcard capture lock").as_ref(), + Some(&reading) + ); + }); + } +} diff --git a/aimdb-data-contracts/src/linkable.rs b/aimdb-data-contracts/src/linkable.rs index e5db2c6..a7e4e45 100644 --- a/aimdb-data-contracts/src/linkable.rs +++ b/aimdb-data-contracts/src/linkable.rs @@ -1,15 +1,16 @@ //! Linkable registrar extension: one-line link verbs for connector wiring. //! -//! Implementing [`Linkable`](crate::Linkable) unlocks two verbs — +//! Implementing [`Linkable`](crate::Linkable) unlocks default-codec verbs — //! [`LinkableRegistrarExt::linked_from`] / [`linked_to`](LinkableRegistrarExt::linked_to) //! — that install the raw `.link_from()`/`.link_to()` builders with the codec //! defaulted to `T::from_bytes`/`T::to_bytes`. The raw builders remain the -//! escape hatch for per-link options (QoS, topic providers/resolvers). +//! escape hatch for per-link options (QoS, topic providers/resolvers), while +//! [`LinkCodecRegistrarExt`] selects a different codec for an individual link. use aimdb_core::connector::SerializeError; use aimdb_core::typed_api::RecordRegistrar; -use crate::Linkable; +use crate::{LinkCodec, LinkCodecBuilderExt, Linkable}; /// Adds `.linked_from(url)` and `.linked_to(url)` to [`RecordRegistrar`] for /// [`Linkable`] types. @@ -53,6 +54,45 @@ where } } +/// Adds per-link codec variants of [`LinkableRegistrarExt`]'s shorthand verbs. +/// +/// This is a sibling trait rather than new required methods on +/// `LinkableRegistrarExt`, preserving source compatibility for downstream +/// implementations of the existing public extension trait. +pub trait LinkCodecRegistrarExt<'a, T> +where + T: Linkable + Send + Sync + Clone + core::fmt::Debug + 'static, +{ + /// `.link_from(url).with_link_codec(codec).finish()`. + fn linked_from_with(&mut self, url: &str, codec: C) -> &mut RecordRegistrar<'a, T> + where + C: LinkCodec; + + /// `.link_to(url).with_link_codec(codec).finish()`. + fn linked_to_with(&mut self, url: &str, codec: C) -> &mut RecordRegistrar<'a, T> + where + C: LinkCodec; +} + +impl<'a, T> LinkCodecRegistrarExt<'a, T> for RecordRegistrar<'a, T> +where + T: Linkable + Send + Sync + Clone + core::fmt::Debug + 'static, +{ + fn linked_from_with(&mut self, url: &str, codec: C) -> &mut RecordRegistrar<'a, T> + where + C: LinkCodec, + { + self.link_from(url).with_link_codec(codec).finish() + } + + fn linked_to_with(&mut self, url: &str, codec: C) -> &mut RecordRegistrar<'a, T> + where + C: LinkCodec, + { + self.link_to(url).with_link_codec(codec).finish() + } +} + #[cfg(test)] mod tests { use alloc::string::{String, ToString}; diff --git a/aimdb-mqtt-connector/Cargo.toml b/aimdb-mqtt-connector/Cargo.toml index 6399170..3865e5f 100644 --- a/aimdb-mqtt-connector/Cargo.toml +++ b/aimdb-mqtt-connector/Cargo.toml @@ -116,6 +116,10 @@ defmt = { workspace = true, optional = true } [dev-dependencies] tokio = { workspace = true, features = ["full"] } tokio-test = "0.4" +serde = { workspace = true } +aimdb-data-contracts = { path = "../aimdb-data-contracts", default-features = false, features = [ + "linkable-postcard", +] } aimdb-tokio-adapter = { path = "../aimdb-tokio-adapter", features = [ "tokio-runtime", ] } diff --git a/aimdb-mqtt-connector/tests/link_ext_tests.rs b/aimdb-mqtt-connector/tests/link_ext_tests.rs index f804af0..60a78ce 100644 --- a/aimdb-mqtt-connector/tests/link_ext_tests.rs +++ b/aimdb-mqtt-connector/tests/link_ext_tests.rs @@ -8,11 +8,13 @@ use aimdb_core::buffer::BufferCfg; use aimdb_core::AimDbBuilder; +use aimdb_data_contracts::{link_codecs, LinkCodec, LinkCodecBuilderExt}; use aimdb_mqtt_connector::{MqttConnector, MqttLinkExt, MqttOutboundLinkExt}; use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; +use serde::{Deserialize, Serialize}; use std::sync::Arc; -#[derive(Clone, Debug)] +#[derive(Clone, Debug, Deserialize, Serialize)] struct Reading { #[allow(dead_code)] value: f32, @@ -76,3 +78,52 @@ async fn inbound_qos_pushes_exact_config_key() { "expected (qos, 0) in {config:?}" ); } + +#[tokio::test] +async fn per_link_codec_preserves_mqtt_extensions_and_wiring() { + let runtime = Arc::new(TokioAdapter::new().unwrap()); + let mut builder = AimDbBuilder::new().runtime(runtime).with_connector( + MqttConnector::new("mqtt://localhost:1883").with_client_id("link-ext-codec"), + ); + + builder.configure::("test.reading.codec", |reg| { + reg.buffer(BufferCfg::SingleLatest); + + reg.link_to("mqtt://sensors/codec") + .with_link_codec(link_codecs::Postcard::<64>) + .with_qos(2) + .with_retain(true) + .finish(); + + reg.link_from("mqtt://commands/codec") + .with_link_codec(link_codecs::Postcard::<64>) + .with_qos(0) + .finish(); + }); + + let (db, _runner) = builder.build().await.expect("build must succeed"); + + let outbound = db.collect_outbound_routes("mqtt"); + assert_eq!(outbound.len(), 1); + assert_eq!(outbound[0].topic, "sensors/codec"); + assert_eq!(outbound[0].source.serializer_scratch_capacity(), Some(64)); + assert!(outbound[0] + .config + .contains(&("qos".to_string(), "2".to_string()))); + assert!(outbound[0] + .config + .contains(&("retain".to_string(), "true".to_string()))); + + let id = db.inner().resolve_str("test.reading.codec").unwrap(); + let record = db.inner().storage(id).unwrap(); + let inbound_config = &record.inbound_connectors()[0].config; + assert!(inbound_config.contains(&("qos".to_string(), "0".to_string()))); + + let inbound = db.collect_inbound_routes("mqtt"); + assert_eq!(inbound.len(), 1); + assert_eq!(inbound[0].0, "commands/codec"); + let encoded = link_codecs::Postcard::<64> + .encode(&Reading { value: 17.5 }) + .expect("Postcard encode must succeed"); + inbound[0].1(&db.runtime_ctx(), &encoded).expect("Postcard ingest must succeed"); +} diff --git a/docs/design/041-data-contracts-integration.md b/docs/design/041-data-contracts-integration.md index a76b04b..38d1ef7 100644 --- a/docs/design/041-data-contracts-integration.md +++ b/docs/design/041-data-contracts-integration.md @@ -275,6 +275,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`. 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. +- **Per-link formats are additive:** Design 045 adds `with_link_codec(codec)` and `.linked_*_with(url, codec)` without changing these default verbs. Codec selection is captured in the existing fused route closures rather than stored in a runtime registry. - **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. - **Three serialization stories, stated once** so users stop guessing: @@ -378,6 +379,7 @@ Consumed since design 039, works no_std. Stretch (recorded, not in scope): auto- ``` aimdb-data-contracts features: linkable (+core), linkable-json (+serde_json, derive), + linkable-postcard (+postcard), observable (+core), migratable, settable, simulatable (+core, rand nf — no std_rng) # dev tier — never a default feature (no new crate: aimdb-simulation split considered and rejected, §3.1.1) diff --git a/docs/design/045-per-link-codec-selection.md b/docs/design/045-per-link-codec-selection.md new file mode 100644 index 0000000..7c40a85 --- /dev/null +++ b/docs/design/045-per-link-codec-selection.md @@ -0,0 +1,330 @@ +# Design 045: Per-link record codec selection + +**Status:** implemented + +**Issue:** [#178](https://github.com/aimdb-dev/aimdb/issues/178) + +**Prerequisites:** issue #177 and PR #180 + +## 1. Problem + +`Linkable` defines one default wire representation for a record type. That is a +useful 80% path, but a single type can cross heterogeneous boundaries: + +```text +Temperature + +-- UART to MCU ------> compact Postcard + +-- MQTT to cloud ----> inspectable JSON +``` + +Before this design, selecting those formats independently meant repeating raw +serializer/deserializer closures at every registration site. The closures were +powerful, but they discarded the `Linkable` registrar ergonomics and made it +easy for applications to implement subtly different error and buffer behavior. + +The goal is per-link selection without turning codec choice into runtime state +or weakening the bounded serialization path added by PR #180. + +## 2. Decision + +`aimdb-data-contracts` owns three additive pieces: + +1. `LinkCodec`: a format-neutral codec value captured at registration. +2. `LinkCodecBuilderExt::with_link_codec(codec)`: works on inbound and + outbound core link builders while preserving their concrete types. +3. `LinkCodecRegistrarExt::{linked_from_with, linked_to_with}`: shorthand for + the common registrar path. + +Built-in markers live in `link_codecs`: + +- `Default` delegates to the record's existing `Linkable` implementation; +- `Json` is available with `linkable-json`; +- `Postcard` is available with `linkable-postcard`. + +Example: + +```rust,ignore +use aimdb_data_contracts::{ + link_codecs::{Json, Postcard}, + LinkCodecBuilderExt, + LinkCodecRegistrarExt, +}; + +builder.configure::("temperature", |reg| { + reg.buffer(BufferCfg::SpmcRing { capacity: 32 }); + + reg.linked_to_with("serial://mcu/temperature", Postcard::<128>); + reg.linked_to_with("mqtt://cloud/temperature", Json); + + reg.link_to("mqtt://cloud/alerts") + .with_link_codec(Json) + .with_qos(1) + .finish(); +}); +``` + +The existing `.linked_from(url)` and `.linked_to(url)` methods are unchanged. + +## 3. Codec contract + +Conceptually, the trait is: + +```rust,ignore +pub trait LinkCodec: Clone + Send + Sync + 'static { + const ENCODE_BUFFER_CAPACITY: Option = None; + + fn decode(&self, bytes: &[u8]) -> Result; + fn encode(&self, value: &T) -> Result, SerializeError>; + fn encode_into( + &self, + value: &T, + out: &mut [u8], + ) -> Result; +} +``` + +The owned encoder is always present because it is the correctness fallback for +oversized values. `Some(N)` opts an outbound route into a reusable scratch +buffer. A codec must advertise a capacity only when its successful bounded +encoder does not allocate. + +`encode_into` is required rather than given an allocating default. That makes +the performance promise explicit at every implementation that opts into +scratch storage. + +The existing `SerializeError` remains the correct error layer: + +```text +typed record <---- LinkCodec / SerializeError ----> opaque connector payload +session bytes <------- CodecError ----------------> AimX/session envelope +``` + +No `LinkCodecError` is added. Decode retains `String` to match the existing core +inbound builder and the compatibility decision from issue #177. + +## 4. Registration-time specialization + +The codec is captured once in AimDB's existing typed closures: + +```text +control plane (once) + +RecordRegistrar + | + | with_link_codec(Postcard::<128>) + v +OutboundConnectorBuilder + | + +-- owned closure: codec.encode(&T) + `-- scratch closure: codec.encode_into(&T, &mut [u8]) + capacity = 128 +``` + +After `finish()`, core erases the route in the same place it already erased raw +serializers. No per-message map lookup or codec enum match is introduced. + +Why a generic codec value instead of an enum: + +- downstream crates can add codecs without modifying AimDB; +- a codec may carry immutable configuration; +- zero-sized markers compile down to no stored payload state; +- the only dynamic call remains the route's existing erased serializer call. + +Why no registry: + +```text +rejected hot path + +message -> lock/map -> string or TypeId lookup -> dyn codec -> encode +``` + +That shape would add failure states, synchronization or lookup overhead and a +second erasure boundary to solve a configuration-time problem. + +## 5. Builder composition + +`with_link_codec` consumes and returns the same core builder type. It does not +wrap the builder in a new generic facade. Connector extension traits therefore +remain available: + +```rust,ignore +reg.link_to("mqtt://cloud/readings") + .with_link_codec(Json) + .with_qos(1) + .with_retain(false) + .finish(); +``` + +The registrar shorthand lives in a new sibling trait. Adding required methods +to the existing public `LinkableRegistrarExt` could break a downstream manual +implementation, even though AimDB itself supplies the normal implementation. + +## 6. Outbound memory and timing + +PR #180 already placed scratch ownership at the route pump, where the lifetime +is easiest to prove: + +```text +one outbound route pump + ++---------------- stack/future state ----------------+ +| scratch: Vec = vec![0; N] | +| ptr ------------------------+ | ++--------------------------------|--------------------+ + v +heap [ N initialized bytes ] +``` + +Per message: + +```text +time --------------------------------------------------------------------> + +recv T + | codec.encode_into(&T, &mut scratch) + | mutable borrow ends + v +Scratch { len } + | connector.publish(&scratch[..len]).await + | immutable borrow held across await + v +publish complete -> scratch may be mutated for the next value +``` + +No reference escapes `recv_into`; only `Scratch { len }` does. The pump checks +that `len <= scratch.len()` before borrowing the prefix. + +Overflow remains a performance event, not data loss: + +```text +encode_into + +-- Ok(len) ----------> Scratch { len } + +-- BufferTooSmall ---> codec.encode(&T) -> Owned(Vec) + `-- InvalidData ------> log/skip under existing fused-reader semantics +``` + +The fallback runs once. There is no resize/retry loop. + +## 7. Concrete codec behavior + +### Default + +`link_codecs::Default` forwards all operations and the capacity constant to +`T: Linkable`. It is useful when generic configuration code wants an explicit +codec value, but existing registrar verbs remain the simplest default path. + +### JSON + +JSON uses `serde_json::from_slice` and `serde_json::to_vec`. Its capacity is +`None`: encoded size has no general fixed upper bound and the implementation +does not claim a zero-allocation hot path. `encode_into` is still logically +correct when called directly, but it serializes through owned bytes; builders +do not install it. + +### Postcard + +`Postcard` uses `postcard::to_slice` for the bounded path and +`postcard::to_allocvec` for fallback. `SerializeBufferFull` maps to the existing +unit `SerializeError::BufferTooSmall`. + +`N` is a per-route RAM/performance budget, not a transport maximum. An +application must still configure connector frame limits independently. + +## 8. Concurrency, cache and code-size analysis + +```text +route A task ---- scratch A ---- codec A ---- publish A +route B task ---- scratch B ---- codec B ---- publish B +route C task ---- scratch C ---- codec C ---- publish C +``` + +- Each route owns its scratch allocation; there is no mutable aliasing. +- No lock, atomic, global pool or shared mutable codec state is added. +- Logical ownership does not force cache-line ping-pong. Allocator placement + may put heap blocks next to each other, but routes never intentionally write + the same bytes. +- Baseline scratch RAM is the sum of active bounded-route capacities. For 32 + `Postcard<256>` routes, payload storage is about 8 KiB plus `Vec` headers and + allocator metadata. +- Each distinct `Postcard` is a monomorphization. Embedded applications + should standardize on a small set of capacity classes to avoid unnecessary + flash growth. +- Codec selection adds no syscall. Transport enqueueing, network syscalls, + protocol copies and scheduler wakeups remain outside the codec seam and + usually dominate end-to-end latency. + +## 9. Framing and adversarial input boundary + +A codec receives one connector payload. It does not own: + +- TCP/serial partial-read assembly; +- length-prefix or delimiter parsing; +- maximum-frame enforcement; +- reconnect/backpressure policy. + +Those checks must happen before decode so an attacker cannot force an unbounded +allocation merely by declaring a huge frame. Codec implementations then treat +malformed payload content as `Err`, never as a panic. + +## 10. Feature layout + +```text +linkable + +-- alloc + `-- aimdb-core/alloc + +linkable-json + +-- linkable + +-- serde_json + `-- aimdb-derive + +linkable-postcard + +-- linkable + `-- postcard (default features off, alloc fallback enabled) +``` + +The Makefile exercises host, no_std and Cortex-M combinations explicitly so a +new codec feature cannot silently work only under workspace feature unification. + +## 11. Validation + +The implementation contains: + +- direct default, stateful custom, JSON and Postcard codec tests; +- malformed-input tests; +- exact and undersized Postcard buffer tests; +- a route-level test that gives the same `Reading` type JSON and Postcard + outbound links, asserting `Owned` versus `Scratch` payload ownership; +- a deliberately undersized Postcard route that exercises the owned fallback; +- route-level inbound JSON and Postcard ingestion; +- builder composition checks using config methods after codec selection; +- an MQTT integration test that compiles and builds outbound + `.with_link_codec(...).with_qos(...).with_retain(...)` and inbound + `.with_link_codec(...).with_qos(...)` chains, then checks the installed + scratch capacity, MQTT options and Postcard ingest path; +- a warmed allocation benchmark comparing direct `Linkable::encode_into` with + `Postcard::encode_into`. + +At implementation time, the B0 benchmark measured 10,000 iterations for each +path and reported zero allocation calls and zero allocated bytes in both +windows. This is a codec-seam claim, not a claim that a complete connector +publish performs no allocation. After capturing both results, a positive +control deliberately allocates a `Vec` with at least 64 bytes of capacity +and asserts that both counters move. Running it after the measured windows +proves the counting allocator is wired without contaminating either zero. + +The final `make check` passed the host test/clippy matrix, Cortex-M and WASM +checks, dependency policy, README and code-generation drift checks, the +generated Postcard round-trip, generated Cortex-M cross-compilation and the +production-graph simulation guard. That run included the MQTT hardening case; +the complete MQTT connector suite passed 7 unit, 3 link-extension, 16 +topic-provider and 2 compiled doctests. + +## 12. Deferred work + +- Codec selection in AimX/MCP/CLI and per-connector code generation. +- Sharing encoded bytes among routes that prove identical codec semantics. +- Schema-derived maximum encoded sizes. +- A genuinely bounded JSON writer, if a useful maximum can be proven. +- Connector-level latency benchmarks separating serialization from transport + scheduling and syscall cost. From 30a9e6f4ab4aa4f0e6d1191d0faa3d74fbefd417 Mon Sep 17 00:00:00 2001 From: Amidamaru Date: Thu, 16 Jul 2026 13:17:53 +0700 Subject: [PATCH 2/2] fix(data-contracts): clear stale scratch serializer on codec replacement --- CHANGELOG.md | 4 +- aimdb-core/CHANGELOG.md | 4 ++ aimdb-core/src/typed_api.rs | 11 +++++ aimdb-data-contracts/CHANGELOG.md | 2 +- aimdb-data-contracts/src/link_codec.rs | 47 +++++++++++++++++---- docs/design/045-per-link-codec-selection.md | 9 ++++ 6 files changed, 67 insertions(+), 10 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d7d71e9..825afea 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -38,7 +38,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 `with_link_codec(codec)`. Selection is fused at registration time; bounded Postcard routes retain the reusable scratch/owned-overflow behavior from issue #177 without a per-message registry, lock, or lookup. See - [Design 045](docs/design/045-per-link-codec-selection.md). + [Design 045](docs/design/045-per-link-codec-selection.md). Re-selecting a + codec replaces both the owned and scratch serializers, so changing a route + from bounded postcard to owned JSON cannot retain stale postcard output. ### Changed — Design 038 simplification pass (breaking) diff --git a/aimdb-core/CHANGELOG.md b/aimdb-core/CHANGELOG.md index a60b0fd..5e03b1f 100644 --- a/aimdb-core/CHANGELOG.md +++ b/aimdb-core/CHANGELOG.md @@ -19,6 +19,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 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. +- **Scratch serializer reset hook.** + `OutboundConnectorBuilder::clear_serializer_into` lets higher-level builder + extensions replace a bounded serialization strategy with an owned-only one + without retaining the previous route-local scratch callback. - **`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/typed_api.rs b/aimdb-core/src/typed_api.rs index 4cf0f79..63b801e 100644 --- a/aimdb-core/src/typed_api.rs +++ b/aimdb-core/src/typed_api.rs @@ -830,6 +830,17 @@ where self } + /// Removes a previously installed into-slice serializer. + /// + /// Higher-level builder extensions use this when replacing a bounded + /// serialization strategy with an owned-only one. Clearing the fast path + /// prevents the old scratch serializer from emitting a different wire + /// format beside the replacement owned serializer. + pub fn clear_serializer_into(mut self) -> Self { + self.context_serializer_into = None; + self + } + /// Sets the operation timeout in milliseconds (the connector interprets /// it; passed as the `timeout_ms` option — see /// `ConnectorConfig::from_query`) diff --git a/aimdb-data-contracts/CHANGELOG.md b/aimdb-data-contracts/CHANGELOG.md index 89854d0..5ef6878 100644 --- a/aimdb-data-contracts/CHANGELOG.md +++ b/aimdb-data-contracts/CHANGELOG.md @@ -16,7 +16,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added -- **Issue #178 — per-link codec selection.** `LinkCodec`, `LinkCodecBuilderExt::with_link_codec` and `LinkCodecRegistrarExt::{linked_from_with, linked_to_with}` let the same record type use different formats on different connector routes without a runtime registry. Built-in `link_codecs::Json` and bounded `link_codecs::Postcard` markers are feature-gated by `linkable-json` and the new `linkable-postcard`; Postcard reuses the route-local scratch and owned overflow fallback added for issue #177. Existing `.linked_from` and `.linked_to` behavior is unchanged. +- **Issue #178 — per-link codec selection.** `LinkCodec`, `LinkCodecBuilderExt::with_link_codec` and `LinkCodecRegistrarExt::{linked_from_with, linked_to_with}` let the same record type use different formats on different connector routes without a runtime registry. Built-in `link_codecs::Json` and bounded `link_codecs::Postcard` markers are feature-gated by `linkable-json` and the new `linkable-postcard`; Postcard reuses the route-local scratch and owned overflow fallback added for issue #177. Existing `.linked_from` and `.linked_to` behavior is unchanged. Repeated `with_link_codec` calls use complete last-selection-wins semantics: an owned-only replacement clears a scratch encoder left by an earlier bounded codec. - **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 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`). diff --git a/aimdb-data-contracts/src/link_codec.rs b/aimdb-data-contracts/src/link_codec.rs index eb33d2f..6abd90a 100644 --- a/aimdb-data-contracts/src/link_codec.rs +++ b/aimdb-data-contracts/src/link_codec.rs @@ -161,6 +161,10 @@ where T: Send + Sync + Clone + Debug + 'static, { /// Install this codec on the current inbound or outbound link. + /// + /// Calling this method again replaces the complete codec strategy. In + /// particular, an owned-only codec clears any scratch serializer installed + /// by an earlier bounded codec. fn with_link_codec(self, codec: C) -> Self where C: LinkCodec; @@ -174,14 +178,17 @@ where where C: LinkCodec, { - let scratch_codec = codec.clone(); - let builder = self.with_serializer(move |_ctx, value| codec.encode(value)); - match C::ENCODE_BUFFER_CAPACITY { - Some(capacity) => builder.with_serializer_into(capacity, move |_ctx, value, out| { - scratch_codec.encode_into(value, out) - }), - None => builder, + Some(capacity) => { + let scratch_codec = codec.clone(); + self.with_serializer(move |_ctx, value| codec.encode(value)) + .with_serializer_into(capacity, move |_ctx, value, out| { + scratch_codec.encode_into(value, out) + }) + } + None => self + .with_serializer(move |_ctx, value| codec.encode(value)) + .clear_serializer_into(), } } } @@ -488,6 +495,11 @@ mod tests { .finish(); registrar .linked_to_with("test://postcard-owned-fallback", link_codecs::Postcard::<1>); + registrar + .link_to("test://postcard-replaced-by-json") + .with_link_codec(link_codecs::Postcard::<64>) + .with_link_codec(link_codecs::Json) + .finish(); }); let json_capture = json_in.clone(); @@ -512,7 +524,7 @@ mod tests { let (db, _runner) = builder.build().await.expect("codec routes build"); let routes = db.collect_outbound_routes("test"); - assert_eq!(routes.len(), 3); + assert_eq!(routes.len(), 4); let json_route = routes .iter() @@ -576,6 +588,25 @@ mod tests { .expect("fallback Postcard decode"); assert_eq!(decoded_fallback, reading); + let replacement_route = routes + .iter() + .find(|route| route.topic == "postcard-replaced-by-json") + .expect("replacement route"); + assert_eq!( + replacement_route.source.serializer_scratch_capacity(), + None, + "owned-only replacement must clear the previous scratch codec" + ); + let mut replacement_reader = replacement_route.source.subscribe(); + let replacement_message = replacement_reader + .recv_into(&db.runtime_ctx(), &mut []) + .await + .expect("replacement JSON payload"); + let SerializedPayload::Owned(replacement_bytes) = replacement_message.payload else { + panic!("replacement JSON codec must use owned serialization"); + }; + assert_eq!(replacement_bytes, json_bytes); + let inbound = db.collect_inbound_routes("test"); assert_eq!(inbound.len(), 2); let json_ingest = inbound diff --git a/docs/design/045-per-link-codec-selection.md b/docs/design/045-per-link-codec-selection.md index 7c40a85..006ecd4 100644 --- a/docs/design/045-per-link-codec-selection.md +++ b/docs/design/045-per-link-codec-selection.md @@ -155,6 +155,13 @@ reg.link_to("mqtt://cloud/readings") .finish(); ``` +Repeated selection replaces the entire serialization strategy, not only its +owned half. A bounded codec installs owned and scratch callbacks; replacing it +with an owned-only codec calls +`OutboundConnectorBuilder::clear_serializer_into`. Without that reset, +`Postcard -> Json` would leave the Postcard scratch callback active for +fitting values and use JSON only on fallback, mixing formats on one route. + The registrar shorthand lives in a new sibling trait. Adding required methods to the existing public `LinkableRegistrarExt` could break a downstream manual implementation, even though AimDB itself supplies the normal implementation. @@ -297,6 +304,8 @@ The implementation contains: outbound links, asserting `Owned` versus `Scratch` payload ownership; - a deliberately undersized Postcard route that exercises the owned fallback; - route-level inbound JSON and Postcard ingestion; +- a bounded-Postcard-to-owned-JSON replacement regression that asserts the + stale scratch capacity is cleared and the route emits owned JSON; - builder composition checks using config methods after codec selection; - an MQTT integration test that compiles and builds outbound `.with_link_codec(...).with_qos(...).with_retain(...)` and inbound