Simplify the implementation of aimdb-sync

aimdb-sync/Cargo.toml

@@ -35,6 +35,12 @@ tokio = { version = "1.40", features = ["full", "test-util"] }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 
+# Self-dependency to force `data-contracts` on for test targets only (feature
+# unification is per-target under resolver = "2"), so `settable_integration.rs`
+# runs under plain `cargo test -p aimdb-sync` without making `data-contracts`
+# a default feature for downstream consumers of the lib.
+aimdb-sync = { path = ".", features = ["data-contracts"] }
+
 [features]
 default = ["std"]
 

aimdb-sync/src/consumer.rs

@@ -1,21 +1,17 @@
 //! Synchronous consumer for typed records.
 
+use aimdb_core::{DbError, Reader};
+
+use crate::waiter::Waiter;
 use crate::{SyncError, SyncResult};
 use std::fmt::Debug;
-use std::sync::mpsc;
-use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
 /// Synchronous consumer for records of type `T`.
 ///
-/// Thread-safe, can be cloned and shared across threads.
-/// Each clone receives data independently according to buffer semantics (SPMC, etc.).
-///
-/// # Thread Safety
-///
-/// Multiple clones of `SyncConsumer<T>` can be used concurrently from
-/// different threads. Each receives data independently based on the
-/// configured buffer type (SPMC, SingleLatest, etc.).
+/// Not thread-safe - can be moved to another thread, but not cloned.
+/// Each instance of SyncConsumer<T> reading from the same producer
+/// receives data independently according to buffer semantics (SPMC, etc.).
 ///
 /// # Example
 ///
@@ -24,7 +20,7 @@ use std::time::Duration;
 /// # use serde::{Serialize, Deserialize};
 /// # #[derive(Debug, Clone, Serialize, Deserialize)]
 /// # struct Temperature { celsius: f32 }
-/// # fn example(consumer: &SyncConsumer<Temperature>) -> SyncResult<()> {
+/// # fn example(consumer: &mut SyncConsumer<Temperature>) -> SyncResult<()> {
 /// // Get value (blocks until available)
 /// let temp = consumer.get()?;
 /// println!("Temperature: {}°C", temp.celsius);
@@ -46,22 +42,27 @@ use std::time::Duration;
 /// ```
 pub struct SyncConsumer<T>
 where
-    T: Send + Sync + 'static + Debug + Clone,
+    T: Send + Debug + Clone,
 {
-    /// Channel receiver for consumer data
-    /// Wrapped in Arc<Mutex> so it can be shared but only one thread receives at a time
-    rx: Arc<Mutex<mpsc::Receiver<T>>>,
+    waiter: Waiter,
+    reader: Reader<T>,
 }
 
 impl<T> SyncConsumer<T>
 where
-    T: Send + Sync + 'static + Debug + Clone,
+    T: Send + Debug + Clone,
 {
     /// Create a new sync consumer (internal use only)
-    pub(crate) fn new(rx: mpsc::Receiver<T>) -> Self {
-        Self {
-            rx: Arc::new(Mutex::new(rx)),
-        }
+    pub(crate) fn new(waiter: Waiter, reader: Reader<T>) -> Self {
+        Self { waiter, reader }
+    }
+
+    async fn get_impl(reader: &mut Reader<T>) -> SyncResult<T> {
+        let res = reader.recv().await;
+        res.map_err(|e| match e {
+            DbError::BufferClosed { .. } => SyncError::RuntimeShutdown,
+            e => SyncError::Db(e),
+        })
     }
 
     /// Get a value, blocking until one is available.
@@ -76,6 +77,7 @@ where
     /// # Errors
     ///
     /// - `SyncError::RuntimeShutdown` if the runtime thread has stopped
+    /// - `SyncError::Db` for other errors occured during read
     ///
     /// # Example
     ///
@@ -92,15 +94,14 @@ where
     /// let handle = AimDbBuilder::new()
     ///     .runtime(Arc::new(TokioAdapter))
     ///     .attach()?;
-    /// let consumer = handle.consumer::<MyData>("my_data")?;
+    /// let mut consumer = handle.consumer::<MyData>("my_data")?;
     /// let data = consumer.get()?; // blocks until value available
     /// println!("Got: {:?}", data);
     /// # Ok(())
     /// # }
     /// ```
-    pub fn get(&self) -> SyncResult<T> {
-        let rx = self.rx.lock().unwrap();
-        rx.recv().map_err(|_| SyncError::RuntimeShutdown)
+    pub fn get(&mut self) -> SyncResult<T> {
+        self.waiter.block_on(Self::get_impl(&mut self.reader))
     }
 
     /// Get a value with a timeout.
@@ -115,6 +116,7 @@ where
     ///
     /// - `SyncError::GetTimeout` if the timeout expires
     /// - `SyncError::RuntimeShutdown` if the runtime thread has stopped
+    /// - `SyncError::Db` for other errors occured during read
     ///
     /// # Example
     ///
@@ -132,20 +134,18 @@ where
     /// let handle = AimDbBuilder::new()
     ///     .runtime(Arc::new(TokioAdapter))
     ///     .attach()?;
-    /// let consumer = handle.consumer::<MyData>("my_data")?;
+    /// let mut consumer = handle.consumer::<MyData>("my_data")?;
     /// match consumer.get_with_timeout(Duration::from_millis(100)) {
     ///     Ok(data) => println!("Got: {:?}", data),
     ///     Err(_) => println!("No data available"),
     /// }
     /// # Ok(())
     /// # }
     /// ```
-    pub fn get_with_timeout(&self, timeout: Duration) -> SyncResult<T> {
-        let rx = self.rx.lock().unwrap();
-        rx.recv_timeout(timeout).map_err(|e| match e {
-            mpsc::RecvTimeoutError::Timeout => SyncError::GetTimeout,
-            mpsc::RecvTimeoutError::Disconnected => SyncError::RuntimeShutdown,
-        })
+    pub fn get_with_timeout(&mut self, timeout: Duration) -> SyncResult<T> {
+        let fut = async { tokio::time::timeout(timeout, Self::get_impl(&mut self.reader)).await };
+        let res = self.waiter.block_on(fut);
+        res.unwrap_or_else(|_| Err(SyncError::GetTimeout))
     }
 
     /// Try to get a value without blocking.
@@ -157,6 +157,7 @@ where
     ///
     /// - `SyncError::GetTimeout` if no data is available (non-blocking)
     /// - `SyncError::RuntimeShutdown` if the runtime thread has stopped
+    /// - `SyncError::Db` for other errors occured during read
     ///
     /// # Example
     ///
@@ -173,25 +174,26 @@ where
     /// let handle = AimDbBuilder::new()
     ///     .runtime(Arc::new(TokioAdapter))
     ///     .attach()?;
-    /// let consumer = handle.consumer::<MyData>("my_data")?;
+    /// let mut consumer = handle.consumer::<MyData>("my_data")?;
     /// match consumer.try_get() {
     ///     Ok(data) => println!("Got: {:?}", data),
     ///     Err(_) => println!("No data yet"),
     /// }
     /// # Ok(())
     /// # }
     /// ```
-    pub fn try_get(&self) -> SyncResult<T> {
-        let rx = self.rx.lock().unwrap();
-        rx.try_recv().map_err(|e| match e {
-            mpsc::TryRecvError::Empty => SyncError::GetTimeout,
-            mpsc::TryRecvError::Disconnected => SyncError::RuntimeShutdown,
+    pub fn try_get(&mut self) -> SyncResult<T> {
+        let res = self.reader.try_recv();
+        res.map_err(|e| match e {
+            DbError::BufferClosed { .. } => SyncError::RuntimeShutdown,
+            DbError::BufferEmpty => SyncError::GetTimeout,
+            e => SyncError::Db(e),
         })
     }
 
     /// Get the latest value by draining all queued values.
     ///
-    /// This method drains the internal channel to get the most recent value,
+    /// This method drains the buffer to get the most recent value,
     /// discarding any intermediate values. This is useful for SingleLatest-like
     /// semantics where you only care about the most recent data.
     ///
@@ -203,8 +205,10 @@ where
     /// The most recent available record of type `T`.
     ///
     /// # Errors
-    ///
+    /// Note that the error is only reported if no value was retrieved at all.
+    /// Errors occuring after that are ignored; the latest obtained value is returned instead.
     /// - `SyncError::RuntimeShutdown` if the runtime thread has stopped
+    /// - `SyncError::Db` if another error occured upon the very first read.
     ///
     /// # Example
     ///
@@ -221,25 +225,24 @@ where
     /// let handle = AimDbBuilder::new()
     ///     .runtime(Arc::new(TokioAdapter))
     ///     .attach()?;
-    /// let consumer = handle.consumer::<MyData>("my_data")?;
+    /// let mut consumer = handle.consumer::<MyData>("my_data")?;
     ///
     /// // Get the latest value, skipping any queued intermediate values
     /// let latest = consumer.get_latest()?;
     /// println!("Latest: {:?}", latest);
     /// # Ok(())
     /// # }
     /// ```
-    pub fn get_latest(&self) -> SyncResult<T> {
-        let rx = self.rx.lock().unwrap();
-
-        // First, block until we have at least one value
-        let mut latest = rx.recv().map_err(|_| SyncError::RuntimeShutdown)?;
-
-        // Then drain all remaining values to get the most recent
-        while let Ok(value) = rx.try_recv() {
-            latest = value;
+    pub fn get_latest(&mut self) -> SyncResult<T> {
+        // 1) can simply sequence get and try_get -
+        //    no one else does it simultaneously thanks to &mut self
+        // 2) if draining ends up with an error, we follow the previous impl
+        //    and return the latest succesfully read value
+        // 3) potentially loops forever if producer keeps producing
+        let mut latest = self.get()?;
+        while let Ok(upd) = self.try_get() {
+            latest = upd;
         }
-
         Ok(latest)
     }
 
@@ -274,7 +277,7 @@ where
     /// let handle = AimDbBuilder::new()
     ///     .runtime(Arc::new(TokioAdapter))
     ///     .attach()?;
-    /// let consumer = handle.consumer::<MyData>("my_data")?;
+    /// let mut consumer = handle.consumer::<MyData>("my_data")?;
     ///
     /// // Get the latest value within 100ms
     /// match consumer.get_latest_with_timeout(Duration::from_millis(100)) {
@@ -284,49 +287,22 @@ where
     /// # Ok(())
     /// # }
     /// ```
-    pub fn get_latest_with_timeout(&self, timeout: Duration) -> SyncResult<T> {
-        let rx = self.rx.lock().unwrap();
-
-        // First, block with timeout until we have at least one value
-        let mut latest = rx.recv_timeout(timeout).map_err(|e| match e {
-            mpsc::RecvTimeoutError::Timeout => SyncError::GetTimeout,
-            mpsc::RecvTimeoutError::Disconnected => SyncError::RuntimeShutdown,
-        })?;
-
-        // Then drain all remaining values to get the most recent
-        while let Ok(value) = rx.try_recv() {
-            latest = value;
+    pub fn get_latest_with_timeout(&mut self, timeout: Duration) -> SyncResult<T> {
+        // see internal comments for get_latest
+        let mut latest = self.get_with_timeout(timeout)?;
+        while let Ok(upd) = self.try_get() {
+            latest = upd;
         }
-
         Ok(latest)
     }
 }
 
-impl<T> Clone for SyncConsumer<T>
-where
-    T: Send + Sync + 'static + Debug + Clone,
-{
-    /// Clone the consumer to share across threads.
-    ///
-    /// Note: All clones share the same receiver, so only one thread
-    /// will receive each value. For independent subscriptions, call
-    /// `handle.consumer()` multiple times instead.
-    fn clone(&self) -> Self {
-        Self {
-            rx: self.rx.clone(),
-        }
-    }
-}
-
-// Safety: SyncConsumer uses Arc internally and is safe to send/share
-unsafe impl<T> Send for SyncConsumer<T> where T: Send + Sync + 'static + Debug + Clone {}
-unsafe impl<T> Sync for SyncConsumer<T> where T: Send + Sync + 'static + Debug + Clone {}
-
 #[cfg(test)]
 mod tests {
-    #[test]
-    fn test_sync_consumer_is_send_sync() {
-        // Just checking that the type implements Send + Sync
-        // Actual functionality tests will come later
+    // TODO: is it possible with static_assertions?
+    fn assert_send<T: Send>() {}
+    #[allow(dead_code)]
+    fn check<X: Send + std::fmt::Debug + Clone>() {
+        assert_send::<crate::SyncConsumer<X>>();
     }
 }

aimdb-sync/src/error.rs

@@ -5,9 +5,9 @@ use aimdb_core::DbError;
 
 /// Errors from the synchronous (blocking) API.
 ///
-/// Facade-specific failures (attach/detach, channel timeouts, runtime-thread
-/// shutdown) are their own variants; anything from the underlying database
-/// wraps a [`DbError`] via [`SyncError::Db`].
+/// Facade-specific failures (attach/detach, runtime-thread shutdown) are their
+/// own variants; anything from the underlying database wraps a [`DbError`]
+/// via [`SyncError::Db`].
 #[derive(Debug, thiserror::Error)]
 pub enum SyncError {
     /// Failed to attach the database to the runtime thread.