88 replace spawn trait with futuresunordered (#117)

* feat: add design document for removing `Spawn` trait and refactoring of futures handling in AimDB

* Refactor AimDB to remove `Spawn` trait and collect futures in `build()`

- Updated `WebSocketConnectorImpl` to collect outbound publisher futures instead of spawning them directly.
- Refactored `ServerState` to build the WebSocket Axum server future and return it as a `BoxFuture`.
- Revised design documentation to reflect the removal of the `Spawn` trait and the new approach for handling futures.
- Modified examples to adapt to the new `build()` and `run()` pattern, ensuring proper initialization and concurrent execution of tasks.
- Removed unnecessary `Spawn` trait bounds from various components and updated related code generation.
- Cleaned up example projects to align with the new architecture, including adjustments to how the Embassy adapter is initialized and how the database runner is executed.

* feat: add configurable command channel capacity for KNX connector

* chore: update dependencies in Cargo.lock and embassy submodule

* feat: remove `Spawn` trait and update `AimDbBuilder::build()` to return `(AimDb, AimDbRunner)` across the workspace

* feat: Remove runtime parameter `R` from `Producer<T>` and `Consumer<T>`

- Updated `Producer<T, R>` to `Producer<T>` and `Consumer<T, R>` to `Consumer<T>`, simplifying user-facing signatures.
- Removed the `.key()` method from `Producer<T>`, as the record key is now captured at registration.
- Adjusted tests and examples to reflect the new type signatures.
- Updated documentation to describe the architectural changes and the motivation behind the removal of `R`.
- Ensured backward compatibility for existing async function signatures while optimizing the hot path for production and consumption of records.

* Refactor produce method to be synchronous

- Changed the `produce` method in various modules from async to synchronous, removing unnecessary await calls.
- Updated related documentation and examples to reflect the new synchronous behavior.
- Ensured that all instances of `producer.produce(value).await` were replaced with `producer.produce(value)`.
- This change simplifies the API and improves performance by eliminating the overhead of async handling where it is not needed.

* feat: simplify DewPoint production by removing await and error handling

* feat: add "macros" feature to Tokio dependency in Cargo.toml

* feat: update documentation and design notes for removing `Spawn` trait and `R` parameter from `Producer<T>` and `Consumer<T>`

* feat: add build command for remote-access-demo in Makefile and update config producer in server

* feat: update dependencies and remove alloc feature checks across the codebase

* feat: add "generic-queue-16" feature to embassy-time dependency in Cargo.toml

* feat: update embassy-time dependency to include generic-queue-16 feature for FuturesUnordered compatibility

Author: lxsaah

CHANGELOG.md

@@ -27,6 +27,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Changed (breaking)
+
+- **M13 — `Spawn` trait removed across the workspace; `AimDbBuilder::build()` now returns `(AimDb, AimDbRunner)` (Issue #88, [Design 028](docs/design/028-M13-remove-spawn-trait.md)).** Every future the database drives — producer services, taps, transforms, join forwarders, connector loops, the remote-access supervisor, `on_start` tasks — is collected at build time and driven by a single `FuturesUnordered` inside `runner.run().await`. Adapter implementations (`TokioAdapter`, `EmbassyAdapter`, `WasmAdapter`) drop their `impl Spawn`. The `embassy-task-pool-8/16/32` features are deleted and `EmbassyAdapter::new_with_network` no longer takes a `Spawner`. Connector authors must update `ConnectorBuilder::build()` to return `Vec<BoxFuture>` instead of `Arc<dyn Connector>`. See each crate's CHANGELOG for the per-crate impact.
+
 ## [1.1.0] - 2026-05-22
 
 ### Added

Cargo.lock

@@ -75,6 +75,7 @@ tracing = { version = "0.1", default-features = false }
 
 # Async utilities
 futures = "0.3"
+futures-util = { version = "0.3", default-features = false, features = ["alloc"] }
 
 # CLI (for aimdb-cli)
 clap = { version = "4.0", features = ["derive"] }

Cargo.toml

@@ -277,10 +277,14 @@ examples:
 	@printf "$(GREEN)Building all example projects...$(NC)\n"
 	@printf "$(YELLOW)  → Building sync-api-demo (synchronous API wrapper)$(NC)\n"
 	cargo build --package sync-api-demo
+	@printf "$(YELLOW)  → Building mqtt-connector-demo-common (shared MQTT demo code, runtime-agnostic)$(NC)\n"
+	cargo build --package mqtt-connector-demo-common
 	@printf "$(YELLOW)  → Building tokio-mqtt-connector-demo (native, tokio runtime)$(NC)\n"
 	cargo build --package tokio-mqtt-connector-demo
 	@printf "$(YELLOW)  → Building embassy-mqtt-connector-demo (embedded, embassy runtime)$(NC)\n"
 	cargo build --package embassy-mqtt-connector-demo --target thumbv7em-none-eabihf
+	@printf "$(YELLOW)  → Building knx-connector-demo-common (shared KNX demo code, runtime-agnostic)$(NC)\n"
+	cargo build --package knx-connector-demo-common
 	@printf "$(YELLOW)  → Building tokio-knx-connector-demo (native, tokio runtime)$(NC)\n"
 	cargo build --package tokio-knx-connector-demo
 	@printf "$(YELLOW)  → Building embassy-knx-connector-demo (embedded, embassy runtime)$(NC)\n"
@@ -295,6 +299,8 @@ examples:
 	cargo build --package weather-station-beta
 	@printf "$(YELLOW)  → Building weather-station-gamma (embedded, embassy runtime)$(NC)\n"
 	cargo build --package weather-station-gamma --target thumbv7em-none-eabihf
+	@printf "$(YELLOW)  → Building remote-access-demo (AimX server + client)$(NC)\n"
+	cargo build --package remote-access-demo
 	@printf "$(YELLOW)  → Building hello-mailbox (sync)$(NC)\n"
 	cargo build --package hello-mailbox
 	@printf "$(YELLOW)  → Building hello-single-latest-async$(NC)\n"

Makefile

@@ -93,7 +93,11 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             .finish();
     });
 
-    builder.build()?.run().await?;
+    // `.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)`.
+    builder.run().await?;
     Ok(())
 }
 ```

README.md

@@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Changed (breaking)
+
+- Emitted task scaffolds now use `Producer<T>` / `Consumer<T>` (no `, TokioAdapter` second parameter) and emitted doc tables show the same form, matching the M14 cleanup in `aimdb-core` (Design 029). Regenerate downstream scaffolds after upgrading.
+- Emitted `configure_schema` signature changed from `<R: Spawn + 'static>` to `<R: RuntimeAdapter + 'static>`; emitted prelude now imports `aimdb_executor::RuntimeAdapter` instead of `Spawn` (Issue #88). Regenerate downstream schemas.
+
 ## [0.2.0] - 2026-05-22
 
 ### Changed

_external/embassy

@@ -387,12 +387,12 @@ pub fn generate_tasks_rs(state: &ArchitectureState, binary_name: &str) -> Option
             for input in &task.inputs {
                 let arg_name = format_ident!("{}", to_snake_case(&input.record));
                 let value_type = format_ident!("{}Value", input.record);
-                params.push(quote! { #arg_name: Consumer<#value_type, TokioAdapter> });
+                params.push(quote! { #arg_name: Consumer<#value_type> });
             }
             for output in &task.outputs {
                 let arg_name = format_ident!("{}", to_snake_case(&output.record));
                 let value_type = format_ident!("{}Value", output.record);
-                params.push(quote! { #arg_name: Producer<#value_type, TokioAdapter> });
+                params.push(quote! { #arg_name: Producer<#value_type> });
             }
 
             let todo_msg = match &task.task_type {
@@ -557,7 +557,7 @@ fn emit_imports(state: &ArchitectureState) -> TokenStream {
         use aimdb_core::builder::AimDbBuilder;
         use aimdb_core::RecordKey;
         use aimdb_data_contracts::{#(#contract_traits),*};
-        use aimdb_executor::Spawn;
+        use aimdb_executor::RuntimeAdapter;
         use serde::{Deserialize, Serialize};
     }
 }
@@ -731,7 +731,7 @@ fn emit_configure_schema(state: &ArchitectureState) -> TokenStream {
         /// addresses. Producers, consumers, serializers, and deserializers contain
         /// business logic and must be provided by application code — they are not
         /// generated here.
-        pub fn configure_schema<R: Spawn + 'static>(builder: &mut AimDbBuilder<R>) {
+        pub fn configure_schema<R: RuntimeAdapter + 'static>(builder: &mut AimDbBuilder<R>) {
             #(#record_blocks)*
         }
     }
@@ -1250,7 +1250,7 @@ pub fn generate_hub_schema_rs(state: &ArchitectureState) -> String {
     let file_tokens = quote! {
         use aimdb_core::buffer::BufferCfg;
         use aimdb_core::builder::AimDbBuilder;
-        use aimdb_executor::Spawn;
+        use aimdb_executor::RuntimeAdapter;
         use #common_crate::*;
 
         #configure_fn
@@ -1611,10 +1611,10 @@ fn build_transform_call(task: &TaskDef, variant_ident: &syn::Ident) -> TokenStre
 ///
 /// | Inputs | Outputs | API                   | Generated stub            |
 /// |--------|---------|-----------------------|---------------------------|
-/// | N > 1  | ≥ 1     | `.transform_join()`   | `async fn task_handler(JoinEventRx, Producer<O, R>)` |
+/// | N > 1  | ≥ 1     | `.transform_join()`   | `async fn task_handler(JoinEventRx, Producer<O>)` |
 /// | 1      | ≥ 1     | `.transform().map()`  | `fn task_transform(&Input) -> Option<Output>` |
-/// | 0      | ≥ 1     | `.source()`           | `async fn task(RuntimeContext, Producer<O, R>)` |
-/// | ≥ 1    | 0       | `.tap()`              | `async fn task(RuntimeContext, Consumer<I, R>)` |
+/// | 0      | ≥ 1     | `.source()`           | `async fn task(RuntimeContext, Producer<O>)` |
+/// | ≥ 1    | 0       | `.tap()`              | `async fn task(RuntimeContext, Consumer<I>)` |
 pub fn generate_hub_tasks_rs(state: &ArchitectureState) -> String {
     let project = state
         .project
@@ -1660,7 +1660,7 @@ pub fn generate_hub_tasks_rs(state: &ArchitectureState) -> String {
 /// {inputs_doc}\n\
 pub async fn {handler}(\n\
     mut _rx: aimdb_core::transform::JoinEventRx,\n\
-    _producer: aimdb_core::Producer<{out_t}, TokioAdapter>,\n\
+    _producer: aimdb_core::Producer<{out_t}>,\n\
 ) {{\n\
     while let Ok(_trigger) = _rx.recv().await {{\n\
         todo!(\"implement {handler}\")\n\
@@ -1689,7 +1689,7 @@ pub fn {handler}(input: &{in_t}) -> Option<{out_t}> {{\n\
             fns.push_str(&format!(
                 "pub async fn {}(\n\
     _ctx: aimdb_core::RuntimeContext<TokioAdapter>,\n\
-    _producer: aimdb_core::Producer<{out_t}, TokioAdapter>,\n\
+    _producer: aimdb_core::Producer<{out_t}>,\n\
 ) {{\n\
     todo!(\"implement {}\")\n\
 }}\n\n",
@@ -1700,7 +1700,7 @@ pub fn {handler}(input: &{in_t}) -> Option<{out_t}> {{\n\
             fns.push_str(&format!(
                 "pub async fn {}(\n\
     _ctx: aimdb_core::RuntimeContext<TokioAdapter>,\n\
-    _consumer: aimdb_core::Consumer<{in_t}, TokioAdapter>,\n\
+    _consumer: aimdb_core::Consumer<{in_t}>,\n\
 ) {{\n\
     todo!(\"implement {}\")\n\
 }}\n\n",
@@ -1833,8 +1833,8 @@ url = "mqtt://ota/cmd/{variant}"
             "Missing RecordKey import:\n{out}"
         );
         assert!(
-            out.contains("use aimdb_executor::Spawn;"),
-            "Missing Spawn import:\n{out}"
+            out.contains("use aimdb_executor::RuntimeAdapter;"),
+            "Missing RuntimeAdapter import:\n{out}"
         );
         assert!(
             out.contains("use serde::{Deserialize, Serialize};"),
@@ -1938,7 +1938,7 @@ url = "mqtt://ota/cmd/{variant}"
         let out = generated();
         assert!(
             out.contains(
-                "pub fn configure_schema<R: Spawn + 'static>(builder: &mut AimDbBuilder<R>)"
+                "pub fn configure_schema<R: RuntimeAdapter + 'static>(builder: &mut AimDbBuilder<R>)"
             ),
             "Missing configure_schema function:\n{out}"
         );

aimdb-codegen/CHANGELOG.md

@@ -7,6 +7,38 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Changed (breaking)
+
+- **`Producer::produce` is now sync + infallible; `Consumer::subscribe` is now infallible (Design 029 follow-up, M14).** The pre-resolved `WriteHandle::push` cannot fail and the pre-resolved buffer Arc makes `subscribe()` infallible. Call sites collapse: `producer.produce(x).await?` → `producer.produce(x);` and `let Ok(reader) = consumer.subscribe() else { ... }` → `let reader = consumer.subscribe();`. The `ProducerTrait::produce_any` / `ConsumerTrait::subscribe_any` trait surfaces stay `Result`/`async` because the type-erasure downcast remains fallible.
+  - `AimDb::produce<T>(key, value) -> DbResult<()>` is now sync; `.await` on the call site goes away. Only the key lookup can fail.
+  - `Database::produce` likewise sync.
+  - `TypedRecord::produce` is now `pub fn produce(&self, val: T)` (was `pub async fn produce`).
+  - `aimdb-wasm-adapter`: `bindings::poll_sync` helper deleted — no remaining callers now that `TypedRecord::produce` is sync.
+  - Dead `consumer.subscribe()` error arms in `transform/single.rs` and `transform/join.rs` removed (the `Err` branch was unreachable after M14).
+
+- **`Producer<T>` / `Consumer<T>` drop the runtime parameter `R` and pre-resolve the record at build time (Design 029, M14).** Producer/Consumer become handles to a buffer rather than tickets to look one up: `produce()` is one virtual call (no `HashMap<key>` probe, no `TypeId` check, no downcast), and `subscribe()` collapses to `buffer.subscribe_boxed()`. The internal mechanic is a new crate-private `WriteHandle<T>` trait backed by `RecordWriter<T>` (in `aimdb-core/src/buffer/writer.rs`), pre-bound to the record's `Arc<dyn DynBuffer<T>>` + snapshot mutex + metadata tracker.
+  - `Producer<T, R>` → `Producer<T>`; `Consumer<T, R>` → `Consumer<T>`. User code that names the two-parameter form must drop the trailing adapter arg.
+  - `Producer::key(&self) -> &str` is **removed**. Capture the record key at the registration site instead.
+  - `Producer::produce(value) -> ()` and `Consumer::subscribe() -> Box<dyn BufferReader<T> + Send>` (v0.4 revision — see the sync/infallible bullet above for the rationale and migration). The `ProducerTrait::produce_any` / `ConsumerTrait::subscribe_any` trait surfaces retain `async`/`Result` for the type-erased downcast that can still fail.
+  - `AimDb::producer<T>(key)` / `AimDb::consumer<T>(key)` now return `DbResult<…>` (was infallible). They resolve the typed record up front, so callers that previously assumed inference must add `?`.
+  - `Consumer<T>` cannot exist without a buffer: `.tap()` on a record with no `.buffer(...)` now surfaces as `MissingConfiguration` at build time (was a deferred subscribe-time error).
+  - `TypedRecord::buffer` field is `Option<Arc<dyn DynBuffer<T>>>` (was `Box`); `TypedRecord::set_buffer(Box<…>)` keeps its public signature and converts via `Arc::from(box_)` internally.
+  - `TypedRecord::create_producer_trait(&self)` no longer takes `db` / `record_key` — it uses the new `writer_handle()`.
+  - `ConnectorBuilder<R>` cascade is zero-LOC: no connector struct carried `R` after M13. The outbound `consumer_factory` / inbound `producer_factory` callbacks now resolve the record once at link-startup time (via `db.inner().get_typed_record_by_key`) and construct the new handles.
+  - Codegen-emitted task scaffolds use `Producer<T>` / `Consumer<T>` (no `, TokioAdapter`).
+  - `data-contracts` `log_tap` parameter is `Consumer<T>`.
+
+- **`Spawn` trait removed; `AimDbBuilder::build()` now returns `(AimDb<R>, AimDbRunner)` (Issue #88, Design 028).** Every future the database needs — `.source()`/`.tap()`/`.transform()` tasks, on_start hooks, connector loops, the remote-access supervisor — is collected at build time into the new `AimDbRunner`, then driven by a single `FuturesUnordered` from `runner.run().await`. No background work runs until the runner is polled.
+  - `AimDb::spawn_task` is **deleted**. Migrate to `on_start()` (collected at build) or to a private `FuturesUnordered` inside your own future.
+  - The `Runtime` bundle no longer supertrait-requires `Spawn`. Custom adapters drop `impl Spawn`.
+  - `R: Spawn` bounds are gone everywhere in `aimdb-core` (`Producer`, `Consumer`, `TypedRecord`, `TransformDescriptor`, `RecordRegistrar`, `RecordT`, `AnyRecordExt::as_typed`, remote handler/supervisor, `Database<A>`) — replaced by `R: RuntimeAdapter`.
+  - `RecordSpawner<T>` renamed to `RecordFutureCollector<T>`; its `spawn_all_tasks` → `collect_all_futures`. Internal `spawn_consumer_tasks`/`spawn_producer_service`/`spawn_transform_task` on `TypedRecord` become `collect_consumer_futures`/`collect_producer_future`/`collect_transform_futures`.
+  - Join transforms now hoist their per-input forwarder construction to build time — `JoinPipeline::into_descriptor()` returns a `CollectedTransform { task_future, fanin_futures }` and the lazy `runtime.spawn(forwarder)` inside `run_join_transform` is gone.
+  - `ConnectorBuilder::build()` now returns `Vec<BoxFuture<'static, ()>>` instead of `Arc<dyn Connector>` (which `AimDbBuilder` already discarded).
+  - Unsafe `impl Send/Sync` blocks on `Producer<T, R>` / `Consumer<T, R>` deleted — they auto-derive now.
+  - On the AimX remote-access path, three `runtime.spawn(...)` call sites bridge to `tokio::spawn` directly under `#[cfg(feature = "std")]`. These (per-connection handler, per-subscription event stream, `subscribe_record_updates`) are addressed in the AimX portability follow-up.
+- `on_start` no_std bifurcation collapsed: a single `StartFnType<R>` alias replaces the byte-identical std/no_std pair.
+
 ## [1.1.0] - 2026-05-22
 
 ### Added

aimdb-codegen/src/rust.rs

@@ -53,7 +53,7 @@ Portable code uses `R: Runtime` — no platform imports needed:
 
 ```rust
 /// Producer: reads a sensor and pushes typed values into AimDB.
-async fn sensor_producer<R: Runtime>(ctx: RuntimeContext<R>, producer: Producer<Temperature, R>) {
+async fn sensor_producer<R: Runtime>(ctx: RuntimeContext<R>, producer: Producer<Temperature>) {
     loop {
         let reading = read_sensor().await;
         producer.produce(Temperature {
@@ -65,7 +65,7 @@ async fn sensor_producer<R: Runtime>(ctx: RuntimeContext<R>, producer: Producer<
 }
 
 /// Consumer: subscribes to the buffer and reacts to every new value.
-async fn temp_logger<R: Runtime>(ctx: RuntimeContext<R>, consumer: Consumer<Temperature, R>) {
+async fn temp_logger<R: Runtime>(ctx: RuntimeContext<R>, consumer: Consumer<Temperature>) {
     let mut reader = consumer.subscribe().unwrap();
     while let Ok(temp) = reader.recv().await {
         ctx.log().info(&format!("{}: {:.1}°C", temp.sensor_id, temp.celsius));