Skip to content

Commit d641b09

Browse files
authored
Refactor/034 phase2 134 mqtt knobs (#140)
* Refactor AimDB WASM adapter to remove generic type constraints and simplify API - Updated `WasmRecordRegistrarExt` to only use `.buffer(cfg)` method. - Removed `WasmJoinQueue` and related join queue implementations. - Changed `AimDb<WasmAdapter>` to `AimDb` in various locations to simplify type usage. - Updated bindings and schema registry to reflect the removal of the adapter type. - Adjusted WebSocket connector implementations to use the simplified `AimDb`. - Refactored example applications to align with the new API structure. - Enhanced documentation and comments for clarity on changes. * Refactor EmbassyAdapter initialization and improve error handling in KNX and MQTT connectors - Updated the initialization of `EmbassyAdapter` to remove unnecessary `unwrap()` calls, ensuring better error handling. - Enhanced the KNX connector to validate gateway addresses during the build process, preventing runtime errors from invalid configurations. - Modified the connection task in the KNX connector to handle socket errors more gracefully and added logic to manage heartbeat responses, improving connection stability. - Updated MQTT connector to accept a network stack during construction, aligning with the new architecture for network connectors. - Adjusted various tests to reflect changes in the handling of connection states and error scenarios, ensuring robust testing coverage. * chore: update embassy subproject to latest commit * refactor: remove unused EmbassyRecordRegistrarExt from KNX and Weather Station demos * refactor: clean up import statements in KNX connector demo * refactor: simplify rust-analyzer check configuration in devcontainer * refactor: remove unused EmbassyRecordRegistrarExt from MQTT connector demo * refactor: streamline runtime accessors and connection handling in KNX connector * refactor: update documentation to reflect current AimDB API and improve record indexing in AimDbBuilder
1 parent b02ec70 commit d641b09

98 files changed

Lines changed: 2900 additions & 3830 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

.devcontainer/devcontainer.json

Lines changed: 1 addition & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -26,10 +26,7 @@
2626
"terminal.integrated.shell.linux": "/bin/bash",
2727
"rust-analyzer.cargo.features": "all",
2828
"rust-analyzer.check.command": "clippy",
29-
"rust-analyzer.check.extraArgs": [
30-
"--all-targets",
31-
"--all-features"
32-
],
29+
"rust-analyzer.check.allTargets": true,
3330
"rust-analyzer.cargo.buildScripts.enable": true,
3431
"rust-analyzer.procMacro.enable": true,
3532
"files.watcherExclude": {

CHANGELOG.md

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
3131

3232
### Added
3333

34+
- **Design 034 Phase 3 — sans-io KNX/IP tunneling engine shared by both transports (Issue #135, [review doc §3.7](docs/design/034-technical-debt-review.md)).** The entire tunneling lifecycle — CONNECT_REQUEST/RESPONSE handshake, TUNNELING_REQUEST/ACK sequence + pending-ACK bookkeeping, keepalive (CONNECTIONSTATE_REQUEST) scheduling, ACK-timeout sweeps, and reconnect-with-backoff — now lives **once**, in the new runtime-neutral `aimdb_knx_connector::tunnel` module (`no_std + alloc`, no tokio/embassy imports), driven as a poll-based state machine (events in, `Action`s out, `next_deadline()` for timer arming). `tokio_client.rs` (988 → ~530 lines incl. a new fake-gateway integration test) and `embassy_client.rs` (1,055 → ~450 lines) are reduced to socket shims; the previously untestable handshake/ACK/keepalive/reconnect paths now have 15 host-run unit tests plus a scripted localhost-UDP roundtrip test. Behavioral unifications: Embassy gains the 5 s CONNECT_RESPONSE timeout (previously waited forever), both shims reconnect on fatal socket errors, and the dead tokio-only per-publish ACK oneshot is dropped (it was always `None`); the CONNECT_REQUEST HPAI stays per-transport (`LocalEndpoint`: tokio = real bound address, Embassy = NAT mode). ([aimdb-knx-connector](aimdb-knx-connector/CHANGELOG.md))
35+
3436
- **Design 034 Phase 2 — dyn-safe `RuntimeOps` capability trait (Issue #130, [review doc](docs/design/034-technical-debt-review.md)).** New object-safe trait in `aimdb-executor` (`name` / `now_nanos` / `unix_time` / boxed `sleep` / `log(LogLevel, …)`) so a runtime adapter can travel as `Arc<dyn RuntimeOps>` instead of a generic parameter — the groundwork for removing `R` from the record object graph (#131). Implemented by `TokioAdapter`, `EmbassyAdapter`, and `WasmAdapter`, each covered by a shared behavioral contract test. `BoxFuture`'s canonical definition moves to `aimdb-executor` (re-exported unchanged from `aimdb-core`). ([aimdb-executor](aimdb-executor/CHANGELOG.md), [aimdb-tokio-adapter](aimdb-tokio-adapter/CHANGELOG.md), [aimdb-embassy-adapter](aimdb-embassy-adapter/CHANGELOG.md), [aimdb-wasm-adapter](aimdb-wasm-adapter/CHANGELOG.md))
3537

3638
- **M17 — centralized Embassy connector spine: one audited home for the single-core `unsafe` ([Design 033](docs/design/033-M17-unify-connectors-drop-send.md)).** New `aimdb-embassy-adapter::connectors` module (features `connectors` / `connector-io`) collects the force-`Send` plumbing every Embassy connector used to hand-roll: session transports get `EmbassySessionClient`/`EmbassySessionServer`, `OneShotDialer`/`OneShotListener`/`OneShotCell`, and the framed `EmbassyConnection` + `Framer`; data-plane transports get the `EmbassySink`/`EmbassySource` bridges (over `EmbassySinkRaw`/`EmbassySourceRaw`) that ride core's existing `pump_sink`/`pump_source`, plus `into_box_future` for protocol tasks. The serial Embassy half is now thin sugar (just a COBS `Framer`) with **zero `unsafe`** (down from a 407-line hand-roll with 7 `unsafe impl`s); the MQTT and KNX Embassy halves dropped their hand-rolled publisher/router loops and `SendFutureWrapper` use to ride core's pumps (KNX inbound telegrams now flow through `pump_source`). All connector-crate `unsafe`/`SendFutureWrapper` is gone — confined to the adapter. The std/Tokio side, `aimdb-client`, the WebSocket server, examples, and tests are unchanged. (Chosen over Design 033's original "drop `Send` from the contract", which would have pushed `!Send` onto the std side; see the doc's Implementation Decision.) ([aimdb-embassy-adapter](aimdb-embassy-adapter/CHANGELOG.md), [aimdb-serial-connector](aimdb-serial-connector/CHANGELOG.md), [aimdb-mqtt-connector](aimdb-mqtt-connector/CHANGELOG.md), [aimdb-knx-connector](aimdb-knx-connector/CHANGELOG.md))
@@ -44,6 +46,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
4446

4547
### Changed (breaking)
4648

49+
- **Design 034 Phase 3 — runtime type parameter `R` removed from the object graph (Issue #131, [review doc §3.2/§3.3](docs/design/034-technical-debt-review.md)).** The runtime is now a *value* (`Arc<dyn aimdb_executor::RuntimeOps>`, the #130 groundwork) instead of a type parameter; the only generic left on the user-facing object graph is the record type `T`. The full break inventory:
50+
- **Types lose `R`:** `AimDb<R>``AimDb`, `AimDbBuilder<R>``AimDbBuilder` (the `NoRuntime` typestate is gone — a missing runtime is a `build()` error per the #133 contract), `TypedRecord<T, R>``TypedRecord<T>`, `RecordRegistrar<'a, T, R>``RecordRegistrar<'a, T>`, `TransformBuilder<I, O, R>``TransformBuilder<I, O>`, `JoinBuilder<O, R>``JoinBuilder<O>`, `RecordT<R>``RecordT`, and `ConnectorBuilder<R>``ConnectorBuilder` (its `build()` takes `&AimDb`). Turbofish updates: `get_typed_record_by_key::<T, R>``::<T>`, `as_typed::<T, R>``::<T>`.
51+
- **`RuntimeContext` is a concrete struct** wrapping `Arc<dyn RuntimeOps>`. `ctx.time().now()` returns **`u64` nanoseconds** from an arbitrary monotonic epoch (was `R::Instant`); `sleep` takes a plain `core::time::Duration` (plus `sleep_millis`/`sleep_secs` helpers); `millis()`/`secs()`/`micros()`/`duration_since()`/`duration_as_nanos()` are deleted (durations are concrete, instants are integer math); the panicking `extract_from_any` is deleted.
52+
- **`source`/`tap`/`transform`/`transform_join` are inherent methods** on `RecordRegistrar` — the per-adapter extension-trait wrappers and `aimdb-core`'s `ext_macros.rs` are gone, as are the `*_raw` variants (the inherent methods *are* the foundational API; closures keep the `(ctx, producer)` arg order, so user code mostly just drops an import). The adapter ext traits (`TokioRecordRegistrarExt`, `EmbassyRecordRegistrarExt`(+`Custom`), `WasmRecordRegistrarExt`) shrink to the one genuinely adapter-specific step: `.buffer(cfg)` construction.
53+
- **`JoinFanInRuntime` deleted** (with `JoinQueue`/`JoinSender`/`JoinReceiver` and all three per-adapter `join_queue.rs` files): multi-input join fan-in now uses one bounded `async-channel` queue in core — the same primitive the session engine already uses on tokio, Embassy, and WASM. Capacity stays 64 on `std`/wasm32 and rises to 16 on embedded `no_std` (up from Embassy's 8); the queue now closes on every runtime when all forwarders exit (the Embassy queue previously never closed).
54+
- **Runtime accessors:** `runtime_arc()``runtime_ops()` (returns `Arc<dyn RuntimeOps>`), new `runtime_ctx()`; the borrowed `AimDb::runtime()` and the type-erased `runtime_any()` are deleted (zero callers in aimdb/aimdb-pro — `&*db.runtime_ops()` covers the borrowed flavor; [follow-up doc §2.5](docs/design/035-review-followups-deferred.md)). `AimDbBuilder::on_start` closures receive `RuntimeContext` (was `Arc<R>`). Context-aware (de)serializers receive the concrete `RuntimeContext` (was `Arc<dyn Any + Send + Sync>`); `Router::route`'s ctx argument follows. The `RuntimeForProfiling` marker-trait workaround is deleted (profiling clocks ride `RuntimeOps::now_nanos`); the session client engine's clock is `Arc<dyn RuntimeOps>` (was `Arc<R: TimeOps>`).
55+
- **Embassy network capability moves to construction:** the `EmbassyNetwork` runtime trait is deleted (a `dyn RuntimeOps` can't surface adapter-specific capabilities) along with `EmbassyAdapter::new_with_network``EmbassyAdapter` is a stateless unit type with **zero `unsafe`**. The Embassy MQTT/KNX connector builders take the `embassy_net::Stack` at construction (`MqttConnectorBuilder::new(url, stack)` / `KnxConnectorBuilder::new(url, stack)`), wrapped in the new force-`Send + Sync` `aimdb_embassy_adapter::connectors::NetStack` so the single-core `unsafe` stays in the one audited module.
56+
- **Downstream follows mechanically:** `aimdb-sync` (`AimDb` handles), `aimdb-persistence` (`.persist()`/`.with_persistence()` ext traits de-genericized), `aimdb-data-contracts::log_tap`, `aimdb-codegen` templates (generated `configure_schema(builder: &mut AimDbBuilder)`), and all examples/tools.
57+
- Acceptance: `grep -rn "extract_from_any|runtime_any|RuntimeForProfiling|JoinFanInRuntime"` over the workspace sources returns nothing; the remaining `dyn Any` in core is data-plane only (`SerializerFn`/`produce_any`/`JoinTrigger`/`TopicProviderAny` — the #131 §6 stretch, tracked as a follow-up). ([aimdb-core](aimdb-core/CHANGELOG.md), [aimdb-executor](aimdb-executor/CHANGELOG.md), [aimdb-tokio-adapter](aimdb-tokio-adapter/CHANGELOG.md), [aimdb-embassy-adapter](aimdb-embassy-adapter/CHANGELOG.md), [aimdb-wasm-adapter](aimdb-wasm-adapter/CHANGELOG.md), [aimdb-mqtt-connector](aimdb-mqtt-connector/CHANGELOG.md), [aimdb-knx-connector](aimdb-knx-connector/CHANGELOG.md), [aimdb-persistence](aimdb-persistence/CHANGELOG.md), [aimdb-sync](aimdb-sync/CHANGELOG.md))
58+
4759
- **Design 034 Phase 2 — MQTT knobs move out of core; `ConnectorConfig` pruned (Issue #134, [review doc §3.6](docs/design/034-technical-debt-review.md)).** Core's generic link builders drop `with_qos`/`with_retain` (`with_timeout_ms` stays, de-MQTT'd); the knobs now live in `aimdb-mqtt-connector` as the `MqttLinkExt` (qos, outbound + inbound) and `MqttOutboundLinkExt` (retain, publish-side only) extension traits, pushing the **same** `("qos", …)`/`("retain", …)` option keys the MQTT clients have always read — wire behavior unchanged; importing the trait makes the MQTT intent explicit at the call site (generic escape hatch: `with_config(key, value)`). `ConnectorConfig` loses its never-read typed `qos`/`retain` fields and the speculative Kafka/HTTP/shmem interpretation docs — it keeps `timeout_ms` + `protocol_options`, and core now documents no protocol that lacks an in-tree connector. ([aimdb-core](aimdb-core/CHANGELOG.md), [aimdb-mqtt-connector](aimdb-mqtt-connector/CHANGELOG.md))
4860

4961
- **Design 034 Phase 2 — panic-free builder validation: `build()` reports every configuration mistake at once (Issue #133, [review doc §3.4](docs/design/034-technical-debt-review.md)).** Builder methods never panic on user mistakes anymore. Conflicting `.source()`/`.transform()`/`.link_from()` registrations, missing serializers/deserializers, invalid connector URLs, unregistered schemes, and key-reused-with-different-type are *recorded* (the conflicting registration is skipped) and `build()` returns one `DbError::InvalidConfiguration { errors: Vec<ConfigError> }` carrying **all** findings — each with the record key and, where applicable, the connector URL. The worst panic — "requires a buffer" firing at spawn time inside a connector factory closure — is now a build()-time check, which also makes `.buffer()` after `.link_to()`/`.link_from()` legal (order-independent). Duplicate keys and dependency-graph cycles fold into the same collected report (previously distinct `DuplicateRecordKey`/`CyclicDependency` returns from `build()`). Remaining `panic!`/`expect`s in the builder path are internal invariants and say "this is a bug in aimdb-core". ([aimdb-core](aimdb-core/CHANGELOG.md))

Cargo.lock

Lines changed: 1 addition & 1 deletion
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

_external/embassy

Submodule embassy updated 48 files

aimdb-client/tests/pump_client.rs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,7 @@ struct Msg {
3333
/// is robust against subscription-registration timing (a fresh subscriber may
3434
/// only see values produced after it attaches).
3535
async fn mirror_reaches(
36-
db: &Arc<AimDb<TokioAdapter>>,
36+
db: &Arc<AimDb>,
3737
key: &str,
3838
want: &serde_json::Value,
3939
mut push: impl FnMut(),

aimdb-codegen/src/lib.rs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
//! - **Mermaid diagram** — `.aimdb/architecture.mermaid`, a read-only graph
77
//! projection of the architecture (see [`generate_mermaid`])
88
//! - **Rust source** — `src/generated_schema.rs`, compilable AimDB schema
9-
//! using the actual 0.5.x API (see [`generate_rust`])
9+
//! using the current AimDB API (see [`generate_rust`])
1010
//!
1111
//! # Usage
1212
//!

aimdb-codegen/src/rust.rs

Lines changed: 6 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
//! Rust source code generator
22
//!
33
//! Converts an [`ArchitectureState`] into compilable Rust source that uses the
4-
//! actual AimDB 0.5.x API: `#[derive(RecordKey)]`, `BufferCfg`, and
4+
//! current AimDB API: `#[derive(RecordKey)]`, `BufferCfg`, and
55
//! `AimDbBuilder::configure()`.
66
//!
77
//! Uses [`quote`] for quasi-quoting token streams and [`prettyplease`] for
@@ -383,7 +383,7 @@ pub fn generate_tasks_rs(state: &ArchitectureState, binary_name: &str) -> Option
383383
let fn_name = format_ident!("{}", task.name);
384384

385385
// Build parameter list
386-
let mut params: Vec<TokenStream> = vec![quote! { ctx: RuntimeContext<TokioAdapter> }];
386+
let mut params: Vec<TokenStream> = vec![quote! { ctx: RuntimeContext }];
387387
for input in &task.inputs {
388388
let arg_name = format_ident!("{}", to_snake_case(&input.record));
389389
let value_type = format_ident!("{}Value", input.record);
@@ -418,7 +418,6 @@ pub fn generate_tasks_rs(state: &ArchitectureState, binary_name: &str) -> Option
418418

419419
let file_tokens = quote! {
420420
use aimdb_core::{Consumer, DbResult, Producer, RuntimeContext};
421-
use aimdb_tokio_adapter::TokioAdapter;
422421
use #common_crate::*;
423422

424423
#(#task_fns)*
@@ -557,7 +556,6 @@ fn emit_imports(state: &ArchitectureState) -> TokenStream {
557556
use aimdb_core::builder::AimDbBuilder;
558557
use aimdb_core::RecordKey;
559558
use aimdb_data_contracts::{#(#contract_traits),*};
560-
use aimdb_executor::RuntimeAdapter;
561559
use serde::{Deserialize, Serialize};
562560
}
563561
}
@@ -731,7 +729,7 @@ fn emit_configure_schema(state: &ArchitectureState) -> TokenStream {
731729
/// addresses. Producers, consumers, serializers, and deserializers contain
732730
/// business logic and must be provided by application code — they are not
733731
/// generated here.
734-
pub fn configure_schema<R: RuntimeAdapter + 'static>(builder: &mut AimDbBuilder<R>) {
732+
pub fn configure_schema(builder: &mut AimDbBuilder) {
735733
#(#record_blocks)*
736734
}
737735
}
@@ -1250,7 +1248,6 @@ pub fn generate_hub_schema_rs(state: &ArchitectureState) -> String {
12501248
let file_tokens = quote! {
12511249
use aimdb_core::buffer::BufferCfg;
12521250
use aimdb_core::builder::AimDbBuilder;
1253-
use aimdb_executor::RuntimeAdapter;
12541251
use #common_crate::*;
12551252

12561253
#configure_fn
@@ -1688,7 +1685,7 @@ pub fn {handler}(input: &{in_t}) -> Option<{out_t}> {{\n\
16881685
// Pure source
16891686
fns.push_str(&format!(
16901687
"pub async fn {}(\n\
1691-
_ctx: aimdb_core::RuntimeContext<TokioAdapter>,\n\
1688+
_ctx: aimdb_core::RuntimeContext,\n\
16921689
_producer: aimdb_core::Producer<{out_t}>,\n\
16931690
) {{\n\
16941691
todo!(\"implement {}\")\n\
@@ -1699,7 +1696,7 @@ pub fn {handler}(input: &{in_t}) -> Option<{out_t}> {{\n\
16991696
// Pure sink / tap
17001697
fns.push_str(&format!(
17011698
"pub async fn {}(\n\
1702-
_ctx: aimdb_core::RuntimeContext<TokioAdapter>,\n\
1699+
_ctx: aimdb_core::RuntimeContext,\n\
17031700
_consumer: aimdb_core::Consumer<{in_t}>,\n\
17041701
) {{\n\
17051702
todo!(\"implement {}\")\n\
@@ -1727,7 +1724,6 @@ pub async fn {task_name}() {{\n\
17271724
// This file is scaffolded once — it will not be overwritten on subsequent runs.\n\
17281725
// Regenerate signatures: delete this file, then run `aimdb generate --hub`.\n\
17291726
\n\
1730-
use aimdb_tokio_adapter::TokioAdapter;\n\
17311727
use {common_crate}::*;\n\
17321728
\n\
17331729
{fns}"
@@ -1832,10 +1828,6 @@ url = "mqtt://ota/cmd/{variant}"
18321828
out.contains("use aimdb_core::RecordKey;"),
18331829
"Missing RecordKey import:\n{out}"
18341830
);
1835-
assert!(
1836-
out.contains("use aimdb_executor::RuntimeAdapter;"),
1837-
"Missing RuntimeAdapter import:\n{out}"
1838-
);
18391831
assert!(
18401832
out.contains("use serde::{Deserialize, Serialize};"),
18411833
"Missing serde import:\n{out}"
@@ -1937,9 +1929,7 @@ url = "mqtt://ota/cmd/{variant}"
19371929
fn configure_schema_function_present() {
19381930
let out = generated();
19391931
assert!(
1940-
out.contains(
1941-
"pub fn configure_schema<R: RuntimeAdapter + 'static>(builder: &mut AimDbBuilder<R>)"
1942-
),
1932+
out.contains("pub fn configure_schema(builder: &mut AimDbBuilder)"),
19431933
"Missing configure_schema function:\n{out}"
19441934
);
19451935
}

0 commit comments

Comments
 (0)