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>

Author: lxsaah

aimdb-core/src/buffer/traits.rs

@@ -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/src/buffer/writer.rs

@@ -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/builder.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(())
     }
 

aimdb-core/src/typed_record.rs

@@ -598,16 +598,6 @@ pub struct TypedRecord<
     /// Stores the deserialization logic where T: Deserialize is known at call site
     #[cfg(feature = "std")]
     json_deserializer: Option<JsonDeserializer<T>>,
-
-    /// Latest value snapshot - for latest() API
-    /// Cached atomically on every produce() call to support latest()
-    /// This provides a buffer-agnostic way to query the latest value
-    /// Available in both std and no_std environments
-    #[cfg(feature = "std")]
-    latest_snapshot: Arc<std::sync::Mutex<Option<T>>>,
-
-    #[cfg(not(feature = "std"))]
-    latest_snapshot: Arc<spin::Mutex<Option<T>>>,
 }
 
 impl<T: Send + 'static + Debug + Clone, R: aimdb_executor::RuntimeAdapter + 'static>
@@ -643,10 +633,6 @@ impl<T: Send + 'static + Debug + Clone, R: aimdb_executor::RuntimeAdapter + 'sta
             json_serializer: None,
             #[cfg(feature = "std")]
             json_deserializer: None,
-            #[cfg(feature = "std")]
-            latest_snapshot: Arc::new(std::sync::Mutex::new(None)),
-            #[cfg(not(feature = "std"))]
-            latest_snapshot: Arc::new(spin::Mutex::new(None)),
         }
     }
 
@@ -1011,16 +997,12 @@ impl<T: Send + 'static + Debug + Clone, R: aimdb_executor::RuntimeAdapter + 'sta
         {
             Arc::new(crate::buffer::RecordWriter::new(
                 self.buffer.clone(),
-                self.latest_snapshot.clone(),
                 self.metadata.clone(),
             ))
         }
         #[cfg(not(feature = "std"))]
         {
-            Arc::new(crate::buffer::RecordWriter::new(
-                self.buffer.clone(),
-                self.latest_snapshot.clone(),
-            ))
+            Arc::new(crate::buffer::RecordWriter::new(self.buffer.clone()))
         }
     }
 
@@ -1334,31 +1316,6 @@ impl<T: Send + 'static + Debug + Clone, R: aimdb_executor::RuntimeAdapter + 'sta
         }
     }
 
-    /// Produces a value by pushing to the buffer
-    ///
-    /// Enqueues value for consumer tasks and updates latest snapshot.
-    pub fn produce(&self, val: T) {
-        // Cache snapshot for latest() API (both std and no_std)
-        #[cfg(feature = "std")]
-        {
-            *self.latest_snapshot.lock().unwrap() = Some(val.clone());
-        }
-
-        #[cfg(not(feature = "std"))]
-        {
-            *self.latest_snapshot.lock() = Some(val.clone());
-        }
-
-        // Push to buffer - consumer tasks will receive it
-        if let Some(buf) = &self.buffer {
-            buf.push(val);
-
-            // Update metadata timestamp (std only)
-            #[cfg(feature = "std")]
-            self.metadata.mark_updated();
-        }
-    }
-
     /// Returns whether a producer service is registered
     pub fn has_producer_service(&self) -> bool {
         #[cfg(feature = "std")]
@@ -1399,15 +1356,15 @@ impl<T: Send + 'static + Debug + Clone, R: aimdb_executor::RuntimeAdapter + 'sta
     /// }
     /// ```
     pub fn latest(&self) -> Option<RecordValue<T>> {
+        // Read buffer-native storage via peek() (design 031). Records without
+        // a buffer return None — see Breaking Changes in design 031.
+        let value = self.buffer.as_ref()?.peek()?;
         #[cfg(feature = "std")]
         {
-            let value = self.latest_snapshot.lock().unwrap().clone()?;
             Some(RecordValue::new(value, self.json_serializer.clone()))
         }
-
         #[cfg(not(feature = "std"))]
         {
-            let value = self.latest_snapshot.lock().clone()?;
             Some(RecordValue::new(value, None))
         }
     }
@@ -1596,8 +1553,11 @@ impl<T: Send + Sync + 'static + Debug + Clone, R: aimdb_executor::RuntimeAdapter
             core::any::type_name::<T>()
         );
 
-        // Delegate to latest() which returns RecordValue<T> with serializer attached
-        let result = self.latest().and_then(|v| v.as_json());
+        // Read buffer-native storage via peek() (design 031). Records without
+        // a buffer return None — see Breaking Changes in design doc.
+        let value = self.buffer.as_ref()?.peek()?;
+        let serializer = self.json_serializer.as_ref()?;
+        let result = serializer(&value);
 
         #[cfg(feature = "tracing")]
         tracing::debug!("Serialization result: {:?}", result.is_some());
@@ -1713,7 +1673,9 @@ impl<T: Send + Sync + 'static + Debug + Clone, R: aimdb_executor::RuntimeAdapter
             core::any::type_name::<T>()
         );
 
-        self.produce(value);
+        // Push through the unified write path (design 031). This also marks
+        // metadata as updated — previously skipped on this path.
+        self.writer_handle().push(value);
 
         #[cfg(feature = "tracing")]
         tracing::info!(

aimdb-tokio-adapter/src/buffer.rs

@@ -81,7 +81,11 @@ impl<T: Clone + Send + Sync + 'static> Buffer<T> for TokioBuffer<T> {
                 let _ = tx.send(value);
             }
             TokioBufferInner::Watch { tx } => {
-                let _ = tx.send(Some(value));
+                // send_replace updates the slot unconditionally; send() would
+                // fail (and silently drop the value) when no receivers exist,
+                // which would also break peek() for producers that publish
+                // before any subscriber attaches.
+                tx.send_replace(Some(value));
             }
             TokioBufferInner::Notify { slot, notify } => {
                 *slot.lock().unwrap() = Some(value);
@@ -129,6 +133,17 @@ impl<T: Clone + Send + Sync + 'static> aimdb_core::buffer::DynBuffer<T> for Toki
         self
     }
 
+    fn peek(&self) -> Option<T> {
+        match &*self.inner {
+            // watch::Sender::borrow() reads the slot non-destructively.
+            TokioBufferInner::Watch { tx } => tx.borrow().clone(),
+            // Same Mutex the Mailbox buffer already uses for the slot.
+            TokioBufferInner::Notify { slot, .. } => slot.lock().unwrap().clone(),
+            // broadcast has no canonical latest — see design 031 §SPMC Ring.
+            TokioBufferInner::Broadcast { .. } => None,
+        }
+    }
+
     #[cfg(feature = "metrics")]
     fn metrics_snapshot(&self) -> Option<BufferMetricsSnapshot> {
         Some(<Self as BufferMetrics>::metrics(self))
@@ -1091,6 +1106,102 @@ mod tests {
         assert_eq!(remaining, vec![20, 30]);
     }
 
+    // ========================================================================
+    // peek() Tests — non-destructive buffer-native reads (design 031)
+    // ========================================================================
+
+    mod peek_tests {
+        use super::super::*;
+        use aimdb_core::buffer::DynBuffer;
+
+        #[tokio::test]
+        async fn test_peek_single_latest_empty() {
+            let buffer = TokioBuffer::<i32>::new(&BufferCfg::SingleLatest);
+            assert_eq!(buffer.peek(), None);
+        }
+
+        #[tokio::test]
+        async fn test_peek_single_latest_returns_latest() {
+            let buffer = TokioBuffer::<i32>::new(&BufferCfg::SingleLatest);
+            DynBuffer::push(&buffer, 1);
+            DynBuffer::push(&buffer, 2);
+            DynBuffer::push(&buffer, 3);
+            assert_eq!(buffer.peek(), Some(3));
+        }
+
+        #[tokio::test]
+        async fn test_peek_single_latest_is_non_destructive() {
+            let buffer = TokioBuffer::<i32>::new(&BufferCfg::SingleLatest);
+            // Subscribe BEFORE push so the receiver's version counter advances
+            // on send_replace. (Watch receivers created after a push will only
+            // wake on the *next* push — that's the gap peek() exists to fill.)
+            let mut reader = Buffer::subscribe(&buffer);
+            DynBuffer::push(&buffer, 42);
+
+            // Multiple peeks return the same value.
+            assert_eq!(buffer.peek(), Some(42));
+            assert_eq!(buffer.peek(), Some(42));
+
+            // Peek did not consume the value from the subscriber's perspective.
+            assert_eq!(reader.recv().await.unwrap(), 42);
+
+            // And peek still works after the subscriber received.
+            assert_eq!(buffer.peek(), Some(42));
+        }
+
+        #[tokio::test]
+        async fn test_peek_single_latest_works_without_subscriber() {
+            // The exact case the design 031 snapshot was originally added for:
+            // a producer pushes before anyone subscribes. peek() must see it.
+            let buffer = TokioBuffer::<i32>::new(&BufferCfg::SingleLatest);
+            DynBuffer::push(&buffer, 17);
+            assert_eq!(buffer.peek(), Some(17));
+        }
+
+        #[tokio::test]
+        async fn test_peek_mailbox_empty() {
+            let buffer = TokioBuffer::<i32>::new(&BufferCfg::Mailbox);
+            assert_eq!(buffer.peek(), None);
+        }
+
+        #[tokio::test]
+        async fn test_peek_mailbox_returns_pending() {
+            let buffer = TokioBuffer::<i32>::new(&BufferCfg::Mailbox);
+            DynBuffer::push(&buffer, 7);
+            assert_eq!(buffer.peek(), Some(7));
+        }
+
+        #[tokio::test]
+        async fn test_peek_mailbox_drained_after_recv() {
+            let buffer = TokioBuffer::<i32>::new(&BufferCfg::Mailbox);
+            DynBuffer::push(&buffer, 99);
+            assert_eq!(buffer.peek(), Some(99));
+            // Subscriber takes the slot.
+            let mut reader = Buffer::subscribe(&buffer);
+            assert_eq!(reader.recv().await.unwrap(), 99);
+            // After take(), peek sees the slot is empty.
+            assert_eq!(buffer.peek(), None);
+        }
+
+        #[tokio::test]
+        async fn test_peek_mailbox_reflects_overwrite() {
+            let buffer = TokioBuffer::<i32>::new(&BufferCfg::Mailbox);
+            DynBuffer::push(&buffer, 1);
+            DynBuffer::push(&buffer, 2);
+            assert_eq!(buffer.peek(), Some(2));
+        }
+
+        #[tokio::test]
+        async fn test_peek_spmc_ring_returns_none() {
+            // Broadcast has no canonical latest — see design 031 §SPMC Ring.
+            let buffer = TokioBuffer::<i32>::new(&BufferCfg::SpmcRing { capacity: 8 });
+            assert_eq!(buffer.peek(), None);
+            DynBuffer::push(&buffer, 1);
+            DynBuffer::push(&buffer, 2);
+            assert_eq!(buffer.peek(), None);
+        }
+    }
+
     // ========================================================================
     // Metrics Tests (feature-gated)
     // ========================================================================

aimdb-wasm-adapter/src/bindings.rs

@@ -545,12 +545,10 @@ where
     let val: T = serde_wasm_bindgen::from_value(value)
         .map_err(|e| JsError::new(&format!("Contract violation: {e}")))?;
 
-    let inner = db.inner();
-    let typed = inner
-        .get_typed_record_by_key::<T, WasmAdapter>(key)
-        .map_err(|e| JsError::new(&format!("{e:?}")))?;
-
-    typed.produce(val);
+    // Single write path via Producer<T> (design 031).
+    db.producer::<T>(key)
+        .map_err(|e| JsError::new(&format!("{e:?}")))?
+        .produce(val);
     Ok(())
 }