Remove latest snapshot (#119)

* refactor: remove latest_snapshot and use buffer-native peek() (M15)

Replaces the per-record latest_snapshot Mutex with a new DynBuffer::peek()
method that reads from the buffer's native storage (watch::Sender slot for
SingleLatest, Mutex<Option<T>> slot for Mailbox, None for SPMC Ring).

The snapshot was a redundant second copy of every produced value, updated
on two divergent code paths (TypedRecord::produce and RecordWriter::push),
and the only mutex on the SingleLatest hot path. After this change the
write path is unified through RecordWriter::push() and the SingleLatest
produce path is lock-free.

Also fixes a latent bug in TokioBuffer's Watch::push: it used tx.send()
which returns Err and silently drops the value when no receivers exist.
Switched to tx.send_replace() which always updates the slot — the snapshot
had been masking this for record.get callers.

Behaviour changes (documented in design 031 §Breaking Changes):
- record.get returns not_found on SPMC Ring and bufferless records
- record.get on Mailbox after record.drain returns not_found (slot was
  taken; previously the independent snapshot survived)
- metadata mark_updated now fires on AimX record.set and the WASM adapter
  produce paths, which previously bypassed it

Design: docs/design/031-M15-remove-latest-snapshot.md
Verified: make check (fmt, clippy, std/no_std/embedded/wasm builds, deny)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* refactor: streamline buffer configuration handling across modules

* refactor: rename producer_service to producer and update related methods

* feat(codec): introduce JSON codec for records with no_std support

- Added `RemoteSerialize` trait for JSON serialization/deserialization, blanket-implemented for all `serde` types.
- Implemented `JsonCodec<T>` trait for type-erased JSON encoding/decoding.
- Introduced `SerdeJsonCodec` as a zero-sized implementation of `JsonCodec`.
- Updated `TypedRecord` to use a single `remote_codec` field instead of separate serializer/deserializer closures.
- Modified `with_remote_access` to require `RemoteSerialize` and enable JSON codec installation.
- Updated `RecordValue` to utilize the new codec for JSON serialization.
- Removed the deprecated `with_read_only_serialization` method.
- Added documentation for the new codec functionality and its usage.

* refactor: update documentation and validation for remote-access records without buffer

* refactor: update dependencies and enhance buffer functionality with peek() method

* changelog: update changelogs for aimdb-core, aimdb-embassy-adapter, aimdb-tokio-adapter, and aimdb-wasm-adapter with new features, fixes, and breaking changes

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: lxsaah

CHANGELOG.md

@@ -27,8 +27,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- **M16 — JSON codec extracted behind the `json-serialize` feature; `RecordValue::as_json()` now works on `no_std + alloc`, not just `std` ([Design 032](docs/design/032-M16-aimx-json-codec.md)).** New `aimdb-core::codec` module: `RemoteSerialize` (blanket-impl'd for every `serde` `Serialize + DeserializeOwned` type), the object-safe `JsonCodec<T>`, and the zero-sized `SerdeJsonCodec`. `serde_json` runs on `alloc`, so embedded targets can opt in; `std` enables the feature transitively, so std builds are unaffected. ([aimdb-core](aimdb-core/CHANGELOG.md))
+- **Embassy buffer + join-queue tests now run in CI (Issue #85).** The join-queue tests previously sat behind `embassy-runtime`, which pulls `embassy-executor`'s cortex-m assembly and can't compile under `cargo test` on x86_64 — so ordering / backpressure / clone-routing regressions were never caught. The `join_queue` module is now gated on `embassy-sync`, and `make test` runs the embassy adapter's unit tests + doctests on the host (no executor). Also adds `EmbassyBuffer::peek()` and fixes a stale `EmbassyBuffer` doc example. ([aimdb-embassy-adapter](aimdb-embassy-adapter/CHANGELOG.md))
+
 ### Changed (breaking)
 
+- **M15 — `latest_snapshot` removed; point-in-time reads go through the new buffer-native `DynBuffer::peek()` ([Design 031](docs/design/031-M15-remove-latest-snapshot.md)).** `TypedRecord::latest()` and AimX `record.get` read the buffer directly instead of a per-record snapshot mutex (one lock + clone off the `produce()` hot path). Consequences: a `.with_remote_access()` record with **no buffer** now fails `build()` (was a silent runtime no-op); `record.get` / `latest()` on an `SpmcRing` record returns `not_found` / `None` (rings have no canonical latest — use `record.drain` / `record.subscribe`); `SingleLatest` and `Mailbox` are unaffected. `TypedRecord::produce` is removed — all writes go through `WriteHandle::push`. Adapters implement `peek()` per buffer type. ([aimdb-core](aimdb-core/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))
+- **M16 — `with_remote_access()` now requires the `json-serialize` feature (transitively enabled by `std`); `with_read_only_serialization()` removed ([Design 032](docs/design/032-M16-aimx-json-codec.md)).** The stored serializer/deserializer closures are replaced by a type-erased `Arc<dyn JsonCodec<T>>`. A `Serialize`-only record can no longer be exposed read-only over remote access. ([aimdb-core](aimdb-core/CHANGELOG.md))
+
 - **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

Cargo.lock

@@ -104,6 +104,8 @@ test:
 	cargo test --package aimdb-core --no-default-features --features "alloc,profiling"
 	@printf "$(YELLOW)  → Testing aimdb-core (no_std + alloc + metrics)$(NC)\n"
 	cargo test --package aimdb-core --no-default-features --features "alloc,metrics"
+	@printf "$(YELLOW)  → Testing aimdb-core (no_std + alloc + json-serialize)$(NC)\n"
+	cargo test --package aimdb-core --no-default-features --features "alloc,json-serialize"
 	@printf "$(YELLOW)  → Testing aimdb-core remote module$(NC)\n"
 	cargo test --package aimdb-core --lib --features "std" remote::
 	@printf "$(YELLOW)  → Testing tokio adapter$(NC)\n"
@@ -112,6 +114,8 @@ test:
 	cargo test --package aimdb-tokio-adapter --features "tokio-runtime,tracing,metrics"
 	@printf "$(YELLOW)  → Testing tokio adapter (with profiling)$(NC)\n"
 	cargo test --package aimdb-tokio-adapter --features "tokio-runtime,tracing,profiling"
+	@printf "$(YELLOW)  → Testing embassy adapter (host, no executor: buffers, join-queue, doctests)$(NC)\n"
+	cargo test --package aimdb-embassy-adapter --no-default-features --features "alloc,embassy-sync,embassy-time"
 	@printf "$(YELLOW)  → Testing sync wrapper$(NC)\n"
 	cargo test --package aimdb-sync
 	@printf "$(YELLOW)  → Testing codegen library$(NC)\n"
@@ -167,6 +171,8 @@ clippy:
 	cargo clippy --package aimdb-data-contracts --no-default-features --features alloc -- -D warnings
 	@printf "$(YELLOW)  → Clippy on aimdb-core (no_std + alloc)$(NC)\n"
 	cargo clippy --package aimdb-core --no-default-features --features alloc --all-targets -- -D warnings
+	@printf "$(YELLOW)  → Clippy on aimdb-core (no_std + alloc + json-serialize)$(NC)\n"
+	cargo clippy --package aimdb-core --no-default-features --features "alloc,json-serialize" --all-targets -- -D warnings
 	@printf "$(YELLOW)  → Clippy on aimdb-core (std)$(NC)\n"
 	cargo clippy --package aimdb-core --features "std,tracing,metrics" --all-targets -- -D warnings
 	@printf "$(YELLOW)  → Clippy on tokio adapter$(NC)\n"
@@ -253,6 +259,8 @@ test-embedded:
 	cargo check --package aimdb-data-contracts --target thumbv7em-none-eabihf --no-default-features --features alloc
 	@printf "$(YELLOW)  → Checking aimdb-core (no_std minimal) on thumbv7em-none-eabihf target$(NC)\n"
 	cargo check --package aimdb-core --target thumbv7em-none-eabihf --no-default-features --features alloc
+	@printf "$(YELLOW)  → Checking aimdb-core (no_std + alloc + json-serialize) on thumbv7em-none-eabihf target$(NC)\n"
+	cargo check --package aimdb-core --target thumbv7em-none-eabihf --no-default-features --features "alloc,json-serialize"
 	@printf "$(YELLOW)  → Checking aimdb-core (no_std/embassy) on thumbv7em-none-eabihf target$(NC)\n"
 	cargo check --package aimdb-core --target thumbv7em-none-eabihf --no-default-features --features alloc
 	@printf "$(YELLOW)  → Checking aimdb-embassy-adapter on thumbv7em-none-eabihf target$(NC)\n"

Makefile

@@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- **`json-serialize` feature + `codec` module (M16, Design 032).** New `crate::codec` module with `RemoteSerialize` (capability trait, blanket-impl'd for every `serde` `Serialize + DeserializeOwned` type), the object-safe `JsonCodec<T>` storage trait, and the zero-sized `SerdeJsonCodec`. All three are re-exported from the crate root. The feature is `no_std + alloc` compatible (`serde_json` runs on `alloc`), so `RecordValue::as_json()` now works on embedded targets, not just `std`. `std` enables `json-serialize` transitively, so existing std builds are unaffected.
+- **`DynBuffer::peek(&self) -> Option<T>` (M15, Design 031).** Non-destructive, buffer-native point-in-time read; the default impl returns `None` (correct for buffers with no canonical latest, e.g. broadcast/SPMC rings). AimX `record.get` and `TypedRecord::latest()` now route through it. Adapters implement it per buffer type — see the tokio/embassy adapter changelogs.
+
 ### Internal refactors
 
 - **AimX remote-access path is now spawn-free (Issue #114, Design 030).** Every remaining `tokio::spawn` in `aimdb-core/src/remote/` was removed; the supervisor's accept loop and each connection handler now own their own `FuturesUnordered<BoxFuture>` driven by `tokio::select! { biased; }`. Cancellation collapsed to one mechanism — dropping the future.
@@ -17,6 +22,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed (breaking)
 
+- **`latest_snapshot` removed from `TypedRecord`; `latest()` / AimX `record.get` read the buffer via `peek()` (M15, Design 031).** Eliminates one snapshot-mutex lock + `Option<T>` clone per `produce()` on the hot path. Behavioural consequences:
+  - A record configured with `.with_remote_access()` but **no buffer** now fails `build()` with a clear error (previously a silent runtime no-op — reads returned `not_found`, writes were discarded). Add a buffer, e.g. `.buffer(BufferCfg::SingleLatest)`.
+  - `record.get` / `latest()` on an `SpmcRing` record now returns `not_found` / `None` — a ring keeps per-consumer history with no canonical latest. Use `record.drain` (history) or `record.subscribe` (live). `SingleLatest` and `Mailbox` are unaffected.
+  - On `no_std`/embedded, `latest()` now depends on the adapter implementing `peek()` (the Embassy adapter does — see its changelog).
+- **`with_remote_access()` is now gated on `json-serialize` and bounded on `T: codec::RemoteSerialize` (M16, Design 032).** Same effective bound as before (`Serialize + DeserializeOwned`, blanket-impl'd), but the stored serializer/deserializer closures are replaced by a single type-erased `Arc<dyn JsonCodec<T>>`. `std` enables `json-serialize`, so std callers see no change; `no_std + alloc` callers must enable the `json-serialize` feature to call it.
+- **`producer_service` renamed to `producer` (M15).** `TypedRecord::set_producer_service` → `set_producer`, and `has_producer_service` → `has_producer` (the latter also on the `AnyRecord` trait). Affects code that called these methods directly; the public `.source()` registrar API is unchanged. Also collapses the std/no_std `cfg` split on `AnyRecord::buffer_info` / `transform_input_keys` into single signatures.
 - **`AimxConfig` lost `subscription_queue_size` (Issue #114, Design 030).** The field bounded a per-subscription mpsc channel that no longer exists — subscriptions are now one future in a `FuturesUnordered`. The builder method `.subscription_queue_size(n)` is removed; replace it with `.max_subs_per_connection(n)` if you were using the value as a soft cap on subscription count, or just delete the call.
 - **AimX `Welcome.max_subscriptions` now reports the real per-connection cap.** Previously it returned `subscription_queue_size` (default 100) while the actual cap was implicit; it now returns `max_subs_per_connection` (default 32). Clients that displayed this value will see the change.
 - **AimX `record.subscribe` response no longer carries `queue_size`.** Result object is now `{ "subscription_id": "..." }` — the previous `"queue_size"` reported a number that no longer corresponded to anything in the implementation.
@@ -25,7 +36,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - **`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`).
+  - `TypedRecord::produce` was made sync here (was `pub async fn produce`), then **removed entirely in M15** — see _Removed (breaking)_ below.
   - `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).
 
@@ -52,6 +63,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   - On the AimX remote-access path, three `runtime.spawn(...)` call sites were temporarily bridged to bare `tokio::spawn` under `#[cfg(feature = "std")]`. These have since been removed by the AimX spawn-free follow-up — see the "AimX remote-access path is now spawn-free" entry above.
 - `on_start` no_std bifurcation collapsed: a single `StartFnType<R>` alias replaces the byte-identical std/no_std pair.
 
+### Removed (breaking)
+
+- **`TypedRecord::produce` removed (M15, Design 031).** The M14 step (above) made it sync; M15 removes it entirely. All writes now go through `WriteHandle::push` via `TypedRecord::writer_handle()`. `AimDb::produce` and AimX `set_from_json` route through it; as a side effect `set_from_json` now marks record metadata as updated (previously skipped on that path). `WriteHandle` / `RecordWriter` no longer carry the snapshot mutex.
+- **`with_read_only_serialization()` removed (M16, Design 032).** A `Serialize`-only record can no longer be exposed read-only over remote access. Use `with_remote_access()`, which additionally requires `DeserializeOwned`. No in-tree callers existed.
+
 ## [1.1.0] - 2026-05-22
 
 ### Added

_external/embassy

@@ -23,14 +23,19 @@ std = [
     "serde",
     "thiserror",
     "anyhow",
-    "serde_json",
+    "json-serialize",
     "tokio",
     "aimdb-executor/std",
 ]
 
 # Heap allocation in no_std environments
 alloc = ["serde"] # Enable heap in no_std
 
+# JSON codec (`crate::codec`): serde_json-backed `RemoteSerialize` / `JsonCodec`.
+# no_std-compatible (serde_json runs on alloc); opt in on embedded targets to
+# get `record.latest()?.as_json()` without std/AimX. `std` enables it for AimX.
+json-serialize = ["alloc", "serde_json"]
+
 # Observability features (available on both std/no_std)
 tracing = ["dep:tracing"]        # Works in both std and no_std environments
 defmt = ["dep:defmt"]            # Embedded logging via probe (no_std)

aimdb-core/CHANGELOG.md

@@ -75,6 +75,22 @@ pub trait DynBuffer<T: Clone + Send>: Send + Sync {
     /// Returns self as Any for downcasting to concrete buffer types
     fn as_any(&self) -> &dyn core::any::Any;
 
+    /// Non-destructive read of the buffer's current value.
+    ///
+    /// Returns `Some(T)` if the buffer holds a current value that can be read
+    /// without affecting any consumer's position. Returns `None` if the buffer
+    /// type has no canonical "current value" concept (e.g., SPMC Ring) or if
+    /// no value has been produced yet.
+    ///
+    /// This is the buffer-native point-in-time read used by AimX `record.get`
+    /// (design 031). Implementations must not advance any reader position.
+    ///
+    /// The default returns `None`, which is the correct behaviour for buffers
+    /// without a canonical latest value.
+    fn peek(&self) -> Option<T> {
+        None
+    }
+
     /// Get buffer metrics snapshot (metrics feature only)
     ///
     /// Returns `Some(snapshot)` if the buffer implementation supports metrics,

aimdb-core/Cargo.toml

@@ -1,8 +1,8 @@
 //! `RecordWriter<T>` — the sole implementor of `WriteHandle<T>` (design 029).
 //!
-//! Pre-binds the three Arcs a `TypedRecord<T, R>` already owns (buffer,
-//! latest-snapshot, metadata tracker) so `Producer<T>` can push values without
-//! holding a `Arc<AimDb<R>>` or running a `HashMap` lookup per call.
+//! Pre-binds the buffer and (std-only) metadata tracker so `Producer<T>` can
+//! push values without holding a `Arc<AimDb<R>>` or running a `HashMap`
+//! lookup per call.
 
 #[cfg(not(feature = "std"))]
 extern crate alloc;
@@ -16,15 +16,9 @@ use std::sync::Arc;
 use super::traits::{DynBuffer, WriteHandle};
 
 pub(crate) struct RecordWriter<T: Clone + Send + 'static> {
-    /// `None` for records that only support `latest()` (no buffer configured).
+    /// `None` for records without a configured buffer.
     buffer: Option<Arc<dyn DynBuffer<T>>>,
 
-    /// Snapshot slot shared with `TypedRecord` and any `latest()` reader.
-    #[cfg(feature = "std")]
-    latest_snapshot: Arc<std::sync::Mutex<Option<T>>>,
-    #[cfg(not(feature = "std"))]
-    latest_snapshot: Arc<spin::Mutex<Option<T>>>,
-
     /// Metadata tracker (already `Clone` with shared inner `Arc<Mutex>` /
     /// `Arc<AtomicBool>`). std-only.
     #[cfg(feature = "std")]
@@ -35,39 +29,19 @@ impl<T: Clone + Send + 'static> RecordWriter<T> {
     #[cfg(feature = "std")]
     pub(crate) fn new(
         buffer: Option<Arc<dyn DynBuffer<T>>>,
-        latest_snapshot: Arc<std::sync::Mutex<Option<T>>>,
         metadata: crate::typed_record::RecordMetadataTracker,
     ) -> Self {
-        Self {
-            buffer,
-            latest_snapshot,
-            metadata,
-        }
+        Self { buffer, metadata }
     }
 
     #[cfg(not(feature = "std"))]
-    pub(crate) fn new(
-        buffer: Option<Arc<dyn DynBuffer<T>>>,
-        latest_snapshot: Arc<spin::Mutex<Option<T>>>,
-    ) -> Self {
-        Self {
-            buffer,
-            latest_snapshot,
-        }
+    pub(crate) fn new(buffer: Option<Arc<dyn DynBuffer<T>>>) -> Self {
+        Self { buffer }
     }
 }
 
 impl<T: Clone + Send + 'static> WriteHandle<T> for RecordWriter<T> {
     fn push(&self, value: T) {
-        #[cfg(feature = "std")]
-        {
-            *self.latest_snapshot.lock().unwrap() = Some(value.clone());
-        }
-        #[cfg(not(feature = "std"))]
-        {
-            *self.latest_snapshot.lock() = Some(value.clone());
-        }
-
         if let Some(buf) = &self.buffer {
             buf.push(value);
             #[cfg(feature = "std")]

aimdb-core/src/buffer/traits.rs

@@ -1102,8 +1102,10 @@ impl<R: aimdb_executor::RuntimeAdapter + 'static> AimDb<R> {
     where
         T: Send + 'static + Debug + Clone,
     {
+        // Single write path via WriteHandle (design 031). For hot paths,
+        // prefer `db.producer::<T>(key)` once and reuse the returned handle.
         let typed_rec = self.inner.get_typed_record_by_key::<T, R>(key)?;
-        typed_rec.produce(value);
+        typed_rec.writer_handle().push(value);
         Ok(())
     }