diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 848ab895..a9823d60 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -75,6 +75,9 @@ jobs: - name: Check codegen template drift run: make codegen-drift + - name: Prove production is simulation-free + run: make check-no-sim + # Embedded target testing embedded-targets: name: Embedded Cross-compilation diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 9314ee3b..a23b2252 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -235,12 +235,7 @@ examples/ # Demo applications ## Next Development Areas -See `.github/copilot-instructions.md` for current implementation status and priorities: - -- **Kafka Connector** - Kafka integration using `rdkafka` -- **DDS Connector** - DDS protocol support -- **CLI Tools** - Introspection, monitoring, debugging commands -- **Performance** - Benchmarks and profiling infrastructure +See the [GitHub issues](https://github.com/aimdb-dev/aimdb/issues) for planned work and open feature requests. ## Getting Help diff --git a/Cargo.lock b/Cargo.lock index a97e7db2..9eb299a9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -97,7 +97,7 @@ dependencies = [ [[package]] name = "aimdb-core" -version = "1.1.0" +version = "1.2.0" dependencies = [ "aimdb-derive", "anyhow", @@ -120,7 +120,7 @@ dependencies = [ [[package]] name = "aimdb-data-contracts" -version = "0.2.0" +version = "0.3.0" dependencies = [ "aimdb-core", "aimdb-derive", @@ -132,7 +132,7 @@ dependencies = [ [[package]] name = "aimdb-derive" -version = "0.2.0" +version = "0.3.0" dependencies = [ "proc-macro2", "quote", @@ -280,9 +280,10 @@ dependencies = [ [[package]] name = "aimdb-sync" -version = "0.5.0" +version = "0.6.0" dependencies = [ "aimdb-core", + "aimdb-data-contracts", "aimdb-tokio-adapter", "serde", "serde_json", @@ -2531,6 +2532,7 @@ name = "mqtt-connector-demo-common" version = "0.1.0" dependencies = [ "aimdb-core", + "aimdb-data-contracts", "defmt 1.0.1", "serde", "serde_json", @@ -3726,6 +3728,7 @@ name = "tokio-mqtt-connector-demo" version = "1.1.0" dependencies = [ "aimdb-core", + "aimdb-data-contracts", "aimdb-mqtt-connector", "aimdb-tokio-adapter", "mqtt-connector-demo-common", @@ -4447,7 +4450,6 @@ dependencies = [ "aimdb-data-contracts", "aimdb-mqtt-connector", "aimdb-tokio-adapter", - "chrono", "rand 0.10.1", "tokio", "tracing", diff --git a/Makefile b/Makefile index 89670759..2999ff7f 100644 --- a/Makefile +++ b/Makefile @@ -1,7 +1,7 @@ # AimDB Makefile # Simple automation for common development tasks -.PHONY: help build test clean clean-embedded fmt fmt-check clippy doc all check test-embedded test-wasm wasm wasm-test examples deny audit security publish publish-check readme-check codegen-drift +.PHONY: help build test clean clean-embedded fmt fmt-check clippy doc all check test-embedded test-wasm wasm wasm-test examples deny audit security publish publish-check readme-check codegen-drift check-no-sim .DEFAULT_GOAL := help # Separate target dir for embedded checks so an interrupted example build @@ -555,8 +555,46 @@ codegen-drift: @printf "$(GREEN)Checking codegen templates against the workspace API...$(NC)\n" ./tools/scripts/codegen-drift-check.sh +# Prove that simulation code (the dev-tier `simulatable` contract) never +# reaches a production binary. `rand` is the tracer: it is reachable iff +# `simulatable` is enabled (aimdb-data-contracts/src/simulatable.rs). +# The guard must not fail open: +# "rand is absent" is accepted only when `cargo tree -i rand` fails with its +# specific "did not match any packages" error — any other failure (missing +# submodule, typo'd package name, registry trouble) aborts the check instead of +# passing it vacuously. A positive control per example asserts the sim build +# DOES find `rand`, proving the tracer still traces. Also asserts `simulatable` +# is not a default feature of the contracts crate (which would pull `rand` into +# its own default graph). +SIM_EXAMPLES := weather-station-beta weather-station-gamma +check-no-sim: + @printf "$(GREEN)Proving production graphs are simulation-free...$(NC)\n" + @tree_rand() { cargo tree -p "$$1" $$2 -e normal -i rand 2>&1; }; \ + assert_rand_free() { \ + if out=$$(tree_rand "$$1" "$$2"); then \ + printf "$(RED)✗ $$3$(NC)\n"; \ + exit 1; \ + elif ! printf '%s\n' "$$out" | grep -q 'did not match any packages'; then \ + printf "$(RED)✗ cargo tree for '$$1' failed for a reason other than 'rand is absent' — refusing to pass vacuously:$(NC)\n"; \ + printf '%s\n' "$$out"; \ + exit 1; \ + fi; \ + }; \ + for bin in $(SIM_EXAMPLES); do \ + assert_rand_free "$$bin" "" "'$$bin' (default/production, no sim) pulls in 'rand' — simulation code leaked into production"; \ + printf "$(BLUE)✓ $$bin production graph is rand-free$(NC)\n"; \ + if ! tree_rand "$$bin" "--features sim" >/dev/null; then \ + printf "$(RED)✗ positive control failed: '$$bin --features sim' does not pull 'rand' — the tracer no longer traces, so the rand-free results above prove nothing$(NC)\n"; \ + exit 1; \ + fi; \ + printf "$(BLUE)✓ $$bin sim graph finds rand (tracer positive control)$(NC)\n"; \ + done; \ + assert_rand_free "aimdb-data-contracts" "" "aimdb-data-contracts pulls 'rand' with default features — 'simulatable' must never be a default feature"; \ + printf "$(BLUE)✓ 'simulatable' is not a default feature of aimdb-data-contracts$(NC)\n"; \ + printf "$(GREEN)✓ Production is simulation-free$(NC)\n" + ## Convenience commands -check: fmt-check clippy test test-embedded test-wasm deny readme-check codegen-drift +check: fmt-check clippy test test-embedded test-wasm deny readme-check codegen-drift check-no-sim @printf "$(GREEN)Comprehensive development checks completed!$(NC)\n" @printf "$(BLUE)✓ Code formatting verified$(NC)\n" @printf "$(BLUE)✓ Linter passed$(NC)\n" diff --git a/README.md b/README.md index 0a2cef9c..f30d8ea0 100644 --- a/README.md +++ b/README.md @@ -39,12 +39,22 @@ AimDB is not a storage engine. It's a typed data plane where the Rust type *is* ## Why AimDB exists -Distributed systems spend most of their complexity budget translating between layers. IDLs, codegen, serialization, schema registries and glue services. AimDB removes that layer by making **the Rust type the contract**: defined once, compiled unchanged from a `no_std` microcontroller to the browser. +Most of a distributed system's complexity is translation: IDLs, codegen, serialization glue, schema registries. AimDB deletes that layer by making **the Rust type the contract** — defined once, compiled unchanged from a `no_std` microcontroller to the browser. - **One type, every tier.** The same struct compiles for firmware and cloud. No conversion layer between them. - **The buffer defines how data moves.** No manual queue wiring, no separate transport config. - **No untyped boundaries.** Capabilities, like streaming, migration, observability and connectors, are unlocked by traits. +### The receipts + +None of this is roadmap. Every claim is backed by code, tests or committed benchmarks: + +- **Data contracts as schema** → the Rust type *is* the wire contract. No IDL, no codegen, [CI cross-compiles](Makefile) unchanged from Cortex-M to WASM. +- **Typed data migrations** → [`migration_chain!`](aimdb-data-contracts/src/migratable.rs) const-validates the chain and works `no_std`, so old and new nodes coexist on the wire. +- **Zero allocation per message** → [allocation baselines](aimdb-bench/data/baselines) across Tokio, Embassy and WASM. +- **Non-blocking producers** → synchronous [`produce()`](aimdb-core/src/typed_api.rs) calls and overwrite semantics on all buffers. +- **Identical buffer contracts across runtimes** → [shared conformance suite](aimdb-core/src/buffer/test_support.rs) validates SPMC Ring, SingleLatest and Mailbox on every adapter. + [The Next Era of Software Architecture Is Data-First](https://aimdb.dev/blog/data-driven-design) --- @@ -93,10 +103,7 @@ async fn main() -> Result<(), Box> { }); }); - // `.run()` builds the database, collects every producer/consumer/transform - // future, and drives them all on a single `FuturesUnordered`. It blocks - // until shutdown. For programmatic access to the `AimDb` handle, call - // `.build().await?` directly — it returns `(AimDb, AimDbRunner)`. + // Build the db and drive every source/tap future until shutdown. builder.run().await?; Ok(()) } @@ -128,20 +135,45 @@ docker compose up | [**SingleLatest**](examples/hello-single-latest-async) | Only the current value matters | Feature flags, config, UI state | | [**Mailbox**](examples/hello-mailbox) / [**async Mailbox**](examples/hello-mailbox-async)| Latest instruction wins | Device commands, actuation, RPC | -**Four capability traits** — opt-in, type-checked: +**One async API across runtimes.** Tokio, Embassy, WASM — swap the runtime adapter, keep the code. → [How the runtime abstraction works](https://aimdb.dev/blog/building-aimdb-one-async-api) -| Trait | What it unlocks | Runtimes | -| --- | --- | --- | -| [`Streamable`](https://aimdb.dev/blog/streamable-crossing-boundaries) | Crossing WASM / WebSocket / CLI boundaries | std, no_std | -| [`Migratable`](https://aimdb.dev/blog/schema-migration-without-ceremony) | Typed schema evolution across deployed fleets | std, no_std | -| `Observable` | Automatic per-record metrics | std, no_std | -| [`Linkable`](https://aimdb.dev/blog/connectors-where-aimdb-meets-the-real-world) | Wire-format connectors | std, no_std | +**Connectors that ship today:** MQTT, KNX, WebSocket, TCP, serial (COBS-framed UART) and Unix domain sockets. Writing your own is one trait impl. → [Connector status](#connectors) -**One async API across runtimes.** Tokio, Embassy, WASM — swap the runtime adapter, keep the code. → [How the runtime abstraction works](https://aimdb.dev/blog/building-aimdb-one-async-api) +**Optional persistence.** The core is an in-memory data plane; [`aimdb-persistence`](aimdb-persistence) adds `.persist()` with a SQLite backend ([`aimdb-persistence-sqlite`](aimdb-persistence-sqlite)) for records whose history must survive restarts. + +Deep dives: [source/tap/transform](https://aimdb.dev/blog/source-tap-transform) · [schema migration](https://aimdb.dev/blog/schema-migration-without-ceremony) · [reactive pipelines](https://aimdb.dev/blog/reactive-pipelines) + +--- -**Connectors that ship today:** MQTT, KNX, WebSocket. Writing your own is one trait impl. +## Data contracts: one method per capability -Deep dives: [data contracts](https://aimdb.dev/blog/data-contracts-deep-dive) · [source/tap/transform](https://aimdb.dev/blog/source-tap-transform) · [schema migration](https://aimdb.dev/blog/schema-migration-without-ceremony) · [reactive pipelines](https://aimdb.dev/blog/reactive-pipelines) +Every capability is an opt-in trait on your schema type: implement it and **exactly one method** appears on the registrar. + +| 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)` (`#[derive(Linkable)]` for JSON) | 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** | + +`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 +builder.configure::(KEY, |reg| { + #[cfg(feature = "sim")] + reg.simulate(profile, rng); + #[cfg(not(feature = "sim"))] + reg.source(read_hardware); +}); +``` + +Build with `sim` off and the simulation code is *gone* — no `T::simulate` impls, no `rand`, nothing to audit. + +**Old and new nodes coexist.** Migration steps are typed and bidirectional and `migration_chain!` checks the whole chain at compile time — [roundtrip tests](aimdb-data-contracts/tests/migration_roundtrip.rs) cover upgrade *and* downgrade across multi-step chains. It's `no_std` too: even an MCU can accept a payload one schema version behind or downgrade its output for an older peer. + +Deep dive: [data contracts](https://aimdb.dev/blog/data-contracts-deep-dive) --- @@ -157,10 +189,10 @@ A record is written by a `Source`, lands in a typed `Buffer` and fans out to in- Source ───► │ Buffer │ (typed) │ SPMC / SL / │ ───► Tap (another subscriber) │ Mailbox │ - └──────────────┘ ───► Link ──► MQTT / KNX / WebSocket + └──────────────┘ ───► Link ──► MQTT / KNX / WS / UDS / serial ``` -The Rust type system enforces correctness at compile time, buffer semantics enforce flow guarantees at runtime and connectors wire to your infrastructure without an integration layer. The same code compiles for MCU, edge, cloud or browser — see [Platform Support](#platform-support) below. +Types check correctness at compile time, buffers enforce flow semantics at runtime and connectors bridge to your infrastructure, with no integration layer in between. The same code compiles for MCU, edge, cloud or browser — see [Platform Support](#platform-support) below. --- @@ -211,6 +243,8 @@ See the [MCP server docs](tools/aimdb-mcp/) for Claude Desktop and other editors | **Kafka** | 📋 Planned | std | | **Modbus** | 📋 Planned | std, no_std | +The serial, TCP and UDS connectors carry both record mirroring and the AimX remote-access protocol used by the CLI and MCP server. Typed records over multiple transports, from bare metal to cloud. + --- ### Platform Support @@ -228,14 +262,8 @@ See the [MCP server docs](tools/aimdb-mcp/) for Claude Desktop and other editors We're a small team building something ambitious. The fastest way to help is to take on a scoped piece of it. Each of these is sized for a few hours and includes file pointers, acceptance criteria and a place to ask questions: -- [#92 — `no_std` `Display` for `DbError` should include numeric fields](https://github.com/aimdb-dev/aimdb/issues/92) · 2–3h · core · embedded - [#93 — Minimal example: `hello-single-latest`](https://github.com/aimdb-dev/aimdb/issues/93) · 2–3h · docs -- [#95 — CLI: add `aimdb instance ping` subcommand](https://github.com/aimdb-dev/aimdb/issues/95) · 3–4h · cli -- [#96 — CI: fail on broken rustdoc links](https://github.com/aimdb-dev/aimdb/issues/96) · 1–2h · docs -- [#97 — Doctests for `BufferCfg` variants](https://github.com/aimdb-dev/aimdb/issues/97) · 2–3h · core · docs -- [#99 — Async example: `hello-mailbox-async`](https://github.com/aimdb-dev/aimdb/issues/99) · 2–3h · docs -- [#100 — Async example: `hello-single-latest-async`](https://github.com/aimdb-dev/aimdb/issues/100) · 2–3h · docs -- [#101 — Async example: `hello-spmc-ring-async`](https://github.com/aimdb-dev/aimdb/issues/101) · 2–3h · docs +- [#101 — Minimal example: `hello-spmc-ring-async`](https://github.com/aimdb-dev/aimdb/issues/101) · 2–3h · docs [See all good first issues →](https://github.com/aimdb-dev/aimdb/labels/good%20first%20issue) diff --git a/aimdb-codegen/src/rust.rs b/aimdb-codegen/src/rust.rs index 00be6772..ffb95ccf 100644 --- a/aimdb-codegen/src/rust.rs +++ b/aimdb-codegen/src/rust.rs @@ -132,11 +132,18 @@ pub fn generate_cargo_toml(state: &ArchitectureState) -> String { .iter() .any(|r| r.serialization.as_ref() == Some(&SerializationType::Postcard)); let has_observable = state.records.iter().any(|r| r.observable.is_some()); + let has_settable = state + .records + .iter() + .any(|r| r.fields.iter().any(|f| f.settable)); let mut data_contracts_features = Vec::new(); if has_non_custom_ser { data_contracts_features.push("\"linkable\""); } + if has_settable { + data_contracts_features.push("\"settable\""); + } let dc_features_str = if data_contracts_features.is_empty() { String::new() @@ -1009,55 +1016,19 @@ fn emit_observable_impl(rec: &RecordDef) -> Option { let signal_type: syn::Type = syn::parse_str(&signal_field.field_type).ok()?; let signal_ident = format_ident!("{}", obs.signal_field); - let icon = &obs.icon; let unit = &obs.unit; - // Timestamp heuristic: first u64 field named timestamp/computed_at/fetched_at - let timestamp_names = ["timestamp", "computed_at", "fetched_at"]; - let timestamp_field = rec - .fields - .iter() - .find(|f| f.field_type == "u64" && timestamp_names.contains(&f.name.as_str())); - - let format_log_body = if let Some(ts) = timestamp_field { - let ts_ident = format_ident!("{}", ts.name); - quote! { - alloc::format!( - "{} [{}] {}: {:.1}{} at {}", - Self::ICON, - node_id, - Self::NAME, - self.signal(), - Self::UNIT, - self.#ts_ident, - ) - } - } else { - quote! { - alloc::format!( - "{} [{}] {}: {:.1}{}", - Self::ICON, - node_id, - Self::NAME, - self.signal(), - Self::UNIT, - ) - } - }; - + // Observable is a kernel-only trait: the numeric projection + UNIT label. + // `SIGNAL` defaults to the schema name; presentation (icons, log formatting) + // is not part of the trait, so nothing else is emitted. Some(quote! { impl Observable for #struct_name { type Signal = #signal_type; - const ICON: &'static str = #icon; const UNIT: &'static str = #unit; fn signal(&self) -> #signal_type { self.#signal_ident } - - fn format_log(&self, node_id: &str) -> alloc::string::String { - #format_log_body - } } }) } diff --git a/aimdb-core/CHANGELOG.md b/aimdb-core/CHANGELOG.md index 1f6e504d..9e503924 100644 --- a/aimdb-core/CHANGELOG.md +++ b/aimdb-core/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- **`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 - **AimX protocol doc rot cleaned up; `remote::PROTOCOL_VERSION` corrected to `"2.0"` and exported.** The `remote` module docs claimed "AimX v1" and linked a spec file that no longer exists; they now describe the v2 NDJSON tagged-frame wire and point at `crate::session::aimx` / `docs/design/remote-access-via-connectors.md`. The AimX dispatch's Welcome uses the constant instead of a hardcoded `"2.0"` (same bytes on the wire). The dead, never-exported v1 `Message` untagged envelope and its helpers were removed from `remote::protocol`. Also de-advertised Kafka/HTTP connector semantics from `ConnectorUrl` docs (the parser is scheme-agnostic; those connectors never existed) and updated the `connector` module docs from the removed `.link()` API to `.link_to()`/`.link_from()`. @@ -452,7 +456,8 @@ warning fires if you exceed 1000 interned keys. --- -[Unreleased]: https://github.com/aimdb-dev/aimdb/compare/v1.1.0...HEAD +[Unreleased]: https://github.com/aimdb-dev/aimdb/compare/v1.2.0...HEAD +[1.2.0]: https://github.com/aimdb-dev/aimdb/compare/v1.1.0...v1.2.0 [1.1.0]: https://github.com/aimdb-dev/aimdb/compare/v1.0.0...v1.1.0 [1.0.0]: https://github.com/aimdb-dev/aimdb/compare/v0.5.0...v1.0.0 [0.5.0]: https://github.com/aimdb-dev/aimdb/compare/v0.4.0...v0.5.0 diff --git a/aimdb-core/Cargo.toml b/aimdb-core/Cargo.toml index 8d5bd2a4..45353a9f 100644 --- a/aimdb-core/Cargo.toml +++ b/aimdb-core/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "aimdb-core" -version = "1.1.0" +version = "1.2.0" edition = "2021" license.workspace = true repository.workspace = true @@ -68,7 +68,7 @@ test-utils = ["std"] [dependencies] # Derive macros (optional) -aimdb-derive = { version = "0.2.0", path = "../aimdb-derive", optional = true } +aimdb-derive = { version = "0.3.0", path = "../aimdb-derive", optional = true } # Stream trait for bidirectional connectors (minimal, no_std compatible) futures-core = { version = "0.3", default-features = false } @@ -116,6 +116,9 @@ spin = { version = "0.9", default-features = false, features = ["mutex", "spin_m hashbrown = { version = "0.15", default-features = false, features = ["default-hasher"] } [dev-dependencies] +# Serde-attribute roundtrip tests (src/profiling/info.rs) must run in every +# feature config, including ones where the optional `serde_json` dep is off. +serde_json = { workspace = true } # For no_std testing heapless = "0.9.1" # For async testing (`sync` covers tokio::sync::mpsc in tests/session_engine.rs, diff --git a/aimdb-core/src/lib.rs b/aimdb-core/src/lib.rs index 846e5285..1074f011 100644 --- a/aimdb-core/src/lib.rs +++ b/aimdb-core/src/lib.rs @@ -39,6 +39,7 @@ pub mod remote; pub mod router; #[cfg(feature = "connector-session")] pub mod session; +pub mod signal; pub mod transform; pub mod transport; pub mod typed_api; @@ -83,9 +84,15 @@ pub use session::{ SessionLimits, Source, TransportError, TransportResult, }; +// Signal gauge handle (always available; inert without `observability`) +pub use signal::SignalGaugeHandle; + // Stage profiling exports (feature-gated) #[cfg(feature = "observability")] -pub use profiling::{RecordProfilingMetrics, StageMetrics, StageProfilingInfo}; +pub use profiling::{ + RecordProfilingMetrics, SignalGauge, SignalStats, SignalStatsInfo, StageMetrics, + StageProfilingInfo, +}; // Connector Infrastructure exports pub use connector::TopicProvider; diff --git a/aimdb-core/src/profiling/info.rs b/aimdb-core/src/profiling/info.rs index 359b2e32..b6a422c6 100644 --- a/aimdb-core/src/profiling/info.rs +++ b/aimdb-core/src/profiling/info.rs @@ -4,7 +4,7 @@ use alloc::{string::String, vec::Vec}; use serde::{Deserialize, Serialize}; -use crate::profiling::{RecordProfilingMetrics, StageEntry}; +use crate::profiling::{RecordProfilingMetrics, SignalGauge, StageEntry}; /// A point-in-time snapshot of one execution stage's timing metrics. /// @@ -31,6 +31,48 @@ pub struct StageProfilingInfo { pub max_time_ns: u64, } +/// A point-in-time snapshot of one record's signal gauge (`.observe()`). +/// +/// Carried in `RecordMetadata::signal_stats` so a record's live domain signal +/// (last/min/max/mean) surfaces on `record.list`/`record.get` and the MCP tools. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct SignalStatsInfo { + /// Signal label (`Observable::SIGNAL`, defaults to the schema name). + pub signal: String, + /// Unit label (`Observable::UNIT`), omitted when empty. + /// + /// `default` matches the skip: `Observable::UNIT` defaults to `""`, so a + /// snapshot without a unit must deserialize (aimdb-client/aimdb-mcp read + /// this back from `record.list`/`record.get`). + #[serde(default, skip_serializing_if = "String::is_empty")] + pub unit: String, + /// Number of samples observed. + pub count: u64, + /// Most recent sample. + pub last: f64, + /// Smallest sample observed. + pub min: f64, + /// Largest sample observed. + pub max: f64, + /// Mean of all samples. + pub mean: f64, +} + +impl SignalStatsInfo { + fn from_gauge(gauge: &SignalGauge) -> Self { + let s = &gauge.stats; + Self { + signal: gauge.name.clone(), + unit: gauge.unit.clone(), + count: s.count(), + last: s.last(), + min: s.min(), + max: s.max(), + mean: s.mean(), + } + } +} + impl StageProfilingInfo { fn from_entry(stage_type: &str, index: usize, entry: &StageEntry) -> Self { let m = &entry.metrics; @@ -64,4 +106,40 @@ impl RecordProfilingMetrics { } out } + + /// Returns a serializable snapshot of every registered signal gauge. + pub fn signal_snapshot(&self) -> Vec { + self.signals() + .iter() + .map(SignalStatsInfo::from_gauge) + .collect() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use alloc::string::ToString; + + /// `unit` is skipped when empty (`Observable::UNIT` defaults to `""`), so + /// the emitted JSON must deserialize back without it — aimdb-client and + /// aimdb-mcp read these snapshots from `record.list`/`record.get`. + #[test] + fn empty_unit_roundtrips() { + let info = SignalStatsInfo { + signal: "temperature".to_string(), + unit: String::new(), + count: 3, + last: 1.0, + min: 0.5, + max: 2.0, + mean: 1.2, + }; + + let json = serde_json::to_string(&info).unwrap(); + assert!(!json.contains("unit"), "empty unit must be omitted: {json}"); + + let back: SignalStatsInfo = serde_json::from_str(&json).unwrap(); + assert_eq!(back, info); + } } diff --git a/aimdb-core/src/profiling/mod.rs b/aimdb-core/src/profiling/mod.rs index 02b8baa9..2f0b9fb5 100644 --- a/aimdb-core/src/profiling/mod.rs +++ b/aimdb-core/src/profiling/mod.rs @@ -17,10 +17,12 @@ mod info; mod record_profiling; +mod signal_stats; mod stage_metrics; -pub use info::StageProfilingInfo; -pub use record_profiling::{RecordProfilingMetrics, StageEntry}; +pub use info::{SignalStatsInfo, StageProfilingInfo}; +pub use record_profiling::{RecordProfilingMetrics, SignalGauge, StageEntry}; +pub use signal_stats::SignalStats; pub use stage_metrics::StageMetrics; use alloc::{boxed::Box, sync::Arc}; diff --git a/aimdb-core/src/profiling/record_profiling.rs b/aimdb-core/src/profiling/record_profiling.rs index b524c4ee..5be34e18 100644 --- a/aimdb-core/src/profiling/record_profiling.rs +++ b/aimdb-core/src/profiling/record_profiling.rs @@ -2,7 +2,7 @@ use alloc::{string::String, sync::Arc, vec::Vec}; -use crate::profiling::StageMetrics; +use crate::profiling::{SignalStats, StageMetrics}; use crate::StageKind; /// One registered stage: its shared metrics plus an optional human-readable name @@ -24,6 +24,18 @@ impl StageEntry { } } +/// One registered signal gauge: its label/unit plus the shared statistics that +/// `.observe()` folds `Observable::signal()` into. +#[derive(Debug)] +pub struct SignalGauge { + /// Signal label (`Observable::SIGNAL`, defaults to the schema name). + pub name: String, + /// Unit label (`Observable::UNIT`, e.g. `"°C"`). + pub unit: String, + /// Shared cumulative statistics (last/min/max/mean). + pub stats: Arc, +} + /// All stage profiling metrics for a single record, indexed by registration order /// within each stage kind (`sources[0]` is the first `.source()`, `taps[1]` the /// second `.tap()`, etc.). @@ -33,6 +45,7 @@ pub struct RecordProfilingMetrics { taps: Vec, links: Vec, transforms: Vec, + signals: Vec, } impl RecordProfilingMetrics { @@ -41,12 +54,13 @@ impl RecordProfilingMetrics { Self::default() } - /// `true` if no stages have been registered. + /// `true` if no stages and no signal gauges have been registered. pub fn is_empty(&self) -> bool { self.sources.is_empty() && self.taps.is_empty() && self.links.is_empty() && self.transforms.is_empty() + && self.signals.is_empty() } /// Registers a new source stage; returns its index and shared metrics handle. @@ -120,6 +134,23 @@ impl RecordProfilingMetrics { &self.transforms } + /// Registers a new signal gauge; returns its shared stats handle. Called by + /// `RecordRegistrar::signal_gauge` (which `.observe()` builds on). + pub fn push_signal_gauge(&mut self, name: &str, unit: &str) -> Arc { + let stats = Arc::new(SignalStats::new()); + self.signals.push(SignalGauge { + name: String::from(name), + unit: String::from(unit), + stats: stats.clone(), + }); + stats + } + + /// All registered signal gauges, in registration order. + pub fn signals(&self) -> &[SignalGauge] { + &self.signals + } + /// Assigns a name to a previously registered stage. No-op if `idx` is out of range. pub fn set_stage_name(&mut self, kind: StageKind, idx: usize, name: &str) { let vec = match kind { @@ -133,7 +164,7 @@ impl RecordProfilingMetrics { } } - /// Resets every stage's counters. + /// Resets every stage's counters and signal-gauge statistics. pub fn reset_all(&self) { for e in self .sources @@ -144,6 +175,9 @@ impl RecordProfilingMetrics { { e.metrics.reset(); } + for g in &self.signals { + g.stats.reset(); + } } } diff --git a/aimdb-core/src/profiling/signal_stats.rs b/aimdb-core/src/profiling/signal_stats.rs new file mode 100644 index 00000000..0b1a6f57 --- /dev/null +++ b/aimdb-core/src/profiling/signal_stats.rs @@ -0,0 +1,185 @@ +//! Per-record signal-gauge statistics (feature `observability`). +//! +//! Where [`StageMetrics`](crate::profiling::StageMetrics) times *how long* a +//! stage runs, `SignalStats` folds the *domain value* a record carries — +//! `Observable::signal()` fed in via `.observe()` — into last/min/max/count/mean. +//! It surfaces on the same introspection paths (`record.list`/`record.get`, +//! stage profiling) so a temperature's live °C shows up next to its timing. + +use core::sync::atomic::Ordering; +use portable_atomic::AtomicU64; + +/// Cumulative statistics for one domain signal. +/// +/// Values are `f64`; each field is an `AtomicU64` holding the IEEE-754 bits, so +/// updates are lock-free on every target (no `AtomicF64` feature needed). +/// Updated with `Ordering::Relaxed` — these are diagnostics, not synchronization +/// primitives; concurrent `.observe()` taps (should there be more than one) may +/// call [`update`](Self::update) at once. +#[derive(Debug)] +pub struct SignalStats { + count: AtomicU64, + last_bits: AtomicU64, + min_bits: AtomicU64, + max_bits: AtomicU64, + sum_bits: AtomicU64, +} + +impl Default for SignalStats { + fn default() -> Self { + Self { + count: AtomicU64::new(0), + last_bits: AtomicU64::new(0), + // Seeded to ±∞ so the first real sample always wins the min/max CAS. + min_bits: AtomicU64::new(f64::INFINITY.to_bits()), + max_bits: AtomicU64::new(f64::NEG_INFINITY.to_bits()), + sum_bits: AtomicU64::new(0), + } + } +} + +impl SignalStats { + /// Creates empty stats (count 0). + pub fn new() -> Self { + Self::default() + } + + /// Records one signal sample. + pub fn update(&self, value: f64) { + self.last_bits.store(value.to_bits(), Ordering::Relaxed); + self.count.fetch_add(1, Ordering::Relaxed); + Self::accumulate(&self.sum_bits, |cur| cur + value); + Self::keep_if(&self.min_bits, value, |value, cur| value < cur); + Self::keep_if(&self.max_bits, value, |value, cur| value > cur); + } + + /// CAS loop applying `f` to the stored `f64`. + fn accumulate(cell: &AtomicU64, f: impl Fn(f64) -> f64) { + let mut cur = cell.load(Ordering::Relaxed); + loop { + let next = f(f64::from_bits(cur)).to_bits(); + match cell.compare_exchange_weak(cur, next, Ordering::Relaxed, Ordering::Relaxed) { + Ok(_) => break, + Err(actual) => cur = actual, + } + } + } + + /// CAS loop storing `value` while `keep(value, current)` holds. + fn keep_if(cell: &AtomicU64, value: f64, keep: impl Fn(f64, f64) -> bool) { + let mut cur = cell.load(Ordering::Relaxed); + while keep(value, f64::from_bits(cur)) { + match cell.compare_exchange_weak( + cur, + value.to_bits(), + Ordering::Relaxed, + Ordering::Relaxed, + ) { + Ok(_) => break, + Err(actual) => cur = actual, + } + } + } + + /// Number of samples recorded. + pub fn count(&self) -> u64 { + self.count.load(Ordering::Relaxed) + } + + /// Most recent sample (0.0 if none). + pub fn last(&self) -> f64 { + f64::from_bits(self.last_bits.load(Ordering::Relaxed)) + } + + /// Smallest sample seen (0.0 if none). + pub fn min(&self) -> f64 { + if self.count() == 0 { + 0.0 + } else { + f64::from_bits(self.min_bits.load(Ordering::Relaxed)) + } + } + + /// Largest sample seen (0.0 if none). + pub fn max(&self) -> f64 { + if self.count() == 0 { + 0.0 + } else { + f64::from_bits(self.max_bits.load(Ordering::Relaxed)) + } + } + + /// Mean of all samples (0.0 if none). + pub fn mean(&self) -> f64 { + let count = self.count(); + if count == 0 { + 0.0 + } else { + f64::from_bits(self.sum_bits.load(Ordering::Relaxed)) / count as f64 + } + } + + /// Clears all statistics back to their initial state. + pub fn reset(&self) { + self.count.store(0, Ordering::Relaxed); + self.last_bits.store(0, Ordering::Relaxed); + self.min_bits + .store(f64::INFINITY.to_bits(), Ordering::Relaxed); + self.max_bits + .store(f64::NEG_INFINITY.to_bits(), Ordering::Relaxed); + self.sum_bits.store(0, Ordering::Relaxed); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn empty_is_zero() { + let s = SignalStats::new(); + assert_eq!(s.count(), 0); + assert_eq!(s.last(), 0.0); + assert_eq!(s.min(), 0.0); + assert_eq!(s.max(), 0.0); + assert_eq!(s.mean(), 0.0); + } + + #[test] + fn tracks_last_min_max_mean() { + let s = SignalStats::new(); + for v in [10.0, 30.0, 20.0] { + s.update(v); + } + assert_eq!(s.count(), 3); + assert_eq!(s.last(), 20.0); + assert_eq!(s.min(), 10.0); + assert_eq!(s.max(), 30.0); + assert_eq!(s.mean(), 20.0); + } + + #[test] + fn handles_negative_values() { + let s = SignalStats::new(); + for v in [-5.0, -20.0, -1.0] { + s.update(v); + } + assert_eq!(s.min(), -20.0); + assert_eq!(s.max(), -1.0); + assert_eq!(s.last(), -1.0); + } + + #[test] + fn reset_clears_then_reusable() { + let s = SignalStats::new(); + s.update(5.0); + s.update(7.0); + s.reset(); + assert_eq!(s.count(), 0); + assert_eq!(s.min(), 0.0); + s.update(3.0); + assert_eq!(s.count(), 1); + assert_eq!(s.min(), 3.0); + assert_eq!(s.max(), 3.0); + } +} diff --git a/aimdb-core/src/remote/metadata.rs b/aimdb-core/src/remote/metadata.rs index 01c322e2..fb2c8426 100644 --- a/aimdb-core/src/remote/metadata.rs +++ b/aimdb-core/src/remote/metadata.rs @@ -86,6 +86,14 @@ pub struct RecordMetadata { #[cfg(feature = "observability")] #[serde(skip_serializing_if = "Option::is_none")] pub stage_profiling: Option>, + + // ===== Signal gauges (feature-gated) ===== + /// Per-record domain-signal statistics (last/min/max/mean) fed by + /// `Observable::observe()`, if the `observability` feature is enabled and any + /// gauge has been registered. + #[cfg(feature = "observability")] + #[serde(skip_serializing_if = "Option::is_none")] + pub signal_stats: Option>, } impl RecordMetadata { @@ -139,6 +147,8 @@ impl RecordMetadata { occupancy: None, #[cfg(feature = "observability")] stage_profiling: None, + #[cfg(feature = "observability")] + signal_stats: None, } } @@ -169,6 +179,15 @@ impl RecordMetadata { } self } + + /// Attaches a signal-gauge snapshot (observability feature only). + #[cfg(feature = "observability")] + pub fn with_signal_stats(mut self, signals: Vec) -> Self { + if !signals.is_empty() { + self.signal_stats = Some(signals); + } + self + } } #[cfg(test)] diff --git a/aimdb-core/src/signal.rs b/aimdb-core/src/signal.rs new file mode 100644 index 00000000..4e6a9d6d --- /dev/null +++ b/aimdb-core/src/signal.rs @@ -0,0 +1,49 @@ +//! Signal gauge handle: per-record domain-signal statistics. +//! +//! `RecordRegistrar::signal_gauge` hands back a [`SignalGaugeHandle`]. Feeding +//! values in with [`update`](SignalGaugeHandle::update) folds them into the +//! record's `SignalStats` (last/min/max/mean), which surface on `record.list` / +//! `record.get` and stage-profiling output. +//! +//! Like [`with_name`](crate::typed_api::RecordRegistrar::with_name), the handle +//! is **always available**: when the `observability` feature is off (or no gauge +//! was registered) it is inert and `update` is a no-op, so callers — including +//! the `.observe()` extension in `aimdb-data-contracts` — never `#[cfg]` on +//! core's features. + +/// A handle to a per-record signal gauge. +/// +/// Cheap to clone (an `Arc` bump, or nothing when inert). Obtained from +/// [`RecordRegistrar::signal_gauge`](crate::typed_api::RecordRegistrar::signal_gauge). +#[derive(Clone)] +pub struct SignalGaugeHandle { + #[cfg(feature = "observability")] + stats: Option>, +} + +impl SignalGaugeHandle { + /// An inert handle that records nothing (feature off, or no gauge registered). + #[cfg_attr(feature = "observability", allow(dead_code))] + pub(crate) fn inert() -> Self { + Self { + #[cfg(feature = "observability")] + stats: None, + } + } + + /// A live handle backed by shared statistics. + #[cfg(feature = "observability")] + pub(crate) fn live(stats: alloc::sync::Arc) -> Self { + Self { stats: Some(stats) } + } + + /// Records one signal sample. No-op on an inert handle. + pub fn update(&self, value: f64) { + #[cfg(feature = "observability")] + if let Some(stats) = &self.stats { + stats.update(value); + } + #[cfg(not(feature = "observability"))] + let _ = value; + } +} diff --git a/aimdb-core/src/typed_api.rs b/aimdb-core/src/typed_api.rs index ce8d35ff..136f0dc3 100644 --- a/aimdb-core/src/typed_api.rs +++ b/aimdb-core/src/typed_api.rs @@ -440,6 +440,33 @@ where self } + /// Registers a signal gauge on this record and returns a handle to feed it. + /// + /// Values pushed via [`SignalGaugeHandle::update`](crate::SignalGaugeHandle::update) + /// fold into per-record last/min/max/mean statistics that surface on + /// `record.list` / `record.get` and stage profiling. This is the core hook + /// behind `aimdb-data-contracts`' `Observable::observe()`. + /// + /// Always available, mirroring [`with_name`](Self::with_name): when the + /// `observability` feature is disabled it returns an inert handle whose + /// `update` is a no-op, so callers never `#[cfg]` on core's features. + pub fn signal_gauge( + &mut self, + name: &'static str, + unit: &'static str, + ) -> crate::signal::SignalGaugeHandle { + #[cfg(feature = "observability")] + { + let stats = self.rec.profiling_mut().push_signal_gauge(name, unit); + crate::signal::SignalGaugeHandle::live(stats) + } + #[cfg(not(feature = "observability"))] + { + let _ = (name, unit); + crate::signal::SignalGaugeHandle::inert() + } + } + /// Registers a producer service for this record type. /// /// The closure receives the [`RuntimeContext`](crate::RuntimeContext) diff --git a/aimdb-core/src/typed_record.rs b/aimdb-core/src/typed_record.rs index 3a503ada..a2ffc33d 100644 --- a/aimdb-core/src/typed_record.rs +++ b/aimdb-core/src/typed_record.rs @@ -1183,6 +1183,10 @@ impl AnyRecord for TypedRecord { #[cfg(feature = "observability")] let metadata = metadata.with_stage_profiling(self.profiling.snapshot()); + // Attach signal-gauge statistics (from `.observe()`) when enabled. + #[cfg(feature = "observability")] + let metadata = metadata.with_signal_stats(self.profiling.signal_snapshot()); + metadata } diff --git a/aimdb-data-contracts/CHANGELOG.md b/aimdb-data-contracts/CHANGELOG.md index 13362c87..c4d41cb8 100644 --- a/aimdb-data-contracts/CHANGELOG.md +++ b/aimdb-data-contracts/CHANGELOG.md @@ -7,7 +7,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] -### Changed (breaking) +### Changed (breaking) — Design 041: capability traits as first-class verbs + +- **`Simulatable` reshaped for compile-time-only simulation.** `simulate(config: &SimulationConfig, ...)` → `simulate(params: &Self::Params, ...)` with a new associated `type Params`. `SimulationConfig` (including its runtime `enabled` gate) and `SimulationParams` are removed; `SimProfile

{ interval_ms, params }` and off-the-shelf `RandomWalkParams` replace them (`type Params = RandomWalkParams` is the migration path for existing scalar-walk impls). `simulatable` now also requires `aimdb-core` (for the new `SimulatableRegistrarExt`) and `rand` drops the `std_rng` feature (RNG is always caller-supplied). +- **`Observable` loses `ICON` and `format_log()`.** The trait is now numeric-projection-only: `type Signal`, `const SIGNAL` (defaults to `Self::NAME`), `const UNIT`, `fn signal()`. Presentation moved to `ObservableRegistrarExt::log(node_id)`. +- **`Settable` moves behind a new `settable` feature** (was compiled unconditionally) for tier symmetry with the other wire contracts. + +### Added + +- `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`. +- `#[derive(Linkable)]` (via `aimdb-derive`) — emits a JSON `Linkable` impl (`serde_json::to_vec`/`from_slice`), `no_std + alloc` compatible. + +### Changed - **`migratable` no longer requires `std`.** `migratable = ["alloc", "serde_json"]` (was `["std", "serde_json"]`); the crate's `serde_json` dependency now follows the workspace pin (`default-features = false, features = ["alloc"]`) instead of pulling in `serde_json/std` by default. `migration_chain!` and both `Linkable`-with-migration patterns now compile on `no_std + alloc` (e.g. `thumbv7em-none-eabihf`). A crate that enabled only `migratable` and relied on it transitively activating `std` will need to enable `std` explicitly. In-repo consumers are unaffected (they enable `std` by default). - `log_tap(ctx, consumer, node_id)` now takes `Consumer` instead of `Consumer` — `R` is still on `RuntimeContext` (Design 029, M14). User-visible signature shrink only; behaviour unchanged. diff --git a/aimdb-data-contracts/Cargo.toml b/aimdb-data-contracts/Cargo.toml index 1153576b..f8e5ff67 100644 --- a/aimdb-data-contracts/Cargo.toml +++ b/aimdb-data-contracts/Cargo.toml @@ -1,12 +1,12 @@ [package] name = "aimdb-data-contracts" -version = "0.2.0" +version = "0.3.0" edition.workspace = true authors.workspace = true license.workspace = true repository.workspace = true homepage.workspace = true -description = "Trait definitions for AimDB data contracts: SchemaType, Streamable, Observable, Linkable, Simulatable, Migratable" +description = "Trait definitions for AimDB data contracts: SchemaType, Streamable, Observable, Linkable, Simulatable, Migratable, Settable" keywords = ["aimdb", "iot", "edge", "schema", "contracts"] categories = ["embedded", "no-std"] @@ -14,30 +14,42 @@ categories = ["embedded", "no-std"] default = ["std"] alloc = ["serde/alloc"] std = ["alloc", "serde/std", "serde_json?/std"] -linkable = ["alloc", "serde_json"] -simulatable = ["rand"] +# The ext trait needs core's registrar/connector-builder types (same wiring as +# `observable`); `dep:aimdb-derive` powers `#[derive(Linkable)]`. +linkable = ["alloc", "serde_json", "aimdb-core", "dep:aimdb-derive"] +# 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`). +simulatable = ["rand", "aimdb-core"] migratable = ["alloc", "serde_json", "dep:aimdb-derive"] observable = ["alloc", "aimdb-core"] +# Feature-gated for tier symmetry with the other wire contracts. A pure trait — +# no dependencies of its own. +settable = [] [dependencies] serde = { version = "1.0", default-features = false, features = ["derive"] } serde_json = { workspace = true, optional = true } -# migration_chain! proc-macro (build-time only; no target/runtime/no_std -# impact — see aimdb-derive::migration_chain). No cycle: aimdb-derive emits -# token paths to this crate but does not depend on it. -aimdb-derive = { version = "0.2.0", path = "../aimdb-derive", optional = true } +# migration_chain! + #[derive(Linkable)] proc-macros (build-time only; no +# target/runtime/no_std impact — see aimdb-derive). No cycle: aimdb-derive +# emits token paths to this crate but does not depend on it. +aimdb-derive = { version = "0.3.0", path = "../aimdb-derive", optional = true } -# Optional dependencies for observable feature (log_tap function) -# Note: aimdb-core requires alloc feature for core functionality -aimdb-core = { version = "1.1.0", path = "../aimdb-core", optional = true, default-features = false, features = [ +# Optional: shared by the `linkable`, `observable`, and `simulatable` registrar +# ext traits (RecordRegistrar/RuntimeContext). 1.2.0 is the true minimum: the +# `observable` feature's `.observe()` calls `RecordRegistrar::signal_gauge`, +# added in 1.2.0. Note: aimdb-core requires the alloc feature for core +# functionality. +aimdb-core = { version = "1.2.0", path = "../aimdb-core", optional = true, default-features = false, features = [ "alloc", ] } +# RNG is caller-supplied, so we need only the `Rng`/`RngExt` traits and the +# always-available `SmallRng` — no `std_rng`. [dependencies.rand] version = "0.10.1" optional = true default-features = false -features = ["std_rng"] [dev-dependencies] serde_json = "1.0" diff --git a/aimdb-data-contracts/src/lib.rs b/aimdb-data-contracts/src/lib.rs index e8792f63..f6f5c573 100644 --- a/aimdb-data-contracts/src/lib.rs +++ b/aimdb-data-contracts/src/lib.rs @@ -71,11 +71,23 @@ pub use streamable::Streamable; #[cfg(feature = "linkable")] mod linkable; +#[cfg(feature = "linkable")] +pub use linkable::LinkableRegistrarExt; + +/// `#[derive(Linkable)]` — see [`Linkable`] and `aimdb_derive::Linkable`'s docs. +/// +/// Re-exported under the same name as the trait, following the trait+derive +/// pairing convention (`serde::Serialize`, `aimdb_derive::RecordKey`) — proc-macro +/// derive names live in a separate namespace from traits, so this is not a +/// conflict. +#[cfg(feature = "linkable")] +pub use aimdb_derive::Linkable; + #[cfg(feature = "observable")] mod observable; #[cfg(feature = "observable")] -pub use observable::log_tap; +pub use observable::{log_tap, ObservableRegistrarExt}; #[cfg(feature = "simulatable")] mod simulatable; @@ -84,7 +96,7 @@ mod simulatable; mod migratable; #[cfg(feature = "simulatable")] -pub use simulatable::{SimulationConfig, SimulationParams}; +pub use simulatable::{RandomWalkParams, SimProfile, Simulatable, SimulatableRegistrarExt}; #[cfg(feature = "migratable")] pub use migratable::{MigrationChain, MigrationError, MigrationStep}; @@ -134,34 +146,12 @@ pub trait SchemaType: Sized { const VERSION: u32 = 1; } -// ═══════════════════════════════════════════════════════════════════ -// SIMULATABLE SUPPORT (feature = "simulatable") -// ═══════════════════════════════════════════════════════════════════ - -/// Generate realistic test/simulation data. -/// -/// This is an intrinsic capability of the schema type itself, -/// not a policy decision. If a type can be simulated, implement this. -#[cfg(feature = "simulatable")] -pub trait Simulatable: SchemaType { - /// Generate a new sample with optional reference to previous value. - /// - /// # Parameters - /// - `config`: Simulation parameters (type-specific) - /// - `previous`: Optional reference to last generated value (for random walks, trends) - /// - `rng`: Random number generator - /// - `timestamp`: Unix timestamp in milliseconds - fn simulate( - config: &SimulationConfig, - previous: Option<&Self>, - rng: &mut R, - timestamp: u64, - ) -> Self; -} - /// Construct a schema instance from its primary value. /// -/// This defines the canonical way to create a new reading/measurement. +/// This defines the canonical way to create a new reading/measurement — the +/// write counterpart to [`Observable::signal`]'s read projection. Implementing +/// it unlocks the `SyncProducer::set_value` family in `aimdb-sync`. +#[cfg(feature = "settable")] pub trait Settable: SchemaType { /// The primary value type (e.g., `f32` for temperature) type Value; @@ -178,52 +168,51 @@ pub trait Settable: SchemaType { // OBSERVABLE SUPPORT // ═══════════════════════════════════════════════════════════════════ -/// Extract a signal value for observation. +/// Project a schema type onto a numeric domain signal. +/// +/// The trait's kernel is the numeric projection: implement it, call +/// [`ObservableRegistrarExt::observe`], +/// and the signal is folded into live last/min/max/mean statistics that surface +/// on `record.list` / `record.get` and stage profiling. The signal can also feed +/// threshold checks, alerting, and aggregation. +/// +/// Presentation is not part of the trait: `.log(node_id)` (also on the ext +/// trait) formats a human-readable line from `Debug` + +/// [`SIGNAL`](Observable::SIGNAL)/[`UNIT`](Observable::UNIT). +/// +/// # Feature layering /// -/// Implement this trait to enable threshold checking, alerting, -/// and other signal-based operations on your schema type. +/// Three layers, each useful without the next: /// -/// The extracted signal can be used by node implementations to: -/// - Check against configured thresholds -/// - Trigger alerts when bounds are exceeded -/// - Compute aggregations (mean, min, max) -/// - Feed into monitoring systems -/// - Format log output with `format_log()` +/// 1. **This trait** — always compiled, no features. [`signal()`](Observable::signal) +/// plus the `SIGNAL`/`UNIT` labels are enough for hand-wired `.tap()`s, +/// threshold checks, and generic code over `T: Observable`. +/// 2. **`observable` (this crate)** — unlocks the registrar verbs `.observe()` +/// and `.log()`. Gated only because the ext trait needs `alloc` and +/// `aimdb-core`. +/// 3. **`observability` (`aimdb-core`)** — the metrics backend. When it is off, +/// `.observe()` still compiles and runs but its gauge is inert: updates are +/// no-ops and nothing surfaces on `record.list` / `record.get` (see +/// `RecordRegistrar::signal_gauge`). `.log()` is unaffected. This lets +/// constrained targets ship `Observable` contracts while compiling the +/// metrics cost away. pub trait Observable: SchemaType { /// The numeric type of the signal (e.g., `f32`, `f64`, `i32`). /// - /// Must be comparable and copyable for threshold checks. + /// Must be comparable and copyable for threshold checks. `.observe()` + /// additionally requires `Signal: Into` at its call site; a type with + /// an exotic signal can still implement `Observable` and write its own tap. type Signal: PartialOrd + Copy; - /// Icon/emoji for log output (e.g., "🌡️", "💧", "📊") - /// - /// Override this to provide a visual indicator for your data type. - const ICON: &'static str = "📊"; + /// What the signal means, for metrics/UI labels (e.g. `"celsius"`). + /// Defaults to the schema name. + const SIGNAL: &'static str = Self::NAME; - /// Unit label for the signal (e.g., "°C", "%", "hPa") - /// - /// Override this to display the appropriate unit in log output. + /// Unit label for the signal (e.g. `"°C"`, `"%"`, `"hPa"`). const UNIT: &'static str = ""; /// Extract the signal value from this instance. fn signal(&self) -> Self::Signal; - - /// Format a log entry for this observation. - /// - /// The default implementation uses `Debug` formatting. Override this - /// for prettier, human-readable output. - /// - /// # Example output - /// ```text - /// 🌡️ [alpha] Temperature: 22.5°C at 1704326400000 - /// 💧 [beta] Humidity: 65.3% at 1704326400000 - /// ``` - fn format_log(&self, node_id: &str) -> alloc::string::String - where - Self: core::fmt::Debug, - { - alloc::format!("{} [{}] {:?}", Self::ICON, node_id, self) - } } // ═══════════════════════════════════════════════════════════════════ @@ -232,26 +221,25 @@ pub trait Observable: SchemaType { /// Types that can be serialized/deserialized for connector links. /// -/// Implement this trait to enable `link_from` and `link_to` operations -/// in AimDB connectors (MQTT, KNX, etc.). This provides the wire format -/// for transporting schema types across network boundaries. +/// Implement this trait, then call +/// [`LinkableRegistrarExt::linked_from`](linkable::LinkableRegistrarExt::linked_from) / +/// [`linked_to`](linkable::LinkableRegistrarExt::linked_to) to wire `link_from` +/// / `link_to` in AimDB connectors (MQTT, KNX, etc.) with the codec defaulted to +/// `from_bytes`/`to_bytes`. This provides the wire format for transporting +/// schema types across network boundaries. /// /// # Example /// -/// Not compiled: the snippet needs `aimdb-core`'s builder, which this crate -/// only depends on under the `observable` feature — `linkable` alone has no -/// core dependency. +/// Not compiled: the snippet needs `aimdb-core`'s builder types in scope. /// /// ```rust,ignore -/// use aimdb_data_contracts::Linkable; +/// use aimdb_data_contracts::{Linkable, LinkableRegistrarExt}; /// use my_app::Temperature; // user-defined type implementing Linkable /// /// // In connector configuration: /// builder.configure::(NODE_ID, |reg| { -/// reg.buffer(BufferCfg::SingleLatest) -/// .link_from("mqtt://sensors/temperature") -/// .with_deserializer(Temperature::from_bytes) -/// .finish(); +/// reg.buffer(BufferCfg::SingleLatest); +/// reg.linked_from("mqtt://sensors/temperature"); /// }); /// ``` #[cfg(feature = "linkable")] diff --git a/aimdb-data-contracts/src/linkable.rs b/aimdb-data-contracts/src/linkable.rs index 17a8d095..162a5679 100644 --- a/aimdb-data-contracts/src/linkable.rs +++ b/aimdb-data-contracts/src/linkable.rs @@ -1,7 +1,51 @@ -//! Linkable trait implementations and tests. +//! Linkable registrar extension: one-line link verbs for connector wiring. //! -//! This module provides the wire format support for transporting schema types -//! across connector links (MQTT, KNX, etc.). +//! Implementing [`Linkable`](crate::Linkable) unlocks two 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). + +use aimdb_core::connector::SerializeError; +use aimdb_core::typed_api::RecordRegistrar; + +use crate::Linkable; + +/// Adds `.linked_from(url)` and `.linked_to(url)` to [`RecordRegistrar`] for +/// [`Linkable`] types. +pub trait LinkableRegistrarExt<'a, T> +where + T: Linkable + Send + Sync + Clone + core::fmt::Debug + 'static, +{ + /// `.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`. + /// + /// `Linkable::to_bytes`'s `String` error is mapped to + /// `SerializeError::InvalidData` — the connector layer's serializer error + /// type carries no string detail. + fn linked_to(&mut self, url: &str) -> &mut RecordRegistrar<'a, T>; +} + +impl<'a, T> LinkableRegistrarExt<'a, T> for RecordRegistrar<'a, T> +where + T: Linkable + Send + Sync + Clone + core::fmt::Debug + 'static, +{ + fn linked_from(&mut self, url: &str) -> &mut RecordRegistrar<'a, T> { + self.link_from(url) + .with_deserializer(|_ctx, bytes| T::from_bytes(bytes)) + .finish() + } + + 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() + } +} #[cfg(test)] mod tests { diff --git a/aimdb-data-contracts/src/observable.rs b/aimdb-data-contracts/src/observable.rs index 105c35c2..503b6a34 100644 --- a/aimdb-data-contracts/src/observable.rs +++ b/aimdb-data-contracts/src/observable.rs @@ -1,55 +1,99 @@ -//! Observable helper functions for data contracts. +//! Observable registrar extension + logging tap for data contracts. //! -//! Provides tap functions for logging observable data types. +//! Implementing [`Observable`](crate::Observable) unlocks one verb — +//! [`ObservableRegistrarExt::observe`] — which feeds the record's domain signal +//! into core signal-gauge metrics. `.log(node_id)` is the human-readable +//! companion for console watching. extern crate alloc; +use aimdb_core::typed_api::RecordRegistrar; + use crate::Observable; // ═══════════════════════════════════════════════════════════════════ -// LOG TAP (feature = "observable") +// LOG TAP // ═══════════════════════════════════════════════════════════════════ -/// Generic logging tap for Observable types. -/// -/// This function can be used with any type implementing `Observable` to -/// log observations as they flow through the mesh. It uses `format_log()` -/// to produce human-readable output. -/// -/// # Example +/// Generic logging tap for [`Observable`] types. /// -/// ```no_run -/// use aimdb_data_contracts::log_tap; -/// # use aimdb_core::AimDbBuilder; -/// # use aimdb_data_contracts::{Observable, SchemaType}; -/// # #[derive(Clone, Debug)] -/// # struct Temperature { celsius: f32 } -/// # impl SchemaType for Temperature { const NAME: &'static str = "temperature"; } -/// # impl Observable for Temperature { -/// # type Signal = f32; -/// # fn signal(&self) -> f32 { self.celsius } -/// # } -/// # fn wire(builder: &mut AimDbBuilder) { -/// builder.configure::("node.alpha", |reg| { -/// // .buffer(BufferCfg::SingleLatest) — via your runtime adapter's ext trait -/// reg.tap(|ctx, consumer| log_tap(ctx, consumer, "alpha")); -/// }); -/// # } -/// ``` -#[cfg(feature = "observable")] +/// Formats each value from `Debug` plus the schema's +/// [`SIGNAL`](Observable::SIGNAL)/[`UNIT`](Observable::UNIT) labels. Prefer +/// [`ObservableRegistrarExt::log`]; this free function remains for hand-wired +/// `.tap()` calls. pub async fn log_tap( ctx: aimdb_core::RuntimeContext, consumer: aimdb_core::typed_api::Consumer, - node_id: &'static str, + node_id: &str, ) where T: Observable + Send + Sync + Clone + core::fmt::Debug + 'static, { let log = ctx.log(); - let mut reader = consumer.subscribe(); while let Ok(value) = reader.recv().await { - log.info(&value.format_log(node_id)); + log.info(&alloc::format!( + "[{}] {}: {:?}{}", + node_id, + T::SIGNAL, + value, + T::UNIT + )); + } +} + +// ═══════════════════════════════════════════════════════════════════ +// REGISTRAR EXTENSION: `.observe()` / `.log(node_id)` +// ═══════════════════════════════════════════════════════════════════ + +/// Adds `.observe()` and `.log(node_id)` to [`RecordRegistrar`] for +/// [`Observable`] types. +pub trait ObservableRegistrarExt<'a, T> +where + T: Observable + Send + Sync + Clone + core::fmt::Debug + 'static, +{ + /// Feed `T::signal()` into the record's signal gauge (last/min/max/mean), + /// visible via `record.list` / `record.get` / stage profiling. + /// + /// Recording requires `aimdb-core`'s `observability` feature: without it + /// [`RecordRegistrar::signal_gauge`] hands back an inert gauge and this + /// tap compiles and runs but records nothing. `.log()` does not depend on + /// that feature. + /// + /// Bounded `T::Signal: Into` here (not on the trait), so `f32`/`i32`/ + /// `u32` signals qualify while a type with an exotic signal can still + /// implement `Observable` and write its own tap. + fn observe(&mut self) -> &mut RecordRegistrar<'a, T> + where + T::Signal: Into; + + /// Log each value to the runtime log, formatted from `Debug` + + /// `SIGNAL`/`UNIT`. For humans watching a console; `.observe()` is the + /// metrics path. + fn log(&mut self, node_id: &'static str) -> &mut RecordRegistrar<'a, T>; +} + +impl<'a, T> ObservableRegistrarExt<'a, T> for RecordRegistrar<'a, T> +where + T: Observable + Send + Sync + Clone + core::fmt::Debug + 'static, +{ + fn observe(&mut self) -> &mut RecordRegistrar<'a, T> + where + T::Signal: Into, + { + let gauge = self.signal_gauge(T::SIGNAL, T::UNIT); + self.tap(move |_ctx, consumer| async move { + let mut reader = consumer.subscribe(); + while let Ok(value) = reader.recv().await { + gauge.update(value.signal().into()); + } + }) + .with_name("observe") + } + + fn log(&mut self, node_id: &'static str) -> &mut RecordRegistrar<'a, T> { + self.tap(move |ctx, consumer| log_tap(ctx, consumer, node_id)) + .with_name("log") } } @@ -71,6 +115,7 @@ mod tests { impl Observable for TestSensor { type Signal = f32; + const UNIT: &'static str = "°C"; fn signal(&self) -> f32 { self.value @@ -78,54 +123,21 @@ mod tests { } #[test] - fn test_signal_extraction() { + fn signal_extraction() { let sensor = TestSensor { value: 42.5 }; assert_eq!(sensor.signal(), 42.5); } #[test] - fn test_threshold_comparison() { - let sensor = TestSensor { value: 25.0 }; - - // Above threshold - assert!(sensor.signal() > 20.0); - - // Below threshold - assert!(sensor.signal() < 30.0); - - // In range - let s = sensor.signal(); - assert!((20.0..=30.0).contains(&s)); - } - - #[test] - fn test_default_icon_and_unit() { - assert_eq!(TestSensor::ICON, "📊"); - assert_eq!(TestSensor::UNIT, ""); + fn signal_label_defaults_to_schema_name() { + assert_eq!(TestSensor::SIGNAL, "test_sensor"); + assert_eq!(TestSensor::UNIT, "°C"); } #[test] - fn test_format_log_default() { - #[derive(Debug)] - struct DebugSensor { - value: f32, - } - - impl SchemaType for DebugSensor { - const NAME: &'static str = "debug_sensor"; - } - - impl Observable for DebugSensor { - type Signal = f32; - fn signal(&self) -> f32 { - self.value - } - } - - let sensor = DebugSensor { value: 42.5 }; - let log = sensor.format_log("node1"); - assert!(log.contains("📊")); - assert!(log.contains("[node1]")); - assert!(log.contains("42.5")); + fn signal_is_into_f64() { + let sensor = TestSensor { value: 25.0 }; + let as_f64: f64 = sensor.signal().into(); + assert_eq!(as_f64, 25.0); } } diff --git a/aimdb-data-contracts/src/simulatable.rs b/aimdb-data-contracts/src/simulatable.rs index 0614b157..eaba29d7 100644 --- a/aimdb-data-contracts/src/simulatable.rs +++ b/aimdb-data-contracts/src/simulatable.rs @@ -1,82 +1,77 @@ -//! Simulation configuration for data contracts. +//! Simulation capability for data contracts (dev-tier, `feature = "simulatable"`). //! -//! These structures configure how `Simulatable` types generate test data. +//! Implementing [`Simulatable`] unlocks exactly one verb — +//! [`SimulatableRegistrarExt::simulate`] — which installs a source that emits +//! synthetic samples on a timer. This is the **dev tier**: it must never ship in +//! a production binary. Sim-to-real selection is a compile-time `#[cfg]` in the +//! application, never a runtime flag, and `rand` is the tracer CI uses to prove +//! production dependency graphs are sim-free (see `make check-no-sim`). use serde::{Deserialize, Serialize}; +use aimdb_core::typed_api::RecordRegistrar; + +use crate::SchemaType; + // ═══════════════════════════════════════════════════════════════════ -// SIMULATION CONFIG +// SIMULATABLE TRAIT // ═══════════════════════════════════════════════════════════════════ -/// Configuration for data simulation. +/// Dev-only capability: generate realistic synthetic data. /// -/// This is passed to `Simulatable::simulate()` to control how -/// test data is generated. -#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] -pub struct SimulationConfig { - /// Whether simulation is active - #[serde(default = "default_true")] - pub enabled: bool, - - /// Interval between simulated samples (milliseconds) - #[serde(default = "default_interval")] - pub interval_ms: u64, - - /// Type-specific parameters (schema implementations interpret these) - #[serde(default)] - pub params: SimulationParams, +/// This is an intrinsic capability of the schema type itself, not a policy +/// decision. If a type can be simulated, implement this — then call +/// [`SimulatableRegistrarExt::simulate`] on the registrar. +pub trait Simulatable: SchemaType { + /// Type-specific generation parameters (a temperature defines walk bounds, + /// a GPS track defines waypoints, …). Use [`RandomWalkParams`] for scalar + /// walks, or define your own. + type Params: Clone + Send + Sync + Default + 'static; + + /// Generate the next sample. + /// + /// - `params`: type-specific generation parameters. + /// - `previous`: last generated value, enabling walks/trends. + /// - `rng`: caller-supplied RNG (bring [`rand::RngExt`] into scope for + /// `.random()`). + /// - `timestamp_ms`: Unix millis supplied by the driving loop. + fn simulate( + params: &Self::Params, + previous: Option<&Self>, + rng: &mut R, + timestamp_ms: u64, + ) -> Self; } -fn default_true() -> bool { - true -} - -fn default_interval() -> u64 { - 1000 -} +// ═══════════════════════════════════════════════════════════════════ +// PROFILE + OFF-THE-SHELF PARAMS +// ═══════════════════════════════════════════════════════════════════ -impl Default for SimulationConfig { - fn default() -> Self { - Self { - enabled: true, - interval_ms: 1000, - params: SimulationParams::default(), - } - } +/// Loop policy + generation params for one record. +#[derive(Clone, Debug, Default)] +pub struct SimProfile

{ + /// Interval between samples (milliseconds). + pub interval_ms: u64, + /// Type-specific generation parameters. + pub params: P, } -/// Type-specific simulation parameters. -/// -/// Common parameters that many schema types use. Schema implementations -/// can interpret these as appropriate for their domain. +/// Off-the-shelf [`Simulatable::Params`] for scalar signals that wander around +/// a base value: set `type Params = RandomWalkParams` and derive the next +/// sample from `previous` plus a bounded random step. #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] -pub struct SimulationParams { - /// Base/center value for the simulation - #[serde(default)] +pub struct RandomWalkParams { + /// Base/center value for the walk. pub base: f64, - - /// Maximum deviation from base (for random walks) - #[serde(default = "default_variation")] + /// Maximum deviation from base. pub variation: f64, - - /// Linear trend per sample (positive = increasing, negative = decreasing) - #[serde(default)] + /// Linear trend applied per sample (positive = increasing). pub trend: f64, - - /// Step size multiplier for random walk (0.0-1.0) - #[serde(default = "default_step")] + /// Step-size multiplier for the random walk (0.0–1.0). pub step: f64, } -fn default_variation() -> f64 { - 1.0 -} - -fn default_step() -> f64 { - 0.2 -} - -impl Default for SimulationParams { +impl Default for RandomWalkParams { fn default() -> Self { Self { base: 0.0, @@ -87,6 +82,61 @@ impl Default for SimulationParams { } } +// ═══════════════════════════════════════════════════════════════════ +// REGISTRAR EXTENSION: `.simulate(profile, rng)` +// ═══════════════════════════════════════════════════════════════════ + +/// Adds `.simulate(profile, rng)` to [`RecordRegistrar`] for [`Simulatable`] types. +pub trait SimulatableRegistrarExt<'a, T> +where + T: Simulatable + Send + Sync + Clone + core::fmt::Debug + 'static, +{ + /// Install a source that emits `T::simulate(...)` every `interval_ms`, + /// driving the caller-supplied RNG (OS entropy on std, seeded PRNG on + /// no_std, fixed seed in tests). + /// + /// Installs a **source**, so single-writer-per-key is enforced by `build()` + /// exactly as for a hardware `.source()` — the two are mutually exclusive by + /// the app's `#[cfg]`, never both present in one binary. + fn simulate( + &mut self, + profile: SimProfile, + rng: R, + ) -> &mut RecordRegistrar<'a, T> + where + R: rand::Rng + Send + 'static; +} + +impl<'a, T> SimulatableRegistrarExt<'a, T> for RecordRegistrar<'a, T> +where + T: Simulatable + Send + Sync + Clone + core::fmt::Debug + 'static, +{ + fn simulate( + &mut self, + profile: SimProfile, + mut rng: R, + ) -> &mut RecordRegistrar<'a, T> + where + R: rand::Rng + Send + 'static, + { + self.source(move |ctx, producer| async move { + let mut prev: Option = None; + loop { + let now_ms = ctx + .time() + .unix_time() + .map(|(s, ns)| s.saturating_mul(1000) + (ns / 1_000_000) as u64) + .unwrap_or(0); + let next = T::simulate(&profile.params, prev.as_ref(), &mut rng, now_ms); + producer.produce(next.clone()); + prev = Some(next); + ctx.time().sleep_millis(profile.interval_ms.max(1)).await; + } + }) + .with_name("simulate") + } +} + // ═══════════════════════════════════════════════════════════════════ // TESTS // ═══════════════════════════════════════════════════════════════════ @@ -94,40 +144,61 @@ impl Default for SimulationParams { #[cfg(test)] mod tests { use super::*; + use rand::{RngExt, SeedableRng}; + + #[derive(Clone, Debug, PartialEq)] + struct Scalar(f64); + + impl SchemaType for Scalar { + const NAME: &'static str = "scalar"; + } + + impl Simulatable for Scalar { + type Params = RandomWalkParams; + + fn simulate( + params: &Self::Params, + previous: Option<&Self>, + rng: &mut R, + _timestamp_ms: u64, + ) -> Self { + let base = previous.map(|p| p.0).unwrap_or(params.base); + let delta = (rng.random::() - 0.5) * params.variation * params.step + params.trend; + Scalar(base + delta) + } + } #[test] - fn test_simulation_config_defaults() { - let config = SimulationConfig::default(); - assert!(config.enabled); - assert_eq!(config.interval_ms, 1000); - assert_eq!(config.params.base, 0.0); - assert_eq!(config.params.variation, 1.0); - assert_eq!(config.params.step, 0.2); + fn random_walk_defaults() { + let p = RandomWalkParams::default(); + assert_eq!(p.variation, 1.0); + assert_eq!(p.step, 0.2); + assert_eq!(p.base, 0.0); + assert_eq!(p.trend, 0.0); } #[test] - fn test_simulation_config_json_roundtrip() { - let json = r#"{ - "enabled": true, - "interval_ms": 500, - "params": { - "base": 22.0, - "variation": 3.0, - "trend": 0.1, - "step": 0.1 - } - }"#; - - let config: SimulationConfig = serde_json::from_str(json).unwrap(); - assert!(config.enabled); - assert_eq!(config.interval_ms, 500); - assert_eq!(config.params.base, 22.0); - assert_eq!(config.params.variation, 3.0); - assert_eq!(config.params.trend, 0.1); - - // Roundtrip - let serialized = serde_json::to_string(&config).unwrap(); - let deserialized: SimulationConfig = serde_json::from_str(&serialized).unwrap(); - assert_eq!(config, deserialized); + fn sim_profile_default_is_zero_interval() { + let profile: SimProfile = SimProfile::default(); + assert_eq!(profile.interval_ms, 0); + assert_eq!(profile.params, RandomWalkParams::default()); + } + + #[test] + fn simulate_is_deterministic_for_fixed_seed() { + let params = RandomWalkParams::default(); + // SmallRng is seedable and available with `default-features = false`. + let mut a = rand::rngs::SmallRng::seed_from_u64(42); + let mut b = rand::rngs::SmallRng::seed_from_u64(42); + + let mut prev_a: Option = None; + let mut prev_b: Option = None; + for ts in 0..5 { + let sa = Scalar::simulate(¶ms, prev_a.as_ref(), &mut a, ts); + let sb = Scalar::simulate(¶ms, prev_b.as_ref(), &mut b, ts); + assert_eq!(sa, sb, "same seed must yield the same walk"); + prev_a = Some(sa); + prev_b = Some(sb); + } } } diff --git a/aimdb-data-contracts/tests/linkable_derive.rs b/aimdb-data-contracts/tests/linkable_derive.rs new file mode 100644 index 00000000..93434067 --- /dev/null +++ b/aimdb-data-contracts/tests/linkable_derive.rs @@ -0,0 +1,49 @@ +//! Behavioral coverage for `#[derive(Linkable)]`. +//! +//! Lives under `tests/` (a separate crate), not `#[cfg(test)] mod tests` +//! inside `src/linkable.rs`: the derive emits absolute +//! `::aimdb_data_contracts::...` paths (matching the `migration_chain!` / +//! `RecordKey` precedent), which only resolve from outside the defining crate. + +#![cfg(feature = "linkable")] + +use aimdb_data_contracts::{Linkable, SchemaType}; +use serde::{Deserialize, Serialize}; + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, Linkable)] +struct DerivedTemperature { + celsius: f32, + timestamp: u64, +} + +impl SchemaType for DerivedTemperature { + const NAME: &'static str = "derived_temperature"; +} + +#[test] +fn roundtrips_through_json() { + let temp = DerivedTemperature { + celsius: 22.5, + timestamp: 1704326400000, + }; + + let bytes = temp.to_bytes().expect("serialization should succeed"); + let restored = DerivedTemperature::from_bytes(&bytes).expect("deserialization should succeed"); + + assert_eq!(temp, restored); +} + +#[test] +fn parses_hand_written_json() { + let json = br#"{"celsius": 25.0, "timestamp": 1704326400000}"#; + let temp = DerivedTemperature::from_bytes(json).expect("should parse valid JSON"); + + assert_eq!(temp.celsius, 25.0); + assert_eq!(temp.timestamp, 1704326400000); +} + +#[test] +fn rejects_invalid_json() { + let invalid = b"not valid json"; + assert!(DerivedTemperature::from_bytes(invalid).is_err()); +} diff --git a/aimdb-derive/CHANGELOG.md b/aimdb-derive/CHANGELOG.md index efd24cbe..fb619634 100644 --- a/aimdb-derive/CHANGELOG.md +++ b/aimdb-derive/CHANGELOG.md @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- **`#[derive(Linkable)]` proc-macro** (design 041 §3.3). Emits a JSON `aimdb_data_contracts::Linkable` impl (`serde_json::to_vec`/`from_slice` via the `__private` re-export, same foreign-crate-path pattern as `migration_chain!`) — replaces the hand-written `from_bytes`/`to_bytes` boilerplate every JSON-wire type repeated. `no_std + alloc` compatible; binary wire formats (e.g. KNX DPT codecs) still implement `Linkable` by hand. - **`migration_chain!` proc-macro.** Variable-arity replacement for the 3-arm `macro_rules!` previously hand-unrolled in `aimdb-data-contracts`; re-exported as `aimdb_data_contracts::migration_chain!` with the same grammar and call path. Generated dispatch is `O(N)` in code size regardless of chain length (one `__up_k`/`__down_k` helper per step). Emits foreign-crate paths into `aimdb_data_contracts` without depending on it (same pattern as `RecordKey` → `aimdb_core`) — a build-time-only dependency with no target/runtime/`no_std` impact. - **Tree-free version probe in `migrate_from_bytes`.** The generated dispatch no longer parses the payload into a full `serde_json::Value` tree to read the version field. It now scans a small `#[derive(serde::Deserialize)]` probe struct for just the version, then parses the same bytes a second time directly into the matched concrete type — peak allocation drops from O(payload tree) to O(concrete struct). `serde_json::Error::is_data()` preserves the existing `MissingVersion` vs `DeserializationFailed` split (missing/wrong-type field vs malformed JSON). Wire behavior is unchanged (same errors for missing/unknown versions). diff --git a/aimdb-derive/Cargo.toml b/aimdb-derive/Cargo.toml index 88353999..00d27d85 100644 --- a/aimdb-derive/Cargo.toml +++ b/aimdb-derive/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "aimdb-derive" -version = "0.2.0" +version = "0.3.0" edition.workspace = true authors.workspace = true license.workspace = true diff --git a/aimdb-derive/src/lib.rs b/aimdb-derive/src/lib.rs index 421fbd1e..d65a51b2 100644 --- a/aimdb-derive/src/lib.rs +++ b/aimdb-derive/src/lib.rs @@ -41,6 +41,62 @@ pub fn migration_chain(input: TokenStream) -> TokenStream { migration_chain::migration_chain(input) } +/// Derive `Linkable` (JSON codec) for a schema type. +/// +/// Emits `from_bytes`/`to_bytes` via `serde_json::from_slice`/`to_vec` — the +/// boilerplate every hand-written `impl Linkable` for a JSON-wire type repeats. +/// Requires `T: Serialize + DeserializeOwned` (a normal +/// compile error surfaces if it's missing) and the `linkable` feature of +/// `aimdb-data-contracts` (for the `Linkable` trait and its `__private::serde_json` +/// re-export). Binary formats (e.g. KNX DPT codecs) still implement `Linkable` +/// by hand — this derive is JSON-only. +/// +/// # Example +/// +/// Illustrative (not compiled: see the crate-level note — compiled +/// integration tests live in `aimdb-data-contracts`). +/// +/// ```rust,ignore +/// use aimdb_data_contracts::Linkable; +/// use serde::{Deserialize, Serialize}; +/// +/// #[derive(Clone, Debug, Serialize, Deserialize, Linkable)] +/// struct Temperature { +/// celsius: f32, +/// timestamp: u64, +/// } +/// ``` +#[proc_macro_derive(Linkable)] +pub fn derive_linkable(input: TokenStream) -> TokenStream { + let input = parse_macro_input!(input as DeriveInput); + let name = &input.ident; + + let expanded = quote! { + impl ::aimdb_data_contracts::Linkable for #name { + fn from_bytes( + data: &[u8], + ) -> Result { + ::aimdb_data_contracts::__private::serde_json::from_slice(data).map_err(|e| { + ::aimdb_data_contracts::__private::alloc::string::ToString::to_string(&e) + }) + } + + fn to_bytes( + &self, + ) -> Result< + ::aimdb_data_contracts::__private::alloc::vec::Vec, + ::aimdb_data_contracts::__private::alloc::string::String, + > { + ::aimdb_data_contracts::__private::serde_json::to_vec(self).map_err(|e| { + ::aimdb_data_contracts::__private::alloc::string::ToString::to_string(&e) + }) + } + } + }; + + expanded.into() +} + /// Derive the `RecordKey` trait for an enum /// /// Each variant must have a `#[key = "..."]` attribute specifying its string key. diff --git a/aimdb-sync/CHANGELOG.md b/aimdb-sync/CHANGELOG.md index a66e8eae..e37d2bf7 100644 --- a/aimdb-sync/CHANGELOG.md +++ b/aimdb-sync/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- **`SyncProducer::set_value`/`try_set_value`/`set_value_at`** (design 041 §3.4, feature `data-contracts`). `set()` took a fully constructed `T`, so every outside-the-thread caller hand-assembled the struct; `set_value(value)` constructs via `T::set(value, timestamp)` and sends in one call — blocking (`set_value`), non-blocking (`try_set_value`), or with an explicit timestamp for replay/testing (`set_value_at`). `set_value`/`try_set_value` stamp the caller's `SystemTime` (crate is std-only). New optional dependency: `aimdb-data-contracts` (feature `settable`), behind the new `data-contracts` feature — the contracts crate gains no `sync` feature (dependency direction unchanged). + ### Changed (breaking) - **Issue #131:** `AimDbSyncExt` extends the non-generic `aimdb_core::AimDb`; internal handles drop the `TokioAdapter` type parameter. @@ -60,7 +64,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 --- -[Unreleased]: https://github.com/aimdb-dev/aimdb/compare/v0.5.0...HEAD +[Unreleased]: https://github.com/aimdb-dev/aimdb/compare/v0.6.0...HEAD +[0.6.0]: https://github.com/aimdb-dev/aimdb/compare/v0.5.0...v0.6.0 [0.5.0]: https://github.com/aimdb-dev/aimdb/compare/v0.4.0...v0.5.0 [0.4.0]: https://github.com/aimdb-dev/aimdb/compare/v0.3.0...v0.4.0 [0.3.0]: https://github.com/aimdb-dev/aimdb/compare/v0.2.0...v0.3.0 diff --git a/aimdb-sync/Cargo.toml b/aimdb-sync/Cargo.toml index 671c113b..1bc8f3ed 100644 --- a/aimdb-sync/Cargo.toml +++ b/aimdb-sync/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "aimdb-sync" -version = "0.5.0" +version = "0.6.0" edition = "2021" authors.workspace = true license.workspace = true @@ -20,6 +20,13 @@ tokio = { version = "1.40", features = ["sync", "rt", "time", "macros"] } # Error handling thiserror = "1.0" +# Optional: Settable::set (feature `data-contracts`) for the `set_value` family. +# aimdb-sync depends on the contracts crate, never the reverse — the contracts +# crate has no `sync` feature. +aimdb-data-contracts = { path = "../aimdb-data-contracts", version = "0.3.0", optional = true, default-features = false, features = [ + "settable", +] } + # For type erasure and Any [dev-dependencies] # Testing dependencies @@ -33,6 +40,9 @@ default = [] # Enable tracing for debugging tracing = ["aimdb-core/tracing"] +# SyncProducer::set_value / try_set_value / set_value_at for Settable types +data-contracts = ["dep:aimdb-data-contracts"] + [package.metadata.docs.rs] all-features = true rustdoc-args = ["--cfg", "docsrs"] diff --git a/aimdb-sync/src/producer.rs b/aimdb-sync/src/producer.rs index 93efef13..e66c9f3c 100644 --- a/aimdb-sync/src/producer.rs +++ b/aimdb-sync/src/producer.rs @@ -220,6 +220,80 @@ where } } +/// Set-by-primitive verbs for `Settable` types (feature `data-contracts`). +/// +/// Where [`set`](Self::set) takes a fully constructed `T`, `set_value` +/// constructs it via `T::set(value, timestamp)` and sends, in one call. +/// Distinct from AimX's `record.set {name, value}` (full JSON value through +/// `JsonCodec`) — this is set-by-primitive. +#[cfg(feature = "data-contracts")] +impl SyncProducer +where + T: aimdb_data_contracts::Settable + Send + 'static + Debug + Clone, +{ + /// Construct via `T::set(value, now)` and send. Blocking, like [`set`](Self::set). + /// + /// Stamps with the *caller's* `SystemTime` (sample time at the edge), not + /// the engine's `ctx.time()` — use [`set_value_at`](Self::set_value_at) for + /// explicit-timestamp control (replay, testing). + /// + /// # Example + /// + /// ```no_run + /// # #[cfg(feature = "data-contracts")] + /// # fn main() -> Result<(), Box> { + /// use aimdb_core::AimDbBuilder; + /// use aimdb_data_contracts::{SchemaType, Settable}; + /// use aimdb_sync::AimDbBuilderSyncExt; + /// use aimdb_tokio_adapter::TokioAdapter; + /// use std::sync::Arc; + /// + /// #[derive(Debug, Clone)] + /// struct Temperature { celsius: f32, timestamp: u64 } + /// + /// impl SchemaType for Temperature { + /// const NAME: &'static str = "temperature"; + /// } + /// + /// impl Settable for Temperature { + /// type Value = f32; + /// fn set(value: f32, timestamp: u64) -> Self { + /// Temperature { celsius: value, timestamp } + /// } + /// } + /// + /// let handle = AimDbBuilder::new().runtime(Arc::new(TokioAdapter)).attach()?; + /// let producer = handle.producer::("temperature")?; + /// producer.set_value(22.5)?; // constructs Temperature::set(22.5, now_ms) and sends + /// # Ok(()) + /// # } + /// # #[cfg(not(feature = "data-contracts"))] + /// # fn main() {} + /// ``` + pub fn set_value(&self, value: T::Value) -> SyncResult<()> { + self.set(T::set(value, unix_now_ms())) + } + + /// Non-blocking variant, like [`try_set`](Self::try_set). + pub fn try_set_value(&self, value: T::Value) -> SyncResult<()> { + self.try_set(T::set(value, unix_now_ms())) + } + + /// Explicit-timestamp variant (replay, testing). + pub fn set_value_at(&self, value: T::Value, timestamp_ms: u64) -> SyncResult<()> { + self.set(T::set(value, timestamp_ms)) + } +} + +/// Current wall-clock time as Unix milliseconds (caller-side clock). +#[cfg(feature = "data-contracts")] +fn unix_now_ms() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as u64 +} + impl Clone for SyncProducer where T: Send + 'static + Debug + Clone, diff --git a/aimdb-sync/tests/settable_integration.rs b/aimdb-sync/tests/settable_integration.rs new file mode 100644 index 00000000..815885b1 --- /dev/null +++ b/aimdb-sync/tests/settable_integration.rs @@ -0,0 +1,140 @@ +//! Integration coverage for `SyncProducer::set_value` (feature `data-contracts`): +//! construct via `Settable::set`, produce, and consume end-to-end through the +//! real sync bridge. + +#![cfg(feature = "data-contracts")] + +use aimdb_core::{buffer::BufferCfg, AimDbBuilder}; +use aimdb_data_contracts::{SchemaType, Settable}; +use aimdb_sync::AimDbBuilderSyncExt; +use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; +use std::sync::Arc; +use std::thread; +use std::time::Duration; + +#[derive(Debug, Clone, PartialEq)] +struct Temperature { + celsius: f32, + timestamp: u64, +} + +impl SchemaType for Temperature { + const NAME: &'static str = "sync_settable.temperature"; +} + +impl Settable for Temperature { + type Value = f32; + + fn set(value: Self::Value, timestamp: u64) -> Self { + Temperature { + celsius: value, + timestamp, + } + } +} + +#[test] +fn set_value_constructs_produces_and_is_consumed() { + let adapter = Arc::new(TokioAdapter); + let mut builder = AimDbBuilder::new().runtime(adapter); + + builder.configure::("temperature", |reg| { + reg.buffer(BufferCfg::SpmcRing { capacity: 10 }) + .tap(|_ctx, _consumer| async move {}); + }); + + let handle = builder.attach().expect("failed to attach"); + let producer = handle + .producer::("temperature") + .expect("failed to create producer"); + let consumer = handle + .consumer::("temperature") + .expect("failed to create consumer"); + + producer.set_value(22.5).expect("set_value should succeed"); + + let received = consumer + .get_with_timeout(Duration::from_secs(2)) + .expect("failed to consume"); + + assert_eq!(received.celsius, 22.5); + assert!(received.timestamp > 0, "timestamp should be stamped"); + + handle.detach().expect("failed to detach"); +} + +#[test] +fn try_set_value_is_non_blocking_and_produces() { + let adapter = Arc::new(TokioAdapter); + let mut builder = AimDbBuilder::new().runtime(adapter); + + builder.configure::("temperature", |reg| { + reg.buffer(BufferCfg::SpmcRing { capacity: 10 }) + .tap(|_ctx, _consumer| async move {}); + }); + + let handle = builder.attach().expect("failed to attach"); + let producer = handle + .producer::("temperature") + .expect("failed to create producer"); + let consumer = handle + .consumer::("temperature") + .expect("failed to create consumer"); + + producer + .try_set_value(18.0) + .expect("try_set_value should succeed"); + + thread::sleep(Duration::from_millis(100)); + + let received = consumer + .get_with_timeout(Duration::from_secs(2)) + .expect("failed to consume"); + assert_eq!(received.celsius, 18.0); + + handle.detach().expect("failed to detach"); +} + +#[test] +fn set_value_at_stamps_the_explicit_timestamp() { + let adapter = Arc::new(TokioAdapter); + let mut builder = AimDbBuilder::new().runtime(adapter); + + builder.configure::("temperature", |reg| { + reg.buffer(BufferCfg::SpmcRing { capacity: 10 }) + .tap(|_ctx, _consumer| async move {}); + }); + + let handle = builder.attach().expect("failed to attach"); + let producer = handle + .producer::("temperature") + .expect("failed to create producer"); + let consumer = handle + .consumer::("temperature") + .expect("failed to create consumer"); + + // Explicit timestamp, not wall-clock — deterministic regardless of when the + // test runs (the replay/testing use case). + producer + .set_value_at(22.5, 1_700_000_000_000) + .expect("set_value_at should succeed"); + + let received = consumer + .get_with_timeout(Duration::from_secs(2)) + .expect("failed to consume"); + + assert_eq!( + received, + Temperature { + celsius: 22.5, + timestamp: 1_700_000_000_000, + } + ); +} + +#[test] +fn settable_set_is_a_pure_deterministic_constructor() { + let a = Temperature::set(22.5, 1_700_000_000_000); + let b = Temperature::set(22.5, 1_700_000_000_000); + assert_eq!(a, b); +} diff --git a/aimdb-wasm-adapter/Cargo.toml b/aimdb-wasm-adapter/Cargo.toml index 9209a880..9b516f70 100644 --- a/aimdb-wasm-adapter/Cargo.toml +++ b/aimdb-wasm-adapter/Cargo.toml @@ -35,7 +35,7 @@ aimdb-core = { version = "1.1.0", path = "../aimdb-core", default-features = fal aimdb-ws-protocol = { version = "0.1.0", path = "../aimdb-ws-protocol" } # Data contracts (alloc only — no std) -aimdb-data-contracts = { version = "0.2.0", path = "../aimdb-data-contracts", default-features = false, features = [ +aimdb-data-contracts = { version = "0.3.0", path = "../aimdb-data-contracts", default-features = false, features = [ "alloc", ] } diff --git a/aimdb-websocket-connector/Cargo.toml b/aimdb-websocket-connector/Cargo.toml index 7005c15e..fe7a47af 100644 --- a/aimdb-websocket-connector/Cargo.toml +++ b/aimdb-websocket-connector/Cargo.toml @@ -33,7 +33,7 @@ tracing = ["dep:tracing"] [dependencies] aimdb-core = { version = "1.1.0", path = "../aimdb-core", default-features = false } -aimdb-data-contracts = { version = "0.2.0", path = "../aimdb-data-contracts", default-features = false } +aimdb-data-contracts = { version = "0.3.0", path = "../aimdb-data-contracts", default-features = false } aimdb-ws-protocol = { version = "0.1.0", path = "../aimdb-ws-protocol" } # Async runtime diff --git a/docs/design/041-data-contracts-integration.md b/docs/design/041-data-contracts-integration.md new file mode 100644 index 00000000..5759d8b3 --- /dev/null +++ b/docs/design/041-data-contracts-integration.md @@ -0,0 +1,377 @@ +# 041 — Data contracts as first-class capabilities + +**Status:** ✅ Implemented. Follow-up to design-038 **D1/D12** (tracking issue [#161](https://github.com/aimdb-dev/aimdb/issues/161)). + +**Scope:** `aimdb-data-contracts` (trait reshapes + three registrar ext traits; `Simulatable` stays here behind its feature), `aimdb-sync` (consumes `Settable`), one small `aimdb-core` observability surface (signal gauges), the README capability table, the weather-mesh example, and a CI guard. + +**Goal:** every capability trait is a promise the engine keeps — implemented ⇒ one specific verb becomes available, and that verb is consumed by real code in the same PR (D1 rule). + +--- + +## 1. Problem + +Design 038 D1/D12: the capability traits were advertised as a headline feature, but core consumed almost none of them. Three specific gaps: + +1. **Simulation was a runtime value, not a compile-time fact.** `SimulationConfig.enabled` gated simulation with a runtime flag, so the sim loop, every `T::simulate` impl, `SimulationConfig`, and `rand` all shipped in the production image even with simulation "off". A build-time *value* selector (e.g. an env-var-driven mode enum) fails the same test: both arms still compile into the binary. The requirement is stronger — production binaries must contain **zero** simulation code, verifiable from the dependency graph. +2. **`Observable` was logging.** The README claimed "automatic per-record metrics", but the trait only powered a console tap (`format_log`). Rewording the claim downward would fix the dishonesty; making the trait actually feed metrics fixes the feature. +3. **`Streamable`/`Linkable`/`Settable` were unintuitive** because a user could not answer *"what does implementing this let me write?"* `Linkable` existed but every example still hand-wrote `with_deserializer` closures; `Settable` was built for `aimdb-sync` but sync never referenced it (`SyncProducer::set` took a full `T`); `Streamable` worked but nothing said its verb is the ws-connector's `.register::()`. + +## 2. Organizing principle: one verb per contract, tiered by deployment role + +A capability trait is a compile-time promise about a schema type. Each promise unlocks **exactly one verb**, and each contract belongs to a **tier** that states whether it may exist in a production binary: + +| Contract | Implement when… | Verb it unlocks | Consumer | Tier | +|---|---|---|---|---| +| `SchemaType` | always (identity) | `configure::(key, …)` | core | identity | +| `Linkable` | the type crosses a per-URL byte boundary (MQTT/KNX/serial/UDS) | `.linked_from(url)` / `.linked_to(url)` (§3.3) | connectors | wire (prod) | +| `Streamable` | the type streams as schema-named JSON (browser/WASM) | ws-connector `.register::()` | [`server/registry.rs:33`](../../aimdb-websocket-connector/src/server/registry.rs) | wire (prod) | +| `Migratable` | the schema evolved across versions | `migration_chain!` | core runtime migration (design 039) | wire (prod) | +| `Settable` | callers outside the AimDB thread set the record from a primitive | `SyncProducer::set_value(v)` (§3.4) | `aimdb-sync` | wire (prod) | +| `Observable` | the type carries a domain signal worth watching | `.observe()` → live signal metrics (§3.2) | introspection surface | introspection (prod, optional) | +| `Simulatable` | the type can generate realistic synthetic data | `.simulate(profile, rng)` (§3.1) | `simulatable` feature ext | **dev-only — never in prod** | + +Mechanism (the `.persist()` precedent — [`aimdb-persistence/src/ext.rs`](../../aimdb-persistence/src/ext.rs)): extension traits over `RecordRegistrar`/`SyncProducer` in the crate that owns the contract, installing plain `.source()`/`.tap()`/`.link_*()` stages. `.source()`/`.tap()` stay unbounded; `aimdb-core` never learns the contracts exist; dependency direction is always *contracts → core*, never the reverse. + +The tier column is the second half of the fix: *wire* contracts ship in production, *introspection* is prod-optional, and the *dev* tier must be **structurally excludable** — which drives §3.1's compile-time-only design. + +## 3. Design + +### 3.1 `Simulatable` → compile-time only, behind the `simulatable` feature + +#### 3.1.1 Stays in `aimdb-data-contracts`, as the dev-tier feature + +`Simulatable` lives in the contracts crate behind the `simulatable` feature. A separate `aimdb-simulation` crate was considered for the structural guarantee and rejected — another crate in the monorepo buys nothing, because the guarantee never depended on a crate boundary. It rests on three enforceable properties: + +1. **`simulatable` is not (and never becomes) a default feature** ([`Cargo.toml`](../../aimdb-data-contracts/Cargo.toml)); CI asserts it stays that way (§3.1.5). +2. **Production binaries resolve features per-package.** The workspace uses `resolver = "2"` ([`Cargo.toml:45`](../../Cargo.toml)), so `cargo build -p --release` unifies features only over that binary's own graph — a sim-enabled example elsewhere in the workspace cannot leak the feature into the release artifact. Corollary, worth one sentence in CONTRIBUTING: release artifacts are built per-package, never lifted out of a `--workspace` build (where unification does cross targets). +3. **`rand` is the tracer.** `Simulatable::simulate` makes `rand` reachable iff the feature is on, so an inverse-dependency check on `rand` proves sim absence from a production graph (§3.1.5). + +Feature wiring: `simulatable = ["rand", "aimdb-core"]` — the ext trait needs `RecordRegistrar`/`Producer`/`RuntimeContext`, the same core coupling `observable` has and `linkable` gains in §3.3. `rand` is `default-features = false` without `std_rng`: the RNG is always caller-supplied (§5.2). The feature is `no_std`-compatible: the "develop against sim, flash against silicon" story is embedded-first. + +#### 3.1.2 Trait reshape: domain params, no runtime gate + +The previous `SimulationParams` (base/variation/trend/step) was a random-walk config pretending to be universal, and `SimulationConfig.enabled` was a runtime gate that contradicts the compile-time stance. Their replacement: + +```rust +// aimdb-data-contracts/src/simulatable.rs (feature = "simulatable") + +/// Dev-only capability: generate realistic synthetic data. +pub trait Simulatable: SchemaType { + /// Type-specific generation parameters (a temperature defines walk bounds, + /// a GPS track defines waypoints, …). + type Params: Clone + Send + Sync + Default + 'static; + + /// Generate the next sample. `previous` enables walks/trends; + /// `timestamp_ms` is Unix millis supplied by the driving loop. + fn simulate( + params: &Self::Params, + previous: Option<&Self>, + rng: &mut R, + timestamp_ms: u64, + ) -> Self; +} + +/// Loop policy + generation params for one record. +#[derive(Clone, Debug, Default)] +pub struct SimProfile

{ + /// Interval between samples (milliseconds). + pub interval_ms: u64, + pub params: P, +} + +/// Off-the-shelf `Params` for scalar random walks. Impls written against the +/// old `SimulationParams` migrate by setting `type Params = RandomWalkParams`. +#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] +pub struct RandomWalkParams { + pub base: f64, + pub variation: f64, + pub trend: f64, + pub step: f64, +} +``` + +There is **no `enabled` field**. "Sim but off" is not a state: production excludes the feature (§3.1.4), and a std dev binary that wants sim conditionally simply branches before calling `.simulate()`. + +#### 3.1.3 `SimulatableRegistrarExt::simulate(profile, rng)` + +The verb installs a plain `.source()` loop over the reshaped types: + +```rust +pub trait SimulatableRegistrarExt<'a, T> +where + T: Simulatable + Send + Sync + Clone + 'static, +{ + /// Install a source that emits `T::simulate(...)` every `interval_ms`, + /// driving the caller-supplied RNG (OS entropy on std, seeded PRNG on + /// no_std, fixed seed in tests). + fn simulate(&mut self, profile: SimProfile, rng: R) -> &mut RecordRegistrar<'a, T> + where + R: rand::Rng + Send + 'static; +} + +impl<'a, T> SimulatableRegistrarExt<'a, T> for RecordRegistrar<'a, T> +where + T: Simulatable + Send + Sync + Clone + 'static, +{ + fn simulate(&mut self, profile: SimProfile, mut rng: R) -> &mut RecordRegistrar<'a, T> + where + R: rand::Rng + Send + 'static, + { + self.source(move |ctx, producer| async move { + let mut prev: Option = None; + loop { + let now_ms = ctx + .time() + .unix_time() + .map(|(s, ns)| s.saturating_mul(1000) + (ns / 1_000_000) as u64) + .unwrap_or(0); + let next = T::simulate(&profile.params, prev.as_ref(), &mut rng, now_ms); + producer.produce(next.clone()); + prev = Some(next); + ctx.time().sleep_millis(profile.interval_ms.max(1)).await; + } + }) + .with_name("simulate") + } +} +``` + +Runtime-neutral (`ctx.time()` for clock and delay, like `.persist()`'s cleanup loop), installs a **source** so writer-exclusivity is enforced by `build()` unchanged ([`typed_record.rs:570`](../../aimdb-core/src/typed_record.rs)). + +#### 3.1.4 Sim-to-real is a `#[cfg]`, not an API + +There is deliberately no selector API — no mode enum, no `source_or_simulate(mode, …)` wrapper. Any selector that takes the choice as a *value* (even one read from the environment at build time) compiles both arms into the binary, which is exactly what the dev tier forbids. The sim-to-real selection is the app's Cargo feature, and that pattern is canonical rather than wrapped: + +```toml +# app Cargo.toml +[features] +sim = ["aimdb-data-contracts/simulatable", "weather-mesh-common/sim"] +``` + +```rust +// app wiring — the record configuration is identical either way +builder.configure::(KEY, |reg| { + #[cfg(feature = "sim")] + reg.simulate(profile, rng); + #[cfg(not(feature = "sim"))] + reg.source(read_hardware); + + reg.buffer_cfg(BufferCfg::SingleLatest); +}); +``` + +The app's contracts crate gates its impls the same way (`#[cfg(feature = "sim")] impl Simulatable for Temperature { … }`). With `sim` off there is nothing to audit: no impls, no `SimProfile`, no `rand` anywhere in the binary's graph. Exactly one producer is installed on either path, so single-writer-per-key holds by construction, with the selection resolved by the compiler instead of a value. + +#### 3.1.5 CI guard: prove production is sim-free + +A CI step (Makefile target `check-no-sim`) builds the production configuration of each example that has a `sim` feature and asserts the exclusion from the resolved graph — `cargo tree -p -e normal -i rand` must report no matching package when `sim` is off (`rand` is the tracer, §3.1.1) — and additionally asserts that `simulatable` never appears in `aimdb-data-contracts`' default features. This turns "never ships in production" from a convention into a gate. + +### 3.2 `Observable` → the signal, not the log line + +#### 3.2.1 Trait reshape: keep the kernel, drop the cosmetics + +The trait's valuable kernel is the numeric projection (`Signal`, `signal()`, `UNIT`); `ICON` and `format_log()` were presentation and made the whole trait read as a log formatter. The kernel-only trait: + +```rust +pub trait Observable: SchemaType { + /// Numeric type of the domain signal. + type Signal: PartialOrd + Copy; + + /// What the signal means, for metrics/UI labels (e.g. "celsius"). + /// Defaults to the schema name. + const SIGNAL: &'static str = Self::NAME; + + /// Unit label (e.g. "°C", "%", "hPa"). + const UNIT: &'static str = ""; + + /// Extract the signal from this instance. + fn signal(&self) -> Self::Signal; +} +``` + +`ICON` and `format_log()` are gone: the logging tap (§3.2.3) formats from `Debug` + `UNIT`, and demo-specific decoration (emoji) belongs to the demo. + +#### 3.2.2 `.observe()` installs a signal-metrics tap + +`.observe()` folds `signal()` into per-record **signal stats** (last/min/max/count/mean as `f64`) that surface on the existing introspection paths. Mechanism — a small, feature-honest core addition mirroring `with_name`: + +- `aimdb-core` provides `RecordRegistrar::signal_gauge(name: &'static str, unit: &'static str) -> SignalGaugeHandle`. Under the `observability` feature it registers atomic `SignalStats` on the record's profiling state and returns a live handle; without the feature it is still callable but returns an inert handle (the `with_name` precedent — callers never `cfg` on core's features). +- Exposure: `RecordMetadata` gains an optional signal-stats field, and the AimX `record.get` response carries it — so the signal shows up in `record.list`/`record.get`, the MCP tools built on them, and stage-profiling output. *Implement `Observable`, call `.observe()`, and your domain signal is live on every introspection surface.* That is the README claim, and the gauge is what makes it true. + +```rust +// aimdb-data-contracts/src/observable.rs (feature = "observable") +pub trait ObservableRegistrarExt<'a, T> +where + T: Observable + Send + Sync + Clone + 'static, + T::Signal: Into, +{ + /// Feed `T::signal()` into the record's signal gauge (last/min/max/mean), + /// visible via record.list / record.get / stage profiling. + fn observe(&mut self) -> &mut RecordRegistrar<'a, T>; +} + +impl<'a, T> ObservableRegistrarExt<'a, T> for RecordRegistrar<'a, T> +where + T: Observable + Send + Sync + Clone + 'static, + T::Signal: Into, +{ + fn observe(&mut self) -> &mut RecordRegistrar<'a, T> { + let gauge = self.signal_gauge(T::SIGNAL, T::UNIT); + self.tap(move |_ctx, consumer| async move { + let mut reader = consumer.subscribe(); + while let Ok(value) = reader.recv().await { + gauge.update(value.signal().into()); + } + }) + .with_name("observe") + } +} +``` + +Notes: +- `T::Signal: Into` is bounded **at the consumption site**, not in the trait — `f32`/`i32`/`u32` signals qualify; a type with an exotic signal can still implement `Observable` and write its own tap. +- `.observe()` takes **no `node_id`** — the record key already identifies the gauge; `node_id` was a logging concern. + +#### 3.2.3 Logging stays, honestly named + +`log_tap` remains for humans watching a console, exposed as `.log(node_id)` on the same ext trait (formats `[node_id] key: {signal}{UNIT} {value:?}` from `Debug` — no trait items needed). The README row for `Observable` says metrics; logging is a bullet under it, not the definition. + +### 3.3 `Linkable` → the default codec that `.link_*` actually consumes + +Before this design, implementing `Linkable` changed nothing — every example hand-wired `with_deserializer`/`with_serializer` closures. The fix: one-line link verbs, ext trait in the contracts crate (dependency direction preserved): + +```rust +// aimdb-data-contracts/src/linkable.rs (feature = "linkable") +pub trait LinkableRegistrarExt<'a, T> +where + T: Linkable + Send + Sync + Clone + core::fmt::Debug + 'static, +{ + /// `.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`. + fn linked_to(&mut self, url: &str) -> &mut RecordRegistrar<'a, T>; +} + +impl<'a, T> LinkableRegistrarExt<'a, T> for RecordRegistrar<'a, T> +where + T: Linkable + Send + Sync + Clone + core::fmt::Debug + 'static, +{ + fn linked_from(&mut self, url: &str) -> &mut RecordRegistrar<'a, T> { + self.link_from(url) + .with_deserializer(|_ctx, bytes| T::from_bytes(bytes)) + .finish() + } + + fn linked_to(&mut self, url: &str) -> &mut RecordRegistrar<'a, T> { + self.link_to(url) + .with_serializer(|_ctx, value| { + value.to_bytes().map_err(|_| SerializeError::InvalidData) + }) + .finish() + } +} +``` + +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. +- **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. +- **Coupling:** the `linkable` feature gains an `aimdb-core` dependency for the ext trait (`default-features = false, features = ["alloc"]`, same wiring as `observable`). +- **Three serialization stories, stated once** so users stop guessing: + + | Boundary | Mechanism | Contract | + |---|---|---| + | Schema-named JSON streaming (browser/WASM) | ws registry keyed by `SchemaType::NAME` | `Streamable` | + | 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.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**: + +```rust +// aimdb-sync/src/producer.rs (inherent impl, feature = "data-contracts") +impl SyncProducer +where + T: aimdb_data_contracts::Settable + Send + 'static + Debug + Clone, +{ + /// Construct via `T::set(value, now)` and send. Blocking, like `set()`. + pub fn set_value(&self, value: T::Value) -> SyncResult<()> { + self.set(T::set(value, unix_now_ms())) + } + + /// Non-blocking variant, like `try_set()`. + pub fn try_set_value(&self, value: T::Value) -> SyncResult<()> { + self.try_set(T::set(value, unix_now_ms())) + } + + /// Explicit-timestamp variant (replay, testing). + pub fn set_value_at(&self, value: T::Value, timestamp_ms: u64) -> SyncResult<()> { + self.set(T::set(value, timestamp_ms)) + } +} +``` + +Notes: +- **Dependency direction:** `aimdb-sync` gains `aimdb-data-contracts = { optional = true, default-features = false, features = ["settable"] }` behind a `data-contracts` feature. The contracts crate does **not** grow a `sync` feature. +- **Clock semantics (documented, one sentence):** `set_value` stamps with the *caller's* `SystemTime` (sample time at the edge), not the engine's `ctx.time()`; `set_value_at` exists for callers that need control. `aimdb-sync` is std-only, so `SystemTime` is always available. +- **Single-writer untouched:** `set_value` flows through the same channel as `set()`; the sync bridge remains the record's one producer, and concurrent `SyncProducer` clones already arbitrate through that one request stream. +- **Feature gate:** `Settable` sits behind a `settable` feature for tier symmetry with the other wire contracts (codegen templates that emit `impl Settable` enable it). +- A future AimX **set-by-primitive** verb (`aimdb set temperature 22.5` from CLI/MCP) becomes a second consumer of the same trait — recorded in §7, not in scope. The read/write symmetry is explicit: `Observable::signal()` is the generic *read* projection, `Settable::set()` the generic *write* injection. + +### 3.5 `Streamable` — unchanged + +Already consumed (ws-connector registry bound). Its fix is the verb table (§2) and the serialization-stories table (§3.3): the README states its verb is `.register::()` on the ws connector builder. No trait change. + +### 3.6 `Migratable` — unchanged + +Consumed since design 039, works no_std. Stretch (recorded, not in scope): auto-apply the migration chain on the `linked_from` deserialize path so connectors migrate old wire data without a manual call. + +## 4. Crate & feature layout + +``` +aimdb-data-contracts + features: linkable (+core), 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) + +aimdb-sync + features: data-contracts → aimdb-data-contracts (nf, settable) + +aimdb-core + observability: + SignalStats / RecordRegistrar::signal_gauge (inert handle when off) + RecordMetadata: + optional signal stats; AimX record.get carries them +``` + +`aimdb-data-contracts` bumps 0.2.0 → 0.3.0 (breaking: `Simulatable` reshaped — `type Params`, `SimulationConfig`/`SimulationParams` replaced by `SimProfile`/`RandomWalkParams`; `Observable` loses `ICON`/`format_log`; `Settable` gains a feature gate). + +## 5. Decisions + +1. **Simulation is compile-time.** No runtime `enabled` flag and no selector API (a `source_or_simulate(mode, …)` wrapper was considered and rejected — both arms compile into the binary); the `#[cfg]` branch is the canonical sim-to-real pattern, enforced by the CI guard (`rand` tracer + never-a-default-feature assert). `Simulatable` **stays in `aimdb-data-contracts`** behind `simulatable` — a separate `aimdb-simulation` crate was rejected to keep the monorepo lean; the guarantee rests on §3.1.1's three properties, not on a crate boundary. +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). +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.** + +## 6. Migration & examples (the D1 callers) + +- **weather-station-beta/gamma:** the manual RNG-loop producers are deleted; `reg.simulate(SimProfile { interval_ms, params: RandomWalkParams { … } }, StdRng::…)` under the app's `sim` feature; contracts impls use `type Params = RandomWalkParams` and `#[cfg(feature = "sim")]`. The stations are the §3.1.4 pattern's reference implementation (with the CI guard building their prod configuration). +- **weather-hub:** `.observe()` replaces the manual `log_tap` tap; one `.log(node_id)` stays where console output is the point of the demo. The hub verifies the signal appears in `record.list`/`record.get`. +- **one connector demo (tokio-mqtt):** `#[derive(Linkable)]` + `.linked_from`/`.linked_to` replace the hand-written closures. +- **aimdb-sync doctest/integration test:** `producer.set_value(22.5)` end-to-end (construct → produce → consume), plus `set_value_at` determinism in a unit test. +- **README:** the capability table is the §2 verb table (Contract / Implement when / Verb / Tier); the `observability` feature keeps the buffer-statistics story; the `simulatable` feature gets a "dev tier — never ships" call-out with the `#[cfg]` pattern. +- **CHANGELOG:** entries for all four crates touched. + +## 7. Work items + +One branch/PR; each work item lands as its own commit carrying its caller: + +1. `Simulatable` reshape (trait + `SimProfile`/`RandomWalkParams` + `.simulate()` ext, `enabled` and `std_rng` dropped), stations migrated, CI `check-no-sim` guard. +2. Core `signal_gauge` surface + `Observable` kernel reshape + `.observe()`/`.log()` ext, hub migrated, README `Observable` row. +3. `LinkableRegistrarExt` + `#[derive(Linkable)]`, mqtt demo migrated. +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). diff --git a/examples/mqtt-connector-demo-common/Cargo.toml b/examples/mqtt-connector-demo-common/Cargo.toml index b4a8fde9..cb583cc2 100644 --- a/examples/mqtt-connector-demo-common/Cargo.toml +++ b/examples/mqtt-connector-demo-common/Cargo.toml @@ -15,6 +15,12 @@ derive = ["aimdb-core/derive"] # needed: the no_std JSON paths are hand-rolled), so an embassy demo can expose # these records over a remote-access (serial) server. serde = ["dep:serde"] +# `SchemaType` + `#[derive(Linkable)]`. `no_std + alloc` +# compatible (aimdb-data-contracts's `linkable` feature only needs `alloc` — +# its `serde_json` dep is alloc-only, no `std`). Not enabled by default: the +# embassy demo's hand-rolled no_std JSON path is an intentional illustration, +# left untouched; only the tokio demo opts into this feature. +data-contracts = ["alloc", "serde", "dep:aimdb-data-contracts"] # Embassy-specific features (no_std) defmt = ["dep:defmt"] @@ -25,6 +31,10 @@ aimdb-core = { path = "../../aimdb-core", default-features = false } # Serialization (std only) serde = { workspace = true, features = ["derive"], optional = true } serde_json = { workspace = true, optional = true } +# Optional: SchemaType + #[derive(Linkable)] (feature `data-contracts`, no_std + alloc) +aimdb-data-contracts = { path = "../../aimdb-data-contracts", default-features = false, optional = true, features = [ + "linkable", +] } # Optional defmt for embedded logging defmt = { workspace = true, optional = true } diff --git a/examples/mqtt-connector-demo-common/src/types.rs b/examples/mqtt-connector-demo-common/src/types.rs index a1c3fac4..76df37bf 100644 --- a/examples/mqtt-connector-demo-common/src/types.rs +++ b/examples/mqtt-connector-demo-common/src/types.rs @@ -9,6 +9,9 @@ use alloc::string::String; #[cfg(feature = "serde")] use serde::{Deserialize, Serialize}; +#[cfg(feature = "data-contracts")] +use aimdb_data_contracts::{Linkable, SchemaType}; + // ============================================================================ // TEMPERATURE READING // ============================================================================ @@ -18,6 +21,7 @@ use serde::{Deserialize, Serialize}; /// Represents a temperature measurement that can be published to MQTT. #[derive(Clone, Debug)] #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))] +#[cfg_attr(feature = "data-contracts", derive(Linkable))] pub struct Temperature { /// Sensor identifier (e.g., "indoor-001", "outdoor-001") pub sensor_id: String, @@ -25,6 +29,11 @@ pub struct Temperature { pub celsius: f32, } +#[cfg(feature = "data-contracts")] +impl SchemaType for Temperature { + const NAME: &'static str = "mqtt_demo.temperature"; +} + impl Temperature { /// Create a new temperature reading pub fn new(sensor_id: &str, celsius: f32) -> Self { @@ -81,6 +90,7 @@ impl Temperature { /// Command for controlling temperature sensors (inbound from MQTT) #[derive(Clone, Debug)] #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))] +#[cfg_attr(feature = "data-contracts", derive(Linkable))] pub struct TemperatureCommand { /// Action to perform (e.g., "read", "calibrate", "reset") pub action: String, @@ -88,6 +98,11 @@ pub struct TemperatureCommand { pub sensor_id: String, } +#[cfg(feature = "data-contracts")] +impl SchemaType for TemperatureCommand { + const NAME: &'static str = "mqtt_demo.temperature_command"; +} + impl TemperatureCommand { /// Create a new temperature command pub fn new(action: &str, sensor_id: &str) -> Self { diff --git a/examples/readme-quickstart/src/main.rs b/examples/readme-quickstart/src/main.rs index 7bae8ab6..b136040e 100644 --- a/examples/readme-quickstart/src/main.rs +++ b/examples/readme-quickstart/src/main.rs @@ -29,10 +29,7 @@ async fn main() -> Result<(), Box> { }); }); - // `.run()` builds the database, collects every producer/consumer/transform - // future, and drives them all on a single `FuturesUnordered`. It blocks - // until shutdown. For programmatic access to the `AimDb` handle, call - // `.build().await?` directly — it returns `(AimDb, AimDbRunner)`. + // Build the db and drive every source/tap future until shutdown. builder.run().await?; Ok(()) } diff --git a/examples/tokio-mqtt-connector-demo/Cargo.toml b/examples/tokio-mqtt-connector-demo/Cargo.toml index ec64d16a..eb5fd4b0 100644 --- a/examples/tokio-mqtt-connector-demo/Cargo.toml +++ b/examples/tokio-mqtt-connector-demo/Cargo.toml @@ -23,6 +23,12 @@ aimdb-tokio-adapter = { path = "../../aimdb-tokio-adapter", features = [ mqtt-connector-demo-common = { path = "../mqtt-connector-demo-common", features = [ "std", "derive", + "data-contracts", +] } + +# LinkableRegistrarExt / #[derive(Linkable)] verb +aimdb-data-contracts = { path = "../../aimdb-data-contracts", features = [ + "linkable", ] } # MQTT connector diff --git a/examples/tokio-mqtt-connector-demo/src/main.rs b/examples/tokio-mqtt-connector-demo/src/main.rs index d6dfbb01..28b64ec6 100644 --- a/examples/tokio-mqtt-connector-demo/src/main.rs +++ b/examples/tokio-mqtt-connector-demo/src/main.rs @@ -36,6 +36,7 @@ use aimdb_core::buffer::BufferCfg; use aimdb_core::{AimDbBuilder, DbResult, Producer, RecordKey, RuntimeContext}; +use aimdb_data_contracts::LinkableRegistrarExt; use aimdb_mqtt_connector::{MqttLinkExt, MqttOutboundLinkExt}; use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; use std::sync::Arc; @@ -139,9 +140,9 @@ async fn main() -> DbResult<()> { ); // Temperature sensors (outbound to MQTT) - using link_address() from key metadata. - // The MQTT knobs come from the connector crate's MqttLinkExt/ - // MqttOutboundLinkExt extension traits; QoS 1 / no - // retain matches the connector defaults. + // TempIndoor needs per-link QoS/retain, so it stays on the raw builder (the + // escape hatch for per-link options); TempOutdoor/TempServerRoom need no + // extra options, so `.linked_to()` (from `#[derive(Linkable)]`) suffices. builder.configure::(SensorKey::TempIndoor, |reg| { reg.buffer(BufferCfg::SpmcRing { capacity: 10 }) .source(indoor_temp_producer) @@ -156,22 +157,20 @@ async fn main() -> DbResult<()> { builder.configure::(SensorKey::TempOutdoor, |reg| { reg.buffer(BufferCfg::SpmcRing { capacity: 10 }) .source(outdoor_temp_producer) - .tap(temperature_logger) - .link_to(SensorKey::TempOutdoor.link_address().unwrap()) - .with_serializer(|_ctx, temp: &Temperature| Ok(temp.to_json_vec())) - .finish(); + .tap(temperature_logger); + reg.linked_to(SensorKey::TempOutdoor.link_address().unwrap()); }); builder.configure::(SensorKey::TempServerRoom, |reg| { reg.buffer(BufferCfg::SpmcRing { capacity: 10 }) .source(server_room_temp_producer) - .tap(temperature_logger) - .link_to(SensorKey::TempServerRoom.link_address().unwrap()) - .with_serializer(|_ctx, temp: &Temperature| Ok(temp.to_json_vec())) - .finish(); + .tap(temperature_logger); + reg.linked_to(SensorKey::TempServerRoom.link_address().unwrap()); }); - // Command consumers (inbound from MQTT) - using link_address() from key metadata + // Command consumers (inbound from MQTT) - using link_address() from key metadata. + // TempIndoor needs subscribe QoS, so it stays on the raw builder; TempOutdoor + // uses `.linked_from()`. builder.configure::(CommandKey::TempIndoor, |reg| { reg.buffer(BufferCfg::SpmcRing { capacity: 10 }) .tap(command_consumer) @@ -183,10 +182,8 @@ async fn main() -> DbResult<()> { builder.configure::(CommandKey::TempOutdoor, |reg| { reg.buffer(BufferCfg::SpmcRing { capacity: 10 }) - .tap(command_consumer) - .link_from(CommandKey::TempOutdoor.link_address().unwrap()) - .with_deserializer(|_ctx, data: &[u8]| TemperatureCommand::from_json(data)) - .finish(); + .tap(command_consumer); + reg.linked_from(CommandKey::TempOutdoor.link_address().unwrap()); }); println!("Subscribe: mosquitto_sub -h localhost -t 'sensors/#' -v"); diff --git a/examples/weather-mesh-demo/weather-hub/Cargo.toml b/examples/weather-mesh-demo/weather-hub/Cargo.toml index bec2f6d3..73bb9c1a 100644 --- a/examples/weather-mesh-demo/weather-hub/Cargo.toml +++ b/examples/weather-mesh-demo/weather-hub/Cargo.toml @@ -16,7 +16,11 @@ weather-mesh-common = { path = "../weather-mesh-common", features = [ "migratable", ] } aimdb-core = { path = "../../../aimdb-core", features = ["derive"] } -aimdb-tokio-adapter = { path = "../../../aimdb-tokio-adapter" } +# `observability` makes `.observe()`'s signal gauge live; without it core hands +# back an inert handle and no signal stats surface on record.list/record.get. +aimdb-tokio-adapter = { path = "../../../aimdb-tokio-adapter", features = [ + "observability", +] } aimdb-data-contracts = { path = "../../../aimdb-data-contracts", features = [ "linkable", "observable", diff --git a/examples/weather-mesh-demo/weather-hub/src/main.rs b/examples/weather-mesh-demo/weather-hub/src/main.rs index 20c3b397..94aefd8f 100644 --- a/examples/weather-mesh-demo/weather-hub/src/main.rs +++ b/examples/weather-mesh-demo/weather-hub/src/main.rs @@ -1,5 +1,5 @@ use aimdb_core::{buffer::BufferCfg, AimDbBuilder, RecordKey}; -use aimdb_data_contracts::{log_tap, Linkable}; +use aimdb_data_contracts::{Linkable, ObservableRegistrarExt}; use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; use weather_mesh_common::{Humidity, HumidityKey, TempKey, Temperature}; @@ -32,9 +32,12 @@ async fn main() -> aimdb_core::DbResult<()> { let topic = key.link_address().unwrap().to_string(); builder.configure::(key, |reg| { - reg.buffer(BufferCfg::SpmcRing { capacity: 100 }) - .tap(move |ctx, consumer| log_tap(ctx, consumer, key.as_str())) - .link_from(&topic) + reg.buffer(BufferCfg::SpmcRing { capacity: 100 }); + // Metrics: fold °C into the record's signal gauge (record.list/get). + reg.observe(); + // Console: one human-readable log line per value (the demo's point). + reg.log(key.as_str()); + reg.link_from(&topic) .with_deserializer(|ctx, data: &[u8]| { ctx.log() .debug("Deserializing temperature from MQTT payload"); @@ -55,9 +58,10 @@ async fn main() -> aimdb_core::DbResult<()> { let topic = key.link_address().unwrap().to_string(); builder.configure::(key, |reg| { - reg.buffer(BufferCfg::SpmcRing { capacity: 100 }) - .tap(move |ctx, consumer| log_tap(ctx, consumer, key.as_str())) - .link_from(&topic) + reg.buffer(BufferCfg::SpmcRing { capacity: 100 }); + // Metrics only for humidity; temperature keeps the console `.log()`. + reg.observe(); + reg.link_from(&topic) .with_deserializer(|ctx, data: &[u8]| { ctx.log().debug("Deserializing humidity from MQTT payload"); let humidity = Humidity::from_bytes(data)?; diff --git a/examples/weather-mesh-demo/weather-mesh-common/Cargo.toml b/examples/weather-mesh-demo/weather-mesh-common/Cargo.toml index 13ed7d06..6a8d694e 100644 --- a/examples/weather-mesh-demo/weather-mesh-common/Cargo.toml +++ b/examples/weather-mesh-demo/weather-mesh-common/Cargo.toml @@ -18,9 +18,11 @@ aimdb-core = { path = "../../../aimdb-core", default-features = false, features "derive", "alloc", ] } -aimdb-data-contracts = { path = "../../../aimdb-data-contracts", default-features = false } +aimdb-data-contracts = { path = "../../../aimdb-data-contracts", default-features = false, features = [ + "settable", +] } serde = { version = "1.0", default-features = false, features = ["derive"] } serde_json = { workspace = true, optional = true } -rand = { version = "0.10.1", optional = true, default-features = false, features = [ - "std_rng", -] } +# RNG is caller-supplied at the station; the contract impls only use the +# `Rng`/`RngExt` traits, so no `std_rng` here. +rand = { version = "0.10.1", optional = true, default-features = false } diff --git a/examples/weather-mesh-demo/weather-mesh-common/src/contracts/dew_point.rs b/examples/weather-mesh-demo/weather-mesh-common/src/contracts/dew_point.rs index d6124aa0..c24dbb2c 100644 --- a/examples/weather-mesh-demo/weather-mesh-common/src/contracts/dew_point.rs +++ b/examples/weather-mesh-demo/weather-mesh-common/src/contracts/dew_point.rs @@ -33,23 +33,11 @@ impl Streamable for DewPoint {} impl Observable for DewPoint { type Signal = f32; - const ICON: &'static str = "🌫️"; const UNIT: &'static str = "°C"; fn signal(&self) -> f32 { self.celsius } - - fn format_log(&self, node_id: &str) -> alloc::string::String { - alloc::format!( - "{} [{}] DewPoint: {:.1}{} at {}", - Self::ICON, - node_id, - self.celsius, - Self::UNIT, - self.timestamp - ) - } } #[cfg(feature = "linkable")] diff --git a/examples/weather-mesh-demo/weather-mesh-common/src/contracts/humidity.rs b/examples/weather-mesh-demo/weather-mesh-common/src/contracts/humidity.rs index 43437a15..87e5c01b 100644 --- a/examples/weather-mesh-demo/weather-mesh-common/src/contracts/humidity.rs +++ b/examples/weather-mesh-demo/weather-mesh-common/src/contracts/humidity.rs @@ -9,7 +9,7 @@ use serde::{Deserialize, Serialize}; use aimdb_data_contracts::Linkable; #[cfg(feature = "simulatable")] -use aimdb_data_contracts::{Simulatable, SimulationConfig}; +use aimdb_data_contracts::{RandomWalkParams, Simulatable}; #[cfg(feature = "simulatable")] use rand::RngExt; @@ -30,44 +30,34 @@ impl Streamable for Humidity {} impl Observable for Humidity { type Signal = f32; - const ICON: &'static str = "💧"; const UNIT: &'static str = "%"; fn signal(&self) -> f32 { self.percent } - - fn format_log(&self, node_id: &str) -> alloc::string::String { - alloc::format!( - "{} [{}] Humidity: {:.1}{} at {}", - Self::ICON, - node_id, - self.percent, - Self::UNIT, - self.timestamp - ) - } } #[cfg(feature = "simulatable")] impl Simulatable for Humidity { + type Params = RandomWalkParams; + /// Simulate humidity readings with random walk behavior. /// - /// # Config params interpretation + /// # Params interpretation /// - `base`: Center humidity value (default: 50.0%) /// - `variation`: Maximum deviation from base (default: 10.0%) /// - `step`: Random walk step multiplier (default: 0.2) /// - `trend`: Linear trend per sample (default: 0.0) fn simulate( - config: &SimulationConfig, + params: &Self::Params, previous: Option<&Self>, rng: &mut R, - timestamp: u64, + timestamp_ms: u64, ) -> Self { - let base = config.params.base as f32; - let variation = config.params.variation as f32; - let step = config.params.step as f32; - let trend = config.params.trend as f32; + let base = params.base as f32; + let variation = params.variation as f32; + let step = params.step as f32; + let trend = params.trend as f32; // Random walk: small delta from previous value, clamped to valid range let current = match previous { @@ -82,7 +72,7 @@ impl Simulatable for Humidity { Humidity { percent: current, - timestamp, + timestamp: timestamp_ms, } } } diff --git a/examples/weather-mesh-demo/weather-mesh-common/src/contracts/location.rs b/examples/weather-mesh-demo/weather-mesh-common/src/contracts/location.rs index 5938e41f..94926f1d 100644 --- a/examples/weather-mesh-demo/weather-mesh-common/src/contracts/location.rs +++ b/examples/weather-mesh-demo/weather-mesh-common/src/contracts/location.rs @@ -9,7 +9,7 @@ use serde::{Deserialize, Serialize}; use aimdb_data_contracts::Linkable; #[cfg(feature = "simulatable")] -use aimdb_data_contracts::{Simulatable, SimulationConfig}; +use aimdb_data_contracts::{RandomWalkParams, Simulatable}; #[cfg(feature = "simulatable")] use rand::RngExt; @@ -41,56 +41,35 @@ impl Observable for GpsLocation { /// Ordering is lexicographic (lat first, then lon). type Signal = (f64, f64); - const ICON: &'static str = "📍"; const UNIT: &'static str = "°"; fn signal(&self) -> Self::Signal { (self.latitude, self.longitude) } - - fn format_log(&self, node_id: &str) -> alloc::string::String { - let alt_str = match self.altitude { - Some(alt) => alloc::format!(" alt={:.1}m", alt), - None => alloc::string::String::new(), - }; - let acc_str = match self.accuracy { - Some(acc) => alloc::format!(" acc=±{:.1}m", acc), - None => alloc::string::String::new(), - }; - alloc::format!( - "{} [{}] GpsLocation: {:.6}{}, {:.6}{}{}{}", - Self::ICON, - node_id, - self.latitude, - Self::UNIT, - self.longitude, - Self::UNIT, - alt_str, - acc_str - ) - } } #[cfg(feature = "simulatable")] impl Simulatable for GpsLocation { + type Params = RandomWalkParams; + /// Simulate GPS readings with random walk behavior around a base location. /// - /// # Config params interpretation + /// # Params interpretation /// - `base`: Base latitude (default: 48.2082 - Vienna) /// - `variation`: Maximum wander radius in degrees (default: 0.001 ≈ 111m) /// - `step`: Random walk step multiplier (default: 0.2) /// - `trend`: Not used for GPS fn simulate( - config: &SimulationConfig, + params: &Self::Params, previous: Option<&Self>, rng: &mut R, - timestamp: u64, + timestamp_ms: u64, ) -> Self { // Use base as latitude, and a fixed longitude offset - let base_lat = config.params.base; + let base_lat = params.base; let base_lon = 16.3738; // Vienna longitude as default - let max_delta = config.params.variation; - let step = config.params.step; + let max_delta = params.variation; + let step = params.step; // Random walk from previous position or start near base let (lat, lon) = match previous { @@ -115,7 +94,7 @@ impl Simulatable for GpsLocation { longitude: lon, altitude: Some(200.0 + rng.random::() * 10.0), accuracy: Some(5.0 + rng.random::() * 10.0), - timestamp, + timestamp: timestamp_ms, } } } diff --git a/examples/weather-mesh-demo/weather-mesh-common/src/contracts/temperature.rs b/examples/weather-mesh-demo/weather-mesh-common/src/contracts/temperature.rs index 74b09d0a..43243ecb 100644 --- a/examples/weather-mesh-demo/weather-mesh-common/src/contracts/temperature.rs +++ b/examples/weather-mesh-demo/weather-mesh-common/src/contracts/temperature.rs @@ -20,7 +20,7 @@ use serde::{Deserialize, Serialize}; use aimdb_data_contracts::Linkable; #[cfg(feature = "simulatable")] -use aimdb_data_contracts::{Simulatable, SimulationConfig}; +use aimdb_data_contracts::{RandomWalkParams, Simulatable}; #[cfg(feature = "simulatable")] use rand::RngExt; @@ -168,44 +168,34 @@ aimdb_data_contracts::migration_chain! { impl Observable for TemperatureV2 { type Signal = f32; - const ICON: &'static str = "🌡️"; const UNIT: &'static str = "°C"; fn signal(&self) -> f32 { self.celsius } - - fn format_log(&self, node_id: &str) -> alloc::string::String { - alloc::format!( - "{} [{}] Temperature: {:.1}{} at {}", - Self::ICON, - node_id, - self.celsius, - Self::UNIT, - self.timestamp - ) - } } #[cfg(feature = "simulatable")] impl Simulatable for TemperatureV2 { + type Params = RandomWalkParams; + /// Simulate temperature readings with random walk behavior. /// - /// # Config params interpretation + /// # Params interpretation /// - `base`: Center temperature value (default: 22.0°C) /// - `variation`: Maximum deviation from base (default: 3.0°C) /// - `step`: Random walk step multiplier (default: 0.2) /// - `trend`: Linear trend per sample (default: 0.0) fn simulate( - config: &SimulationConfig, + params: &Self::Params, previous: Option<&Self>, rng: &mut R, - timestamp: u64, + timestamp_ms: u64, ) -> Self { - let base = config.params.base as f32; - let variation = config.params.variation as f32; - let step = config.params.step as f32; - let trend = config.params.trend as f32; + let base = params.base as f32; + let variation = params.variation as f32; + let step = params.step as f32; + let trend = params.trend as f32; // Random walk: small delta from previous value, clamped to range let current = match previous { @@ -219,7 +209,7 @@ impl Simulatable for TemperatureV2 { TemperatureV2 { schema_version: 2, celsius: current, - timestamp, + timestamp: timestamp_ms, } } } diff --git a/examples/weather-mesh-demo/weather-station-beta/Cargo.toml b/examples/weather-mesh-demo/weather-station-beta/Cargo.toml index c88c7673..55fc6f17 100644 --- a/examples/weather-mesh-demo/weather-station-beta/Cargo.toml +++ b/examples/weather-mesh-demo/weather-station-beta/Cargo.toml @@ -10,16 +10,24 @@ publish = false name = "weather-station-beta" path = "src/main.rs" +[features] +# `sim` is opt-in and dev-only: it pulls the `Simulatable` +# impls + `rand` and wires `.simulate()`. The DEFAULT (production) build reads a +# hardware source and contains no `rand` at all — `make check-no-sim` proves it. +sim = [ + "weather-mesh-common/simulatable", + "aimdb-data-contracts/simulatable", + "dep:rand", +] + [dependencies] weather-mesh-common = { path = "../weather-mesh-common", features = [ - "simulatable", "linkable", "migratable", ] } aimdb-core = { path = "../../../aimdb-core" } aimdb-tokio-adapter = { path = "../../../aimdb-tokio-adapter" } aimdb-data-contracts = { path = "../../../aimdb-data-contracts", features = [ - "simulatable", "linkable", "migratable", ] } @@ -30,5 +38,4 @@ aimdb-mqtt-connector = { path = "../../../aimdb-mqtt-connector", features = [ tokio = { version = "1", features = ["full"] } tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } -rand = "0.10.1" -chrono = "0.4" +rand = { version = "0.10.1", optional = true } diff --git a/examples/weather-mesh-demo/weather-station-beta/src/main.rs b/examples/weather-mesh-demo/weather-station-beta/src/main.rs index a6095c64..ed7a7a3b 100644 --- a/examples/weather-mesh-demo/weather-station-beta/src/main.rs +++ b/examples/weather-mesh-demo/weather-station-beta/src/main.rs @@ -5,14 +5,61 @@ //! without external dependencies. use aimdb_core::{buffer::BufferCfg, AimDbBuilder, RecordKey}; -use aimdb_data_contracts::{Linkable, Simulatable, SimulationConfig, SimulationParams}; +use aimdb_data_contracts::Linkable; +#[cfg(feature = "sim")] +use aimdb_data_contracts::{RandomWalkParams, SimProfile, SimulatableRegistrarExt}; use aimdb_mqtt_connector::MqttConnector; use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; +#[cfg(feature = "sim")] use rand::SeedableRng; use std::sync::Arc; use tracing::info; use weather_mesh_common::{DewPoint, DewPointKey, Humidity, HumidityKey, TempKey, Temperature}; +/// A Send RNG for one simulated record (fresh per source; OS-seeded on std). +#[cfg(feature = "sim")] +fn sim_rng() -> rand::rngs::StdRng { + rand::rngs::StdRng::from_rng(&mut rand::rng()) +} + +/// Placeholder "hardware" temperature source for the production (non-sim) build. +/// +/// A real deployment reads an I2C/SPI sensor here. The demo emits a fixed +/// reading on the same 5 s cadence so the record still has exactly one writer +/// and the production dependency graph stays sim-free (no `rand`). +#[cfg(not(feature = "sim"))] +async fn read_temperature( + ctx: aimdb_core::RuntimeContext, + producer: aimdb_core::Producer, +) { + loop { + let now = unix_millis(&ctx); + producer.produce(Temperature::new(20.0, now)); + ctx.time().sleep_millis(5000).await; + } +} + +/// Placeholder "hardware" humidity source for the production (non-sim) build. +#[cfg(not(feature = "sim"))] +async fn read_humidity(ctx: aimdb_core::RuntimeContext, producer: aimdb_core::Producer) { + loop { + let now = unix_millis(&ctx); + producer.produce(Humidity { + percent: 45.0, + timestamp: now, + }); + ctx.time().sleep_millis(5000).await; + } +} + +#[cfg(not(feature = "sim"))] +fn unix_millis(ctx: &aimdb_core::RuntimeContext) -> u64 { + ctx.time() + .unix_time() + .map(|(s, ns)| s.saturating_mul(1000) + (ns / 1_000_000) as u64) + .unwrap_or(0) +} + #[tokio::main] async fn main() -> Result<(), Box> { // Initialize logging @@ -42,8 +89,26 @@ async fn main() -> Result<(), Box> { // Configure temperature record let temp_topic = TempKey::Beta.link_address().unwrap(); builder.configure::(TempKey::Beta, |reg| { - reg.buffer(BufferCfg::SpmcRing { capacity: 10 }) - .link_to(temp_topic) + reg.buffer(BufferCfg::SpmcRing { capacity: 10 }); + + // Sim-to-real is a compile-time `#[cfg]`: exactly one producer installed. + #[cfg(feature = "sim")] + reg.simulate( + SimProfile { + interval_ms: 5000, + params: RandomWalkParams { + base: 20.0, // Indoor: ~20°C + variation: 3.0, // ±3°C + trend: 0.0, + step: 0.2, + }, + }, + sim_rng(), + ); + #[cfg(not(feature = "sim"))] + reg.source(read_temperature); + + reg.link_to(temp_topic) .with_serializer(|ctx, t: &Temperature| { ctx.log().info(&format!( "Serializing temp {:.1}°C @ tick {:?}", @@ -59,8 +124,25 @@ async fn main() -> Result<(), Box> { // Configure humidity record let humidity_topic = HumidityKey::Beta.link_address().unwrap(); builder.configure::(HumidityKey::Beta, |reg| { - reg.buffer(BufferCfg::SpmcRing { capacity: 10 }) - .link_to(humidity_topic) + reg.buffer(BufferCfg::SpmcRing { capacity: 10 }); + + #[cfg(feature = "sim")] + reg.simulate( + SimProfile { + interval_ms: 5000, + params: RandomWalkParams { + base: 45.0, // Indoor: ~45% + variation: 10.0, // ±10% + trend: 0.0, + step: 0.2, + }, + }, + sim_rng(), + ); + #[cfg(not(feature = "sim"))] + reg.source(read_humidity); + + reg.link_to(humidity_topic) .with_serializer(|ctx, h: &Humidity| { ctx.log().info(&format!( "Serializing humidity {:.1}% @ tick {:?}", @@ -114,79 +196,23 @@ async fn main() -> Result<(), Box> { .finish(); }); - let (db, runner) = builder.build().await?; + let (_db, runner) = builder.build().await?; tokio::spawn(runner.run()); info!("✅ Database initialized with 3 record types"); - info!(" - Temperature: {} (synthetic)", temp_topic); - info!(" - Humidity: {} (synthetic)", humidity_topic); + #[cfg(feature = "sim")] + let mode = "synthetic"; + #[cfg(not(feature = "sim"))] + let mode = "hardware"; + info!(" - Temperature: {} ({})", temp_topic, mode); + info!(" - Humidity: {} ({})", humidity_topic, mode); info!( " - DewPoint: {} (derived via transform_join)", dew_point_topic ); - // Get producers - let temp_producer = db.producer::(TempKey::Beta.as_str())?; - let humidity_producer = db.producer::(HumidityKey::Beta.as_str())?; - - // Spawn synthetic data producer - info!("🎲 Starting synthetic data producer..."); - tokio::spawn(async move { - // Simulation configuration for indoor conditions - let temp_config = SimulationConfig { - enabled: true, - interval_ms: 5000, - params: SimulationParams { - base: 20.0, // Indoor: ~20°C - variation: 3.0, // ±3°C - step: 0.2, // Smooth random walk - trend: 0.0, // No trend - }, - }; - - let humidity_config = SimulationConfig { - enabled: true, - interval_ms: 5000, - params: SimulationParams { - base: 45.0, // Indoor: ~45% - variation: 10.0, // ±10% - step: 0.2, // Smooth random walk - trend: 0.0, // No trend - }, - }; - - // Create RNG (StdRng is Send, ThreadRng is not) - let mut rng = rand::rngs::StdRng::from_rng(&mut rand::rng()); - let mut prev_temp: Option = None; - let mut prev_humidity: Option = None; - - loop { - let now = chrono::Utc::now().timestamp_millis() as u64; - - // Generate synthetic readings using Simulatable trait - let temp = Temperature::simulate(&temp_config, prev_temp.as_ref(), &mut rng, now); - let humidity = - Humidity::simulate(&humidity_config, prev_humidity.as_ref(), &mut rng, now); - - // Produce the readings - temp_producer.produce(temp.clone()); - - humidity_producer.produce(humidity.clone()); - - tracing::info!( - "📊 Published: {:.1}°C, {:.1}%", - temp.celsius, - humidity.percent - ); - - // Store for next iteration (random walk) - prev_temp = Some(temp); - prev_humidity = Some(humidity); - - // Update every 5 seconds for smooth demo - tokio::time::sleep(tokio::time::Duration::from_secs(5)).await; - } - }); + // Producers are installed inside `configure()` (a `.simulate()` source in the + // `sim` build, a hardware `.source()` otherwise) — no manual spawn needed. info!(""); info!("🎯 Weather Station Beta ready!"); diff --git a/examples/weather-mesh-demo/weather-station-gamma/Cargo.toml b/examples/weather-mesh-demo/weather-station-gamma/Cargo.toml index 312b37b5..b5341446 100644 --- a/examples/weather-mesh-demo/weather-station-gamma/Cargo.toml +++ b/examples/weather-mesh-demo/weather-station-gamma/Cargo.toml @@ -9,15 +9,22 @@ publish = false [features] default = ["embassy-runtime"] embassy-runtime = [] +# `sim` is opt-in and dev-only: "develop against sim, flash +# against silicon". It pulls the `Simulatable` impls + `rand` and wires +# `.simulate()`; the default (flash) build reads a hardware source and has no +# `rand` in its graph. +sim = [ + "weather-mesh-common/simulatable", + "aimdb-data-contracts/simulatable", + "dep:rand", +] # Declare but don't enable these features to silence cfg warnings tokio-runtime = [] std = [] [dependencies] # AimDB dependencies -weather-mesh-common = { path = "../weather-mesh-common", default-features = false, features = [ - "simulatable", -] } +weather-mesh-common = { path = "../weather-mesh-common", default-features = false } aimdb-core = { path = "../../../aimdb-core", default-features = false, features = [ "alloc", ] } @@ -25,9 +32,7 @@ aimdb-embassy-adapter = { path = "../../../aimdb-embassy-adapter", default-featu "alloc", "embassy-runtime", ] } -aimdb-data-contracts = { path = "../../../aimdb-data-contracts", default-features = false, features = [ - "simulatable", -] } +aimdb-data-contracts = { path = "../../../aimdb-data-contracts", default-features = false } aimdb-mqtt-connector = { path = "../../../aimdb-mqtt-connector", default-features = false, features = [ "embassy-runtime", ] } @@ -85,8 +90,9 @@ micromath = { workspace = true } stm32-fmc = { workspace = true } embedded-alloc = { version = "0.6", features = ["llff"] } -# Random number generation for simulation -rand = { version = "0.10.1", default-features = false } +# Random number generation for simulation — only in the `sim` build. The +# network-stack seed uses the STM32 HAL RNG (embassy_stm32), not this crate. +rand = { version = "0.10.1", default-features = false, optional = true } [package.metadata.embassy] diff --git a/examples/weather-mesh-demo/weather-station-gamma/src/main.rs b/examples/weather-mesh-demo/weather-station-gamma/src/main.rs index f230423f..6c0eeb4f 100644 --- a/examples/weather-mesh-demo/weather-station-gamma/src/main.rs +++ b/examples/weather-mesh-demo/weather-station-gamma/src/main.rs @@ -24,8 +24,9 @@ extern crate alloc; -use aimdb_core::{AimDbBuilder, Producer, RecordKey, RuntimeContext}; -use aimdb_data_contracts::{Simulatable, SimulationConfig, SimulationParams}; +use aimdb_core::{AimDbBuilder, RecordKey}; +#[cfg(feature = "sim")] +use aimdb_data_contracts::{RandomWalkParams, SimProfile, SimulatableRegistrarExt}; use aimdb_embassy_adapter::{EmbassyAdapter, EmbassyBufferType, EmbassyRecordRegistrarExtCustom}; use aimdb_mqtt_connector::embassy_client::MqttConnectorBuilder; use defmt::*; @@ -37,6 +38,7 @@ use embassy_stm32::peripherals::ETH; use embassy_stm32::rng::Rng; use embassy_stm32::{bind_interrupts, eth, peripherals, rng, Config}; use embassy_time::{Duration, Timer}; +#[cfg(feature = "sim")] use rand::SeedableRng; use static_cell::StaticCell; use weather_mesh_common::{DewPoint, DewPointKey, Humidity, HumidityKey, TempKey, Temperature}; @@ -64,66 +66,64 @@ async fn net_task(mut runner: embassy_net::Runner<'static, Device>) -> ! { runner.run().await } -/// Temperature producer - generates synthetic data -async fn temperature_producer(ctx: RuntimeContext, producer: Producer) { - let log = ctx.log(); - log.info("🌡️ Starting temperature producer..."); - - let config = SimulationConfig { - enabled: true, +/// Sim profile for temperature (dev-tier `sim` build only). +#[cfg(feature = "sim")] +fn temperature_profile() -> SimProfile { + SimProfile { interval_ms: 5000, - params: SimulationParams { + params: RandomWalkParams { base: 22.0, // Portable: ~22°C variation: 5.0, // ±5°C (more variation) step: 0.3, // Random walk trend: 0.0, }, - }; - - let mut rng = rand::rngs::SmallRng::from_seed([42; 32]); - let mut prev: Option = None; - - loop { - let now = ctx.time().now() / 1_000_000; - let temp = Temperature::simulate(&config, prev.as_ref(), &mut rng, now); - - log.info(&alloc::format!("📊 Temp: {:.1}°C", temp.celsius)); - - producer.produce(temp.clone()); - - prev = Some(temp); - Timer::after(Duration::from_secs(5)).await; } } -/// Humidity producer - generates synthetic data -async fn humidity_producer(ctx: RuntimeContext, producer: Producer) { - let log = ctx.log(); - log.info("💧 Starting humidity producer..."); - - let config = SimulationConfig { - enabled: true, +/// Sim profile for humidity (dev-tier `sim` build only). +#[cfg(feature = "sim")] +fn humidity_profile() -> SimProfile { + SimProfile { interval_ms: 5000, - params: SimulationParams { + params: RandomWalkParams { base: 55.0, // Portable: ~55% variation: 15.0, // ±15% step: 0.3, // Random walk trend: 0.0, }, - }; - - let mut rng = rand::rngs::SmallRng::from_seed([84; 32]); - let mut prev: Option = None; + } +} +/// Placeholder "hardware" temperature source for the production (flash) build. +/// +/// A real MCU deployment reads an ADC/I2C sensor here. The demo emits a fixed +/// reading so the record still has exactly one writer and the flashed image +/// carries no `rand`. +#[cfg(not(feature = "sim"))] +async fn read_temperature( + ctx: aimdb_core::RuntimeContext, + producer: aimdb_core::Producer, +) { + let log = ctx.log(); + log.info("🌡️ Starting temperature sensor..."); loop { let now = ctx.time().now() / 1_000_000; - let humidity = Humidity::simulate(&config, prev.as_ref(), &mut rng, now); - - log.info(&alloc::format!("📊 Humidity: {:.1}%", humidity.percent)); - - producer.produce(humidity.clone()); + producer.produce(Temperature::new(22.0, now)); + Timer::after(Duration::from_secs(5)).await; + } +} - prev = Some(humidity); +/// Placeholder "hardware" humidity source for the production (flash) build. +#[cfg(not(feature = "sim"))] +async fn read_humidity(ctx: aimdb_core::RuntimeContext, producer: aimdb_core::Producer) { + let log = ctx.log(); + log.info("💧 Starting humidity sensor..."); + loop { + let now = ctx.time().now() / 1_000_000; + producer.produce(Humidity { + percent: 55.0, + timestamp: now, + }); Timer::after(Duration::from_secs(5)).await; } } @@ -257,9 +257,18 @@ async fn main(spawner: Spawner) { // Configure temperature record let temp_topic = TempKey::Gamma.link_address().unwrap(); builder.configure::(TempKey::Gamma, |reg| { - reg.buffer_sized::<16, 2>(EmbassyBufferType::SpmcRing) - .source(temperature_producer) - .link_to(temp_topic) + reg.buffer_sized::<16, 2>(EmbassyBufferType::SpmcRing); + + // Sim-to-real is a compile-time `#[cfg]`: exactly one producer installed. + #[cfg(feature = "sim")] + reg.simulate( + temperature_profile(), + rand::rngs::SmallRng::from_seed([42; 32]), + ); + #[cfg(not(feature = "sim"))] + reg.source(read_temperature); + + reg.link_to(temp_topic) .with_serializer(|_ctx, t: &Temperature| { // Manual JSON serialization for no_std let whole = t.celsius as i32; @@ -279,9 +288,17 @@ async fn main(spawner: Spawner) { // Configure humidity record let humidity_topic = HumidityKey::Gamma.link_address().unwrap(); builder.configure::(HumidityKey::Gamma, |reg| { - reg.buffer_sized::<16, 2>(EmbassyBufferType::SpmcRing) - .source(humidity_producer) - .link_to(humidity_topic) + reg.buffer_sized::<16, 2>(EmbassyBufferType::SpmcRing); + + #[cfg(feature = "sim")] + reg.simulate( + humidity_profile(), + rand::rngs::SmallRng::from_seed([84; 32]), + ); + #[cfg(not(feature = "sim"))] + reg.source(read_humidity); + + reg.link_to(humidity_topic) .with_serializer(|_ctx, h: &Humidity| { // Manual JSON serialization for no_std let whole = h.percent as i32; diff --git a/tools/aimdb-cli/src/error.rs b/tools/aimdb-cli/src/error.rs index 716f96b0..db09349f 100644 --- a/tools/aimdb-cli/src/error.rs +++ b/tools/aimdb-cli/src/error.rs @@ -45,7 +45,7 @@ pub enum CliError { InvalidJson { input: String, error: String }, /// Server returned an error response - #[error("Server error: {code}\n Message: {message}{}", format_details(.details))] + #[error("Server error: {code}\n Message: {message}{}", .details.as_ref().map(|d| format!("\n Details: {d}")).unwrap_or_default())] ServerError { code: String, message: String, @@ -91,13 +91,6 @@ impl From for CliError { } } -fn format_details(details: &Option) -> String { - match details { - Some(val) => format!("\n Details: {}", val), - None => String::new(), - } -} - #[allow(dead_code)] // Helper methods for future use impl CliError { /// Create a connection failed error