diff --git a/aimdb-sync/src/consumer.rs b/aimdb-sync/src/consumer.rs
index 2b519da3..c8678e4d 100644
--- a/aimdb-sync/src/consumer.rs
+++ b/aimdb-sync/src/consumer.rs
@@ -1,22 +1,17 @@
 //! Synchronous consumer for typed records.
 
+use aimdb_core::{DbError, Reader};
+
+use crate::waiter::Waiter;
 use crate::{SyncError, SyncResult};
-use alloc::sync::Arc;
 use core::fmt::Debug;
 use core::time::Duration;
-use std::sync::mpsc;
-use std::sync::Mutex;
 
 /// 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
 ///
@@ -25,7 +20,7 @@ use std::sync::Mutex;
 /// # 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);
@@ -47,22 +42,27 @@ use std::sync::Mutex;
 /// ```
 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.
@@ -77,6 +77,7 @@ where
     /// # Errors
     ///
     /// - `SyncError::RuntimeShutdown` if the runtime thread has stopped
+    /// - `SyncError::Db` for other errors occured during read
     ///
     /// # Example
     ///
@@ -92,15 +93,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 +115,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
     ///
@@ -131,7 +132,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")?;
     /// match consumer.get_with_timeout(Duration::from_millis(100)) {
     ///     Ok(data) => println!("Got: {:?}", data),
     ///     Err(_) => println!("No data available"),
@@ -139,12 +140,10 @@ where
     /// # 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.
@@ -156,6 +155,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
     ///
@@ -171,7 +171,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")?;
     /// match consumer.try_get() {
     ///     Ok(data) => println!("Got: {:?}", data),
     ///     Err(_) => println!("No data yet"),
@@ -179,17 +179,18 @@ where
     /// # 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.
     ///
@@ -201,8 +202,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
     ///
@@ -218,7 +221,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, skipping any queued intermediate values
     /// let latest = consumer.get_latest()?;
@@ -226,17 +229,16 @@ where
     /// # 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)
     }
 
@@ -270,7 +272,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)) {
@@ -280,49 +282,21 @@ 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
+    fn assert_send<T: Send>() {}
+    #[allow(dead_code)]
+    fn check<X: Send + std::fmt::Debug + Clone>() {
+        assert_send::<crate::SyncConsumer<X>>();
     }
 }
diff --git a/aimdb-sync/src/error.rs b/aimdb-sync/src/error.rs
index 65834104..e647ea29 100644
--- a/aimdb-sync/src/error.rs
+++ b/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.
diff --git a/aimdb-sync/src/handle.rs b/aimdb-sync/src/handle.rs
index f9963f27..98ed5042 100644
--- a/aimdb-sync/src/handle.rs
+++ b/aimdb-sync/src/handle.rs
@@ -1,25 +1,14 @@
 //! AimDB handle for managing the sync API runtime thread.
 
+use crate::waiter::Waiter;
 use crate::{SyncError, SyncResult};
-use aimdb_core::{log_error, log_warn, AimDb, AimDbBuilder, DbError, DbResult};
+use aimdb_core::{log_error, log_warn, AimDb, AimDbBuilder};
 use alloc::sync::Arc;
 use core::fmt::Debug;
 use core::time::Duration;
 use std::thread::{self, JoinHandle};
 use tokio::sync::mpsc;
 
-/// Default channel capacity for sync producers and consumers.
-///
-/// This is the buffer size used by `producer()` and `consumer()` methods.
-/// A capacity of 100 provides a good balance between:
-/// - Memory usage (100 × sizeof(T) per channel)
-/// - Latency (small bursts don't block)
-/// - Backpressure (prevents unbounded growth)
-///
-/// Use `producer_with_capacity()` or `consumer_with_capacity()` if you need
-/// different buffering for specific record types.
-pub const DEFAULT_SYNC_CHANNEL_CAPACITY: usize = 100;
-
 /// Extension trait to add `attach()` method to `AimDbBuilder`.
 ///
 /// This trait provides the entry point to the sync API by allowing
@@ -142,7 +131,7 @@ impl AimDbHandle {
     /// Create a new handle by spawning the runtime thread and building the database inside it.
     pub(crate) fn new_from_builder(builder: AimDbBuilder) -> SyncResult<Self> {
         // Create shutdown channel
-        let (shutdown_tx, mut shutdown_rx) = mpsc::channel::<ShutdownSignal>(1);
+        let (shutdown_tx, shutdown_rx) = mpsc::channel::<ShutdownSignal>(1);
 
         // Create channels for passing the built database and runtime handle back
         let (db_tx, mut db_rx) = mpsc::channel::<Arc<AimDb>>(1);
@@ -151,51 +140,7 @@ impl AimDbHandle {
         // Spawn the runtime thread
         let thread_handle = thread::Builder::new()
             .name("aimdb-sync-runtime".to_string())
-            .spawn(move || {
-                // Create a new Tokio runtime for this thread
-                let runtime = match tokio::runtime::Runtime::new() {
-                    Ok(rt) => rt,
-                    Err(e) => {
-                        log_error!("Failed to create Tokio runtime: {}", e);
-                        return;
-                    }
-                };
-
-                // Get the runtime handle before moving into block_on
-                let rt_handle = runtime.handle().clone();
-
-                // Send the runtime handle to the main thread
-                if handle_tx.blocking_send(rt_handle).is_err() {
-                    log_error!("Failed to send runtime handle to main thread");
-                    return;
-                }
-
-                // Build the database inside the async context
-                runtime.block_on(async move {
-                    let (db, runner) = match builder.build().await {
-                        Ok(d) => (Arc::new(d.0), d.1),
-                        Err(e) => {
-                            log_error!("Failed to build database: {}", e);
-                            return;
-                        }
-                    };
-
-                    // Send the database to the main thread
-                    if db_tx.send(db.clone()).await.is_err() {
-                        log_error!("Failed to send database to main thread");
-                        return;
-                    }
-
-                    // Drive the runner until shutdown.
-                    // If runner.run() completes early (e.g. all tap futures finish),
-                    // we must NOT drop the runtime — tasks spawned via runtime_handle
-                    // would be aborted. Keep waiting for the explicit shutdown signal.
-                    tokio::select! {
-                        _ = runner.run() => { let _ = shutdown_rx.recv().await; }
-                        _ = shutdown_rx.recv() => {}
-                    }
-                });
-            })
+            .spawn(|| Self::setup_background(builder, shutdown_rx, db_tx, handle_tx))
             .map_err(|e| SyncError::AttachFailed {
                 message: format!("Failed to spawn runtime thread: {}", e),
             })?;
@@ -222,8 +167,6 @@ impl AimDbHandle {
         })
     }
 
-    /// Create a new handle from an already-built database (legacy method).
-    #[allow(dead_code)]
     pub(crate) fn new(db: AimDb) -> SyncResult<Self> {
         // Create shutdown channel
         let (shutdown_tx, mut shutdown_rx) = mpsc::channel::<ShutdownSignal>(1);
@@ -312,7 +255,7 @@ impl AimDbHandle {
     where
         T: Send + 'static + Debug + Clone,
     {
-        self.producer_with_capacity(key, DEFAULT_SYNC_CHANNEL_CAPACITY)
+        Ok(crate::SyncProducer::new(Arc::downgrade(&self.db), key))
     }
 
     /// Create a synchronous consumer for type `T`.
@@ -329,6 +272,7 @@ impl AimDbHandle {
     ///
     /// - `DbError::RecordNotFound` if type `T` was not registered
     /// - `SyncError::RuntimeShutdown` if the runtime thread has stopped
+    /// - `SyncError::Db` for other errors upon subscribing
     ///
     /// # Example
     ///
@@ -338,7 +282,7 @@ impl AimDbHandle {
     /// # #[derive(Clone, Debug, Serialize, Deserialize)]
     /// # struct Temperature { celsius: f32 }
     /// # fn example(handle: &AimDbHandle) -> SyncResult<()> {
-    /// let consumer = handle.consumer::<Temperature>("sensor::temp")?;
+    /// let mut consumer = handle.consumer::<Temperature>("sensor::temp")?;
     /// let temp = consumer.get()?;
     /// # Ok(())
     /// # }
@@ -347,184 +291,13 @@ impl AimDbHandle {
     where
         T: Send + Sync + 'static + Debug + Clone,
     {
-        self.consumer_with_capacity(key, DEFAULT_SYNC_CHANNEL_CAPACITY)
-    }
-
-    /// Create a synchronous producer with custom channel capacity.
-    ///
-    /// Like `producer()` but allows specifying the channel buffer size.
-    /// Use this when you need different buffering characteristics for specific record types.
-    ///
-    /// # Arguments
-    ///
-    /// - `key`: The record key identifying this record instance
-    /// - `capacity`: Channel buffer size (number of items that can be buffered)
-    ///
-    /// # Type Parameters
-    ///
-    /// - `T`: The record type, must implement `TypedRecord`
-    ///
-    /// # Errors
-    ///
-    /// - `DbError::RecordNotFound` if type `T` was not registered
-    /// - `SyncError::RuntimeShutdown` if the runtime thread has stopped
-    ///
-    /// # Example
-    ///
-    /// ```no_run
-    /// # use aimdb_sync::*;
-    /// # use serde::{Serialize, Deserialize};
-    /// # #[derive(Debug, Clone, Serialize, Deserialize)]
-    /// # struct HighFrequencySensor { value: f32 }
-    /// # fn example(handle: &AimDbHandle) -> SyncResult<()> {
-    /// // High-frequency sensor needs larger buffer
-    /// let producer = handle.producer_with_capacity::<HighFrequencySensor>("sensor::high_freq", 1000)?;
-    /// producer.set(HighFrequencySensor { value: 42.0 })?;
-    /// # Ok(())
-    /// # }
-    /// ```
-    pub fn producer_with_capacity<T>(
-        &self,
-        key: impl AsRef<str>,
-        capacity: usize,
-    ) -> SyncResult<crate::SyncProducer<T>>
-    where
-        T: Send + 'static + Debug + Clone,
-    {
-        // Create a bounded tokio channel for async/sync bridging
-        // Channel carries (value, result_sender) tuples to propagate errors back
-        let (tx, mut rx) =
-            mpsc::channel::<(T, tokio::sync::oneshot::Sender<DbResult<()>>)>(capacity);
-
-        // Spawn a task on the runtime to forward values to the database
-        let db = self.db.clone();
-        let record_key = key.as_ref().to_string();
-        self.runtime_handle.spawn(async move {
-            while let Some((value, result_tx)) = rx.recv().await {
-                // Forward the value to the database's produce pipeline
-                let result = db.produce(&record_key, value);
-
-                // Send the result back to the caller (may fail if caller dropped)
-                let _ = result_tx.send(result);
-            }
-        });
-
-        Ok(crate::SyncProducer::new(tx, self.runtime_handle.clone()))
-    }
-
-    /// Create a synchronous consumer with custom channel capacity.
-    ///
-    /// Like `consumer()` but allows specifying the channel buffer size.
-    /// Use this when you need different buffering characteristics for specific record types.
-    ///
-    /// # Arguments
-    ///
-    /// - `key`: The record key identifying this record instance
-    /// - `capacity`: Channel buffer size (number of items that can be buffered)
-    ///
-    /// # Type Parameters
-    ///
-    /// - `T`: The record type, must implement `TypedRecord`
-    ///
-    /// # Errors
-    ///
-    /// - `DbError::RecordNotFound` if type `T` was not registered
-    /// - `SyncError::RuntimeShutdown` if the runtime thread has stopped
-    ///
-    /// # Example
-    ///
-    /// ```rust,no_run
-    /// # use aimdb_sync::*;
-    /// # use serde::{Serialize, Deserialize};
-    /// # #[derive(Clone, Debug, Serialize, Deserialize)]
-    /// # struct RareEvent { id: u32 }
-    /// # fn example(handle: &AimDbHandle) -> SyncResult<()> {
-    /// // Rare events need smaller buffer
-    /// let consumer = handle.consumer_with_capacity::<RareEvent>("events::rare", 10)?;
-    /// let event = consumer.get()?;
-    /// # Ok(())
-    /// # }
-    /// ```
-    pub fn consumer_with_capacity<T>(
-        &self,
-        key: impl AsRef<str>,
-        capacity: usize,
-    ) -> SyncResult<crate::SyncConsumer<T>>
-    where
-        T: Send + Sync + 'static + Debug + Clone,
-    {
-        // Create std::sync::mpsc channel for sync API
-        let (std_tx, std_rx) = std::sync::mpsc::sync_channel::<T>(capacity);
-
-        // Create a oneshot channel to confirm subscription succeeded
-        let (ready_tx, ready_rx) = tokio::sync::oneshot::channel();
-
-        // Spawn a task on the runtime to forward buffer data to the std channel
-        let db = self.db.clone();
         let record_key = key.as_ref().to_string();
-        self.runtime_handle.spawn(async move {
-            // Subscribe to the database buffer for type T
-            match db.subscribe::<T>(&record_key) {
-                Ok(mut reader) => {
-                    // Signal that subscription succeeded
-                    let _ = ready_tx.send(());
-
-                    // Forward all values from the buffer reader to the std channel
-                    loop {
-                        match reader.recv().await {
-                            Ok(value) => {
-                                // Send to std channel (non-async operation)
-                                // If the receiver is dropped, send() will fail
-                                if std_tx.send(value).is_err() {
-                                    break;
-                                }
-                            }
-                            Err(DbError::BufferLagged { lag_count, .. }) => {
-                                // Consumer fell behind - this is not fatal
-                                // Log warning but continue receiving
-                                log_warn!(
-                                    "Warning: Consumer for {} lagged by {} messages",
-                                    std::any::type_name::<T>(),
-                                    lag_count
-                                );
-                                // Don't break - next recv() will get latest data
-                            }
-                            Err(DbError::BufferClosed { .. }) => {
-                                // Buffer closed (shutdown) - exit gracefully
-                                break;
-                            }
-                            Err(e) => {
-                                // Other unexpected errors - log and stop
-                                log_error!(
-                                    "Error reading from buffer for {}: {}",
-                                    std::any::type_name::<T>(),
-                                    e
-                                );
-                                break;
-                            }
-                        }
-                    }
-                }
-                Err(e) => {
-                    log_error!(
-                        "Failed to subscribe to record type {}: {}",
-                        std::any::type_name::<T>(),
-                        e
-                    );
-                    // Signal failure (will be ignored if receiver dropped)
-                    let _ = ready_tx.send(());
-                }
-            }
-        });
-
-        // Wait for subscription to complete (with timeout)
-        ready_rx
-            .blocking_recv()
-            .map_err(|_| SyncError::AttachFailed {
-                message: format!("Failed to subscribe to {}", std::any::type_name::<T>()),
-            })?;
-
-        Ok(crate::SyncConsumer::new(std_rx))
+        let reader = self
+            .db
+            .subscribe::<T>(&record_key)
+            .map_err(lift_subscribe_error)?;
+        let waiter = Waiter::new(self.runtime_handle.clone());
+        Ok(crate::SyncConsumer::new(waiter, reader))
     }
 
     /// Gracefully shut down the runtime thread.
@@ -632,6 +405,72 @@ impl AimDbHandle {
 
         Ok(())
     }
+
+    fn setup_background(
+        builder: AimDbBuilder,
+        mut shutdown_rx: mpsc::Receiver<ShutdownSignal>,
+        db_tx: mpsc::Sender<Arc<AimDb>>,
+        handle_tx: mpsc::Sender<tokio::runtime::Handle>,
+    ) {
+        // Create a new Tokio runtime for this thread
+        let runtime = match tokio::runtime::Runtime::new() {
+            Ok(rt) => rt,
+            Err(e) => {
+                log_error!("Failed to create Tokio runtime: {}", e);
+                return;
+            }
+        };
+        // Get the runtime handle before moving into block_on
+        let rt_handle = runtime.handle().clone();
+        // Send the runtime handle to the main thread
+        if handle_tx.blocking_send(rt_handle).is_err() {
+            log_error!("Failed to send runtime handle to main thread");
+            return;
+        }
+        runtime.block_on(async move {
+            // Build the database inside the async context
+            let (db, runner) = match builder.build().await {
+                Ok(d) => (Arc::new(d.0), d.1),
+                Err(e) => {
+                    log_error!("Failed to build database: {}", e);
+                    return;
+                }
+            };
+
+            // Send the database to the main thread
+            if db_tx.send(db.clone()).await.is_err() {
+                log_error!("Failed to send database to main thread");
+                return;
+            }
+
+            // Drive the runner until shutdown.
+            // If runner.run() completes early (e.g. all tap futures finish),
+            // we must NOT drop the runtime — tasks spawned via runtime_handle
+            // would be aborted. Keep waiting for the explicit shutdown signal.
+            tokio::select! {
+                _ = runner.run() => { let _ = shutdown_rx.recv().await; }
+                _ = shutdown_rx.recv() => {}
+            }
+        });
+    }
+}
+
+fn lift_subscribe_error(e: aimdb_core::DbError) -> SyncError {
+    use aimdb_core::DbError;
+    match e {
+        DbError::BufferClosed { .. } => SyncError::RuntimeShutdown,
+        DbError::ConnectionFailed { .. } => SyncError::RuntimeShutdown,
+        DbError::RecordNotFound { record_name } => {
+            SyncError::Db(DbError::RecordNotFound { record_name })
+        }
+        DbError::RecordKeyNotFound { key } => {
+            SyncError::Db(DbError::RecordNotFound { record_name: key })
+        }
+        DbError::InvalidRecordId { id } => SyncError::Db(DbError::RecordNotFound {
+            record_name: id.to_string(),
+        }),
+        e => SyncError::Db(e),
+    }
 }
 
 impl Drop for AimDbHandle {
@@ -652,15 +491,13 @@ impl Drop for AimDbHandle {
     }
 }
 
-// Safety: AimDbHandle owns the runtime thread and channels are Send + Sync
-unsafe impl Send for AimDbHandle {}
-unsafe impl Sync for AimDbHandle {}
-
 #[cfg(test)]
 mod tests {
-    #[test]
-    fn test_extension_trait_exists() {
-        // Just ensure the module compiles
-        // Actual functionality tests will come later
+    fn assert_send<T: Send>() {}
+    fn assert_sync<T: Sync>() {}
+    #[allow(dead_code)]
+    fn check() {
+        assert_send::<crate::AimDbHandle>();
+        assert_sync::<crate::AimDbHandle>();
     }
 }
diff --git a/aimdb-sync/src/lib.rs b/aimdb-sync/src/lib.rs
index 82f00b1c..d8531834 100644
--- a/aimdb-sync/src/lib.rs
+++ b/aimdb-sync/src/lib.rs
@@ -6,23 +6,23 @@
 //! ## Overview
 //!
 //! This crate provides a synchronous interface to AimDB by running the
-//! async runtime on a dedicated background thread and using channels
-//! to bridge between synchronous and asynchronous contexts.
+//! async runtime on a dedicated background thread, blocking on it directly
+//! for reads that must wait for data.
 //!
 //! ## Features
 //!
 //! ### Producer Operations
 //! - **`set()`**: Blocking send, waits if channel is full
-//! - **`set_timeout()`**: Blocking send with timeout
 //! - **`try_set()`**: Non-blocking send, returns immediately
 //!
 //! ### Consumer Operations
 //! - **`get()`**: Blocking receive, waits for value
-//! - **`get_timeout()`**: Blocking receive with timeout
+//! - **`get_with_timeout()`**: Blocking receive with timeout
 //! - **`try_get()`**: Non-blocking receive, returns immediately
 //!
 //! ### General
-//! - **Thread-Safe**: All types are `Send + Sync` and can be shared across threads
+//! - **Thread-Safe**: `SyncProducer` is `Send + Sync` and can be cloned and shared across
+//!   threads; `SyncConsumer` is `Send` only — move it to a thread, don't share it
 //! - **Type-Safe**: Full compile-time type safety with generics
 //! - **Pure Sync Context**: No `#[tokio::main]` required - works in plain `fn main()`
 //!
@@ -67,7 +67,7 @@
 //!
 //! // Create producer and consumer
 //! let producer = handle.producer::<Temperature>("sensor.temp")?;
-//! let consumer = handle.consumer::<Temperature>("sensor.temp")?;
+//! let mut consumer = handle.consumer::<Temperature>("sensor.temp")?;
 //!
 //! // Producer: blocking operations
 //! producer.set(Temperature { celsius: 25.0 })?;
@@ -84,35 +84,31 @@
 //!
 //! ## Multi-threaded Usage
 //!
-//! Both `SyncProducer` and `SyncConsumer` can be cloned and shared across threads:
+//! `SyncProducer` can be cloned and shared across threads:
 //!
 #![cfg_attr(feature = "std", doc = "```no_run")]
 #![cfg_attr(not(feature = "std"), doc = "```ignore")]
 //! use std::thread;
 //! # use aimdb_sync::{SyncConsumer, SyncProducer};
 //! # #[derive(Debug, Clone)] struct Temperature { celsius: f32 }
-//! # fn demo(producer: SyncProducer<Temperature>, consumer: SyncConsumer<Temperature>) {
+//! # fn demo(producer: SyncProducer<Temperature>, mut consumer: SyncConsumer<Temperature>) {
 //!
 //! // Clone for use in another thread
 //! let producer_clone = producer.clone();
-//! let consumer_clone = consumer.clone();
 //!
 //! thread::spawn(move || {
 //!     producer_clone.set(Temperature { celsius: 22.0 }).ok();
 //! });
 //!
-//! thread::spawn(move || {
-//!     if let Ok(temp) = consumer_clone.get() {
-//!         println!("Got: {:.1}°C", temp.celsius);
-//!     }
-//! });
+//! if let Ok(temp) = consumer.get() {
+//!     println!("Got: {:.1}°C", temp.celsius);
+//! };
 //! # }
 //! ```
 //!
 //! ## Independent Subscriptions
 //!
-//! Note: Cloning a `SyncConsumer` shares the same channel, so only one thread
-//! will receive each value. For independent subscriptions, create multiple consumers:
+//! For independent subscriptions, create multiple consumers:
 //!
 #![cfg_attr(feature = "std", doc = "```no_run")]
 #![cfg_attr(not(feature = "std"), doc = "```ignore")]
@@ -127,84 +123,13 @@
 //! # }
 //! ```
 //!
-//! ## Channel Capacity Configuration
-//!
-//! By default, both producers and consumers use a channel capacity of 100.
-//! You can customize this per record type using the `_with_capacity` methods:
-//!
-#![cfg_attr(feature = "std", doc = "```no_run")]
-#![cfg_attr(not(feature = "std"), doc = "```ignore")]
-//! # use aimdb_sync::{AimDbHandle, SyncResult};
-//! # #[derive(Debug, Clone)] struct SensorData { value: f32 }
-//! # #[derive(Debug, Clone)] struct RareEvent { code: u8 }
-//! # #[derive(Debug, Clone)] struct LatestOnly { state: u8 }
-//! # fn demo(handle: &AimDbHandle) -> SyncResult<()> {
-//! // High-frequency sensor data needs larger buffer
-//! let producer = handle.producer_with_capacity::<SensorData>("sensor.fast", 1000)?;
-//!
-//! // Rare events can use smaller buffer
-//! let consumer = handle.consumer_with_capacity::<RareEvent>("events.rare", 10)?;
-//!
-//! // SingleLatest-like behavior: use capacity=1 to minimize queueing
-//! let consumer = handle.consumer_with_capacity::<LatestOnly>("state.latest", 1)?;
-//! # Ok(())
-//! # }
-//! ```
-//!
-//! **When to adjust capacity:**
-//! - **Increase**: High-frequency data, bursty traffic, slow consumers
-//! - **Decrease**: Memory-constrained, rare events, strict backpressure needed
-//! - **Capacity=1**: Approximate SingleLatest semantics (see limitation below)
-//! - **Default (100)**: Good for most use cases
-//!
-//! ## Buffer Semantics Limitation
-//!
-//! **Important**: The sync API adds a queueing layer (`std::sync::mpsc` channel)
-//! between the database buffer and your code. This means:
-//!
-//! - ✅ **SPMC Ring**: Works as expected - each consumer gets independent data
-//! - ✅ **Mailbox**: Works well - last value is preserved
-//! - ⚠️ **SingleLatest**: Best effort only - the sync channel may queue multiple values
-//!
-//! ### Solutions for SingleLatest Semantics
-//!
-//! 1. **Use `get_latest()`** - Drains the channel to get the most recent value:
-#![cfg_attr(feature = "std", doc = "```no_run")]
-#![cfg_attr(not(feature = "std"), doc = "```ignore")]
-//!    # use aimdb_sync::SyncResult;
-//!    # #[derive(Debug, Clone)] struct Temperature { celsius: f32 }
-//!    # fn demo(consumer: &aimdb_sync::SyncConsumer<Temperature>) -> SyncResult<()> {
-//!    // Always get the latest value, skipping queued intermediates
-//!    let latest = consumer.get_latest()?;
-//!    # Ok(())
-//!    # }
-//!    ```
-//!
-//! 2. **Use capacity=1** - Minimize queueing:
-#![cfg_attr(feature = "std", doc = "```no_run")]
-#![cfg_attr(not(feature = "std"), doc = "```ignore")]
-//!    # #[derive(Debug, Clone)] struct Temperature { celsius: f32 }
-//!    # fn demo(handle: &aimdb_sync::AimDbHandle) -> aimdb_sync::SyncResult<()> {
-//!    let consumer = handle.consumer_with_capacity::<Temperature>("sensor.temp", 1)?;
-//!    # Ok(())
-//!    # }
-//!    ```
-//!
-//! 3. **Use the async API directly** - For perfect semantic preservation.
-//!
-//! The sync API is optimized for simplicity and ease of use, not for perfect
-//! semantic preservation across all buffer types.
-//!
 //! ## Threading Model
 //!
 //! - **User threads**: Unlimited - any number of threads can call operations concurrently
 //! - **Runtime thread**: One dedicated thread named "aimdb-sync-runtime"
-//! - **Channels**: Lock-free MPSC channels for efficient communication
 //!
 //! ## Performance
 //!
-//! - **Overhead**: ~100-500μs per operation vs pure async (channel + context switch)
-//! - **Throughput**: Limited by channel capacity (default: 100 items)
 //! - **Latency**: Excellent for <50ms target, not suitable for hard low-latency requirements
 //!
 //! ## Error Handling
@@ -223,9 +148,10 @@
 //! ### Error Propagation
 //!
 //! Producer errors are propagated synchronously back to the caller:
-//! - `set()` and `set_with_timeout()` block until the produce operation completes
-//!   and return any errors that occur in the async context
-//! - `try_set()` sends immediately without waiting for the produce result (fire-and-forget)
+//! - `set()` blocks until the produce operation completes and returns any errors
+//!   that occur
+//! - `try_set()` returns immediately: `Ok(())` if the record's buffer accepted the
+//!   value, `SyncError::SetTimeout` if it didn't (bounded, non-overwriting buffer, full)
 //!
 #![cfg_attr(feature = "std", doc = "```no_run")]
 #![cfg_attr(not(feature = "std"), doc = "```ignore")]
@@ -244,7 +170,9 @@
 //!
 //! ## Safety
 //!
-//! All types are thread-safe and can be shared across threads via `Clone`.
+//! `SyncProducer` is `Clone`, `Send + Sync` — share it freely across threads.
+//! `SyncConsumer` is `Send` only, not `Clone` — move it to a thread, don't share it;
+//! get independent readers via separate `handle.consumer()` calls instead.
 //! The API ensures proper resource cleanup through RAII and explicit `detach()`.
 
 #![warn(missing_docs)]
@@ -261,11 +189,13 @@ mod error;
 mod handle;
 #[cfg(feature = "std")]
 mod producer;
+#[cfg(feature = "std")]
+mod waiter;
 
 #[cfg(feature = "std")]
 pub use consumer::SyncConsumer;
 #[cfg(feature = "std")]
-pub use handle::{AimDbBuilderSyncExt, AimDbHandle, AimDbSyncExt, DEFAULT_SYNC_CHANNEL_CAPACITY};
+pub use handle::{AimDbBuilderSyncExt, AimDbHandle, AimDbSyncExt};
 #[cfg(feature = "std")]
 pub use producer::SyncProducer;
 
diff --git a/aimdb-sync/src/producer.rs b/aimdb-sync/src/producer.rs
index 033fc0b0..61265ae0 100644
--- a/aimdb-sync/src/producer.rs
+++ b/aimdb-sync/src/producer.rs
@@ -1,16 +1,15 @@
 //! Synchronous producer for typed records.
 
 use crate::{SyncError, SyncResult};
-use aimdb_core::DbResult;
-use alloc::sync::Arc;
+use aimdb_core::{AimDb, TryProduceError};
+use alloc::sync::Weak;
 use core::fmt::Debug;
-use core::time::Duration;
-use tokio::sync::{mpsc, oneshot};
+use core::marker::PhantomData;
 
 /// Synchronous producer for records of type `T`.
 ///
 /// Thread-safe, can be cloned and shared across threads.
-/// Values are moved (not cloned) through channels for zero-copy performance.
+/// Values are moved (not cloned) directly into the record's buffer.
 ///
 /// # Thread Safety
 ///
@@ -28,32 +27,23 @@ use tokio::sync::{mpsc, oneshot};
 /// // Set value (blocks until sent)
 /// producer.set(Temperature { celsius: 25.0 })?;
 ///
-/// // Set with timeout
-/// use std::time::Duration;
-/// producer.set_with_timeout(
-///     Temperature { celsius: 26.0 },
-///     Duration::from_millis(100)
-/// )?;
-///
 /// // Try to set (non-blocking)
 /// match producer.try_set(Temperature { celsius: 27.0 }) {
 ///     Ok(()) => println!("Success"),
-///     Err(_) => println!("Channel full, try later"),
+///     Err(_) => println!("Buffer full, try later"),
 /// }
 /// # Ok(())
 /// # }
 /// ```
+#[derive(Clone)]
 pub struct SyncProducer<T>
 where
     T: Send + 'static + Debug + Clone,
 {
-    /// Channel sender for producer commands
-    /// Wrapped in Arc so it can be cloned across threads
-    /// Sends (value, result_sender) tuples to propagate produce errors back to caller
-    tx: Arc<mpsc::Sender<(T, oneshot::Sender<DbResult<()>>)>>,
-
-    /// Runtime handle for executing async operations with timeout
-    runtime_handle: tokio::runtime::Handle,
+    db: Weak<AimDb>,
+    key: String,
+    // same reasons as for Producer in aimdb-core/src/typed_api.rs
+    _phantom: PhantomData<fn() -> T>,
 }
 
 impl<T> SyncProducer<T>
@@ -61,48 +51,14 @@ where
     T: Send + 'static + Debug + Clone,
 {
     /// Create a new sync producer (internal use only)
-    pub(crate) fn new(
-        tx: mpsc::Sender<(T, oneshot::Sender<DbResult<()>>)>,
-        runtime_handle: tokio::runtime::Handle,
-    ) -> Self {
+    pub(crate) fn new(db: Weak<AimDb>, key: impl AsRef<str>) -> Self {
         Self {
-            tx: Arc::new(tx),
-            runtime_handle,
+            db,
+            key: key.as_ref().into(),
+            _phantom: PhantomData,
         }
     }
 
-    /// Internal helper: send value and wait for result with optional timeout
-    fn send_internal(&self, value: T, timeout: Option<Duration>) -> SyncResult<()> {
-        let (result_tx, result_rx) = oneshot::channel();
-        let tx = self.tx.clone();
-
-        self.runtime_handle.block_on(async move {
-            // Send with optional timeout
-            let send_result = match timeout {
-                Some(duration) => tokio::time::timeout(duration, tx.send((value, result_tx))).await,
-                None => Ok(tx.send((value, result_tx)).await),
-            };
-
-            match send_result {
-                Ok(Ok(())) => {
-                    // Successfully sent, now wait for produce result
-                    let recv_result = match timeout {
-                        Some(duration) => tokio::time::timeout(duration, result_rx).await,
-                        None => Ok(result_rx.await),
-                    };
-
-                    match recv_result {
-                        Ok(Ok(result)) => result.map_err(SyncError::from),
-                        Ok(Err(_)) => Err(SyncError::RuntimeShutdown),
-                        Err(_) => Err(SyncError::SetTimeout),
-                    }
-                }
-                Ok(Err(_)) => Err(SyncError::RuntimeShutdown),
-                Err(_) => Err(SyncError::SetTimeout),
-            }
-        })
-    }
-
     /// Set the value, blocking until it can be sent.
     ///
     /// This call will block the current thread until the value can be sent to the runtime thread.
@@ -134,57 +90,22 @@ where
     /// # }
     /// ```
     pub fn set(&self, value: T) -> SyncResult<()> {
-        self.send_internal(value, None)
-    }
-
-    /// Set the value with a timeout.
-    ///
-    /// Attempts to send the value to the runtime thread and wait for produce completion,
-    /// blocking for at most `timeout` duration.
-    ///
-    /// # Errors
-    ///
-    /// Returns `SyncError::SetTimeout` if the timeout expires before the value can be sent
-    /// or if waiting for the produce result exceeds the timeout.
-    /// Returns `SyncError::RuntimeShutdown` if the runtime thread has been detached.
-    /// Returns any error from the underlying `produce()` operation.
-    ///
-    /// # Example
-    ///
-    /// ```no_run
-    /// use aimdb_core::AimDbBuilder;
-    /// use aimdb_sync::{AimDbBuilderSyncExt, SyncResult};
-    /// use aimdb_tokio_adapter::TokioAdapter;
-    /// use std::sync::Arc;
-    /// use std::time::Duration;
-    ///
-    /// # #[derive(Debug, Clone)]
-    /// # struct MyData { value: i32 }
-    /// # fn main() -> SyncResult<()> {
-    /// let handle = AimDbBuilder::new()
-    ///     .runtime(Arc::new(TokioAdapter))
-    ///     .attach()?;
-    /// let producer = handle.producer::<MyData>("my_data")?;
-    /// producer.set_with_timeout(MyData { value: 42 }, Duration::from_millis(100))?;
-    /// # Ok(())
-    /// # }
-    /// ```
-    pub fn set_with_timeout(&self, value: T, timeout: Duration) -> SyncResult<()> {
-        self.send_internal(value, Some(timeout))
+        if let Some(db) = self.db.upgrade() {
+            db.produce(&self.key, value).map_err(SyncError::Db)
+        } else {
+            Err(SyncError::RuntimeShutdown)
+        }
     }
 
     /// Try to set the value without blocking.
     ///
-    /// Attempts to send the value immediately. Returns an error if the channel is full
-    /// or the runtime thread has shut down.
-    ///
-    /// **Note**: This method returns immediately after sending to the channel, but does NOT
-    /// wait for the produce operation to complete. Use `set()` or `set_with_timeout()` if
-    /// you need to know whether the produce operation succeeded.
+    /// Pushes the value directly into the record's buffer. Unlike `set()`, this never
+    /// blocks: it fails immediately if the buffer is full instead of waiting for space.
     ///
     /// # Errors
     ///
-    /// Returns `SyncError::SetTimeout` if the channel is full.
+    /// Returns `SyncError::SetTimeout` for bounded, non-overwriting buffer
+    /// implementations if the buffer is full.
     /// Returns `SyncError::RuntimeShutdown` if the runtime thread has been detached.
     ///
     /// # Example
@@ -204,19 +125,21 @@ where
     /// let producer = handle.producer::<MyData>("my_data")?;
     /// match producer.try_set(MyData { value: 42 }) {
     ///     Ok(()) => println!("Sent immediately"),
-    ///     Err(_) => println!("Channel full or runtime shutdown"),
+    ///     Err(_) => println!("Buffer full or runtime shutdown"),
     /// }
     /// # Ok(())
     /// # }
     /// ```
     pub fn try_set(&self, value: T) -> SyncResult<()> {
-        // Create a oneshot channel but don't wait for the result
-        let (result_tx, _result_rx) = oneshot::channel();
-
-        self.tx.try_send((value, result_tx)).map_err(|e| match e {
-            mpsc::error::TrySendError::Full(_) => SyncError::SetTimeout,
-            mpsc::error::TrySendError::Closed(_) => SyncError::RuntimeShutdown,
-        })
+        if let Some(db) = self.db.upgrade() {
+            let producer = db.producer(&self.key)?;
+            producer.try_produce(value).map_err(|e| match e {
+                TryProduceError::Full(_) => SyncError::SetTimeout,
+                TryProduceError::Closed(_) => SyncError::RuntimeShutdown,
+            })
+        } else {
+            Err(SyncError::RuntimeShutdown)
+        }
     }
 }
 
@@ -292,30 +215,13 @@ fn unix_now_ms() -> u64 {
         .as_millis() as u64
 }
 
-impl<T> Clone for SyncProducer<T>
-where
-    T: Send + 'static + Debug + Clone,
-{
-    /// Clone the producer to share across threads.
-    ///
-    /// Multiple clones can set values concurrently.
-    fn clone(&self) -> Self {
-        Self {
-            tx: self.tx.clone(),
-            runtime_handle: self.runtime_handle.clone(),
-        }
-    }
-}
-
-// Safety: SyncProducer uses Arc internally and is safe to send/share
-unsafe impl<T> Send for SyncProducer<T> where T: Send + 'static + Debug + Clone {}
-unsafe impl<T> Sync for SyncProducer<T> where T: Send + 'static + Debug + Clone {}
-
 #[cfg(test)]
 mod tests {
-    #[test]
-    fn test_sync_producer_is_send_sync() {
-        // Just checking that the type implements Send + Sync
-        // Actual functionality tests will come later
+    fn assert_send<T: Send>() {}
+    fn assert_sync<T: Sync>() {}
+    #[allow(dead_code)]
+    fn check<X: Send + 'static + core::fmt::Debug + Clone>() {
+        assert_send::<crate::SyncProducer<X>>();
+        assert_sync::<crate::SyncProducer<X>>();
     }
 }
diff --git a/aimdb-sync/src/waiter.rs b/aimdb-sync/src/waiter.rs
new file mode 100644
index 00000000..f4003261
--- /dev/null
+++ b/aimdb-sync/src/waiter.rs
@@ -0,0 +1,17 @@
+/// tokio-specific implementation of running the given future
+/// on the current thread until completion
+use std::future::Future;
+
+pub struct Waiter {
+    handle: tokio::runtime::Handle,
+}
+
+impl Waiter {
+    pub fn new(handle: tokio::runtime::Handle) -> Self {
+        Self { handle }
+    }
+
+    pub fn block_on<F: Future>(&self, fut: F) -> F::Output {
+        self.handle.block_on(fut)
+    }
+}
diff --git a/aimdb-sync/tests/integration_test.rs b/aimdb-sync/tests/integration_test.rs
index 52dee837..2e528af4 100644
--- a/aimdb-sync/tests/integration_test.rs
+++ b/aimdb-sync/tests/integration_test.rs
@@ -6,7 +6,7 @@
 #![cfg(feature = "std")]
 use aimdb_core::{buffer::BufferCfg, AimDbBuilder, DbError};
 use aimdb_sync::AimDbBuilderSyncExt;
-use aimdb_sync::SyncError;
+use aimdb_sync::{AimDbHandle, SyncConsumer, SyncError, SyncProducer};
 use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt};
 use serde::{Deserialize, Serialize};
 use std::sync::Arc;
@@ -19,43 +19,55 @@ struct TestData {
     value: String,
 }
 
-/// Test basic producer-consumer flow
-#[test]
-fn test_basic_producer_consumer() {
+fn test_value() -> TestData {
+    TestData {
+        id: 1,
+        value: "test".to_string(),
+    }
+}
+
+fn data(id: u32) -> TestData {
+    TestData {
+        id,
+        value: format!("value-{}", id),
+    }
+}
+
+fn attach(cfg: BufferCfg) -> AimDbHandle {
     let adapter = Arc::new(TokioAdapter);
     let mut builder = AimDbBuilder::new().runtime(adapter);
 
     builder.configure::<TestData>("test.data", |reg| {
-        reg.buffer(BufferCfg::SpmcRing { capacity: 10 })
-            .tap(|_ctx, _consumer| async move {
-                // No-op tap just to satisfy validation
-            });
+        reg.buffer(cfg).tap(|_ctx, _consumer| async move {
+            // No-op tap just to satisfy validation
+        });
     });
 
-    let handle = builder.attach().expect("Failed to attach");
+    builder.attach().expect("Failed to attach")
+}
 
-    // Create producer and consumer
+fn setup(cfg: BufferCfg) -> (AimDbHandle, SyncProducer<TestData>, SyncConsumer<TestData>) {
+    let handle = attach(cfg);
     let producer = handle
         .producer::<TestData>("test.data")
         .expect("Failed to create producer");
     let consumer = handle
         .consumer::<TestData>("test.data")
         .expect("Failed to create consumer");
+    (handle, producer, consumer)
+}
+
+/// Test basic producer-consumer flow
+#[test]
+fn test_basic_producer_consumer() {
+    let (handle, producer, mut consumer) = setup(BufferCfg::SpmcRing { capacity: 10 });
 
     // Produce a value
-    let test_value = TestData {
-        id: 1,
-        value: "test".to_string(),
-    };
+    let test_value = test_value();
     producer.set(test_value.clone()).expect("Failed to produce");
 
-    // Give time for async propagation
-    thread::sleep(Duration::from_millis(100));
-
-    // Consume the value (use timeout to avoid hanging)
-    let received = consumer
-        .get_with_timeout(Duration::from_secs(2))
-        .expect("Failed to consume");
+    // Consume the value
+    let received = consumer.get().expect("Failed to consume");
     assert_eq!(received, test_value);
 
     handle.detach().expect("Failed to detach");
@@ -64,23 +76,13 @@ fn test_basic_producer_consumer() {
 /// Test multiple producers and consumers
 #[test]
 fn test_multi_threaded_producer_consumer() {
-    let adapter = Arc::new(TokioAdapter);
-    let mut builder = AimDbBuilder::new().runtime(adapter);
-
-    builder.configure::<TestData>("test.data", |reg| {
-        reg.buffer(BufferCfg::SpmcRing { capacity: 100 })
-            .tap(|_ctx, _consumer| async move {
-                // No-op tap just to satisfy validation
-            });
-    });
-
-    let handle = builder.attach().expect("Failed to attach");
+    let handle = attach(BufferCfg::SpmcRing { capacity: 100 });
 
     // Create multiple consumers
-    let consumer1 = handle
+    let mut consumer1 = handle
         .consumer::<TestData>("test.data")
         .expect("Failed to create consumer 1");
-    let consumer2 = handle
+    let mut consumer2 = handle
         .consumer::<TestData>("test.data")
         .expect("Failed to create consumer 2");
 
@@ -105,9 +107,6 @@ fn test_multi_threaded_producer_consumer() {
         received
     });
 
-    // Give consumers time to start
-    thread::sleep(Duration::from_millis(50));
-
     // Create multiple producers
     let producer1 = handle
         .producer::<TestData>("test.data")
@@ -150,41 +149,18 @@ fn test_multi_threaded_producer_consumer() {
 /// Test timeout operations
 #[test]
 fn test_timeout_operations() {
-    let adapter = Arc::new(TokioAdapter);
-    let mut builder = AimDbBuilder::new().runtime(adapter);
-
-    builder.configure::<TestData>("test.data", |reg| {
-        reg.buffer(BufferCfg::SpmcRing { capacity: 10 })
-            .tap(|_ctx, _consumer| async move {
-                // No-op tap just to satisfy validation
-            });
-    });
-
-    let handle = builder.attach().expect("Failed to attach");
-
-    let producer = handle
-        .producer::<TestData>("test.data")
-        .expect("Failed to create producer");
-    let consumer = handle
-        .consumer::<TestData>("test.data")
-        .expect("Failed to create consumer");
+    let (handle, producer, mut consumer) = setup(BufferCfg::SpmcRing { capacity: 10 });
 
     // Test get_timeout on empty buffer (should timeout)
     let result = consumer.get_with_timeout(Duration::from_millis(100));
     assert!(matches!(result, Err(SyncError::GetTimeout)));
 
     // Produce a value
-    let test_value = TestData {
-        id: 1,
-        value: "test".to_string(),
-    };
+    let test_value = test_value();
     producer
-        .set_with_timeout(test_value.clone(), Duration::from_secs(1))
+        .set(test_value.clone())
         .expect("Failed to produce with timeout");
 
-    // Give more time for the value to propagate through the async pipeline
-    thread::sleep(Duration::from_millis(200));
-
     // Get with timeout (should succeed)
     let received = consumer
         .get_with_timeout(Duration::from_secs(2))
@@ -197,34 +173,14 @@ fn test_timeout_operations() {
 /// Test non-blocking operations
 #[test]
 fn test_non_blocking_operations() {
-    let adapter = Arc::new(TokioAdapter);
-    let mut builder = AimDbBuilder::new().runtime(adapter);
-
-    builder.configure::<TestData>("test.data", |reg| {
-        reg.buffer(BufferCfg::SpmcRing { capacity: 10 })
-            .tap(|_ctx, _consumer| async move {
-                // No-op tap just to satisfy validation
-            });
-    });
-
-    let handle = builder.attach().expect("Failed to attach");
-
-    let producer = handle
-        .producer::<TestData>("test.data")
-        .expect("Failed to create producer");
-    let consumer = handle
-        .consumer::<TestData>("test.data")
-        .expect("Failed to create consumer");
+    let (handle, producer, mut consumer) = setup(BufferCfg::SpmcRing { capacity: 10 });
 
     // Try get on empty buffer (should fail)
     let result = consumer.try_get();
     assert!(matches!(result, Err(SyncError::GetTimeout)));
 
     // Try set (should succeed immediately)
-    let test_value = TestData {
-        id: 1,
-        value: "test".to_string(),
-    };
+    let test_value = test_value();
     producer
         .try_set(test_value.clone())
         .expect("Failed to try_set");
@@ -246,17 +202,7 @@ fn test_non_blocking_operations() {
 /// Test graceful shutdown
 #[test]
 fn test_graceful_shutdown() {
-    let adapter = Arc::new(TokioAdapter);
-    let mut builder = AimDbBuilder::new().runtime(adapter);
-
-    builder.configure::<TestData>("test.data", |reg| {
-        reg.buffer(BufferCfg::SpmcRing { capacity: 10 })
-            .tap(|_ctx, _consumer| async move {
-                // No-op tap just to satisfy validation
-            });
-    });
-
-    let handle = builder.attach().expect("Failed to attach");
+    let handle = attach(BufferCfg::SpmcRing { capacity: 10 });
 
     let producer = handle
         .producer::<TestData>("test.data")
@@ -264,11 +210,7 @@ fn test_graceful_shutdown() {
 
     // Produce some values
     for i in 0..5 {
-        let data = TestData {
-            id: i,
-            value: format!("value-{}", i),
-        };
-        producer.set(data).expect("Failed to produce");
+        producer.set(data(i)).expect("Failed to produce");
     }
 
     // Detach should succeed
@@ -278,17 +220,7 @@ fn test_graceful_shutdown() {
 /// Test detach with timeout
 #[test]
 fn test_detach_with_timeout() {
-    let adapter = Arc::new(TokioAdapter);
-    let mut builder = AimDbBuilder::new().runtime(adapter);
-
-    builder.configure::<TestData>("test.data", |reg| {
-        reg.buffer(BufferCfg::SpmcRing { capacity: 10 })
-            .tap(|_ctx, _consumer| async move {
-                // No-op tap just to satisfy validation
-            });
-    });
-
-    let handle = builder.attach().expect("Failed to attach");
+    let handle = attach(BufferCfg::SpmcRing { capacity: 10 });
 
     // Detach with timeout should succeed quickly
     handle
@@ -299,81 +231,80 @@ fn test_detach_with_timeout() {
 /// Test error handling - runtime shutdown
 #[test]
 fn test_runtime_shutdown_error() {
-    let adapter = Arc::new(TokioAdapter);
-    let mut builder = AimDbBuilder::new().runtime(adapter);
+    let (handle, producer, mut consumer) = setup(BufferCfg::SpmcRing { capacity: 10 });
 
-    builder.configure::<TestData>("test.data", |reg| {
-        reg.buffer(BufferCfg::SpmcRing { capacity: 10 })
-            .tap(|_ctx, _consumer| async move {
-                // No-op tap just to satisfy validation
-            });
-    });
+    // Shut down the runtime
+    handle.detach().expect("Failed to detach");
 
-    let handle = builder.attach().expect("Failed to attach");
+    // Operations should now fail with RuntimeShutdown
+    let test_value = test_value();
 
-    let producer = handle
-        .producer::<TestData>("test.data")
-        .expect("Failed to create producer");
-    let consumer = handle
-        .consumer::<TestData>("test.data")
-        .expect("Failed to create consumer");
+    let result = producer.set(test_value);
+    assert!(matches!(result, Err(SyncError::RuntimeShutdown)));
+
+    let result = consumer.get();
+    assert!(matches!(result, Err(SyncError::RuntimeShutdown)));
+}
+
+/// Test error handling - runtime shutdown, non-blocking operations
+#[test]
+fn test_runtime_shutdown_error_non_blocking() {
+    let (handle, producer, mut consumer) = setup(BufferCfg::SpmcRing { capacity: 10 });
 
     // Shut down the runtime
     handle.detach().expect("Failed to detach");
 
-    // Operations should now fail with RuntimeShutdown
-    let test_value = TestData {
-        id: 1,
-        value: "test".to_string(),
-    };
+    // Non-blocking operations should now fail with RuntimeShutdown too
+    let test_value = test_value();
 
-    let result = producer.set(test_value);
+    let result = producer.try_set(test_value);
     assert!(matches!(result, Err(SyncError::RuntimeShutdown)));
 
-    let result = consumer.get_with_timeout(Duration::from_millis(100));
-    assert!(matches!(
-        result,
-        Err(SyncError::RuntimeShutdown) | Err(SyncError::GetTimeout)
-    ));
+    let result = consumer.try_get();
+    assert!(matches!(result, Err(SyncError::RuntimeShutdown)));
 }
 
-/// Test buffer semantics - SPMC Ring
+/// Test error handling - reading messages sent before the shutdown
 #[test]
-fn test_spmc_ring_semantics() {
-    let adapter = Arc::new(TokioAdapter);
-    let mut builder = AimDbBuilder::new().runtime(adapter);
+fn test_runtime_shutdown_after_produce_read_error() {
+    let (handle, producer, mut consumer) = setup(BufferCfg::SpmcRing { capacity: 10 });
 
-    builder.configure::<TestData>("test.data", |reg| {
-        reg.buffer(BufferCfg::SpmcRing { capacity: 5 })
-            .tap(|_ctx, _consumer| async move {
-                // No-op tap just to satisfy validation
-            });
-    });
+    let test_value = test_value();
 
-    let handle = builder.attach().expect("Failed to attach");
+    let result = producer.set(test_value.clone());
+    assert!(matches!(result, Ok(())));
+
+    // Shut down the runtime
+    handle.detach().expect("Failed to detach");
+
+    let result = consumer.get().expect("Failed to get the value");
+    assert_eq!(result, test_value);
+
+    let result = consumer.get();
+    println!("{:?}", result);
+    assert!(matches!(result, Err(SyncError::RuntimeShutdown)));
+}
+
+/// Test buffer semantics - SPMC Ring
+#[test]
+fn test_spmc_ring_semantics() {
+    let handle = attach(BufferCfg::SpmcRing { capacity: 5 });
 
     let producer = handle
         .producer::<TestData>("test.data")
         .expect("Failed to create producer");
-    let consumer1 = handle
+    let mut consumer1 = handle
         .consumer::<TestData>("test.data")
         .expect("Failed to create consumer 1");
-    let consumer2 = handle
+    let mut consumer2 = handle
         .consumer::<TestData>("test.data")
         .expect("Failed to create consumer 2");
 
     // Produce multiple values
     for i in 0..5 {
-        let data = TestData {
-            id: i,
-            value: format!("value-{}", i),
-        };
-        producer.set(data).expect("Failed to produce");
+        producer.set(data(i)).expect("Failed to produce");
     }
 
-    // Give time for values to propagate
-    thread::sleep(Duration::from_millis(100));
-
     // Both consumers should be able to get values independently
     let c1_data = consumer1.get().expect("Consumer 1 failed");
     let c2_data = consumer2.get().expect("Consumer 2 failed");
@@ -391,33 +322,14 @@ fn test_spmc_ring_semantics() {
 /// with the sync API by using the get_latest() method.
 #[test]
 fn test_single_latest_semantics() {
-    let adapter = Arc::new(TokioAdapter);
-    let mut builder = AimDbBuilder::new().runtime(adapter);
-
-    builder.configure::<TestData>("test.data", |reg| {
-        reg.buffer(BufferCfg::SingleLatest)
-            .tap(|_ctx, _consumer| async move {
-                // No-op tap just to satisfy validation
-            });
-    });
-
-    let handle = builder.attach().expect("Failed to attach");
-
-    let producer = handle
-        .producer::<TestData>("test.data")
-        .expect("Failed to create producer");
-
-    let consumer = handle
-        .consumer::<TestData>("test.data")
-        .expect("Failed to create consumer");
+    let (handle, producer, mut consumer) = setup(BufferCfg::SingleLatest);
 
     // Produce first value and wait for it to propagate
-    let data = TestData {
+    let initial_value = TestData {
         id: 100,
         value: "initial".to_string(),
     };
-    producer.set(data).expect("Failed to produce");
-    thread::sleep(Duration::from_millis(100));
+    producer.set(initial_value).expect("Failed to produce");
 
     // Consume first value to establish the subscription
     let first = consumer.get().expect("Failed to consume initial value");
@@ -425,17 +337,9 @@ fn test_single_latest_semantics() {
 
     // Now produce multiple values rapidly
     for i in 1..=5 {
-        let data = TestData {
-            id: i,
-            value: format!("value-{}", i),
-        };
-        producer.set(data).expect("Failed to produce value");
-        thread::sleep(Duration::from_millis(5));
+        producer.set(data(i)).expect("Failed to produce value");
     }
 
-    // Wait for all values to propagate
-    thread::sleep(Duration::from_millis(100));
-
     // Use get_latest() to drain the channel and get the most recent value
     let latest = consumer.get_latest().expect("Failed to get latest");
 
@@ -459,25 +363,7 @@ fn test_single_latest_semantics() {
 /// Test get_latest() with timeout
 #[test]
 fn test_get_latest_with_timeout() {
-    let adapter = Arc::new(TokioAdapter);
-    let mut builder = AimDbBuilder::new().runtime(adapter);
-
-    builder.configure::<TestData>("test.data", |reg| {
-        reg.buffer(BufferCfg::SingleLatest)
-            .tap(|_ctx, _consumer| async move {
-                // No-op tap just to satisfy validation
-            });
-    });
-
-    let handle = builder.attach().expect("Failed to attach");
-
-    let producer = handle
-        .producer::<TestData>("test.data")
-        .expect("Failed to create producer");
-
-    let consumer = handle
-        .consumer::<TestData>("test.data")
-        .expect("Failed to create consumer");
+    let (handle, producer, mut consumer) = setup(BufferCfg::SingleLatest);
 
     // Test timeout on empty buffer
     let result = consumer.get_latest_with_timeout(Duration::from_millis(50));
@@ -485,15 +371,9 @@ fn test_get_latest_with_timeout() {
 
     // Produce values rapidly
     for i in 1..=3 {
-        let data = TestData {
-            id: i,
-            value: format!("value-{}", i),
-        };
-        producer.set(data).expect("Failed to produce value");
+        producer.set(data(i)).expect("Failed to produce value");
     }
 
-    thread::sleep(Duration::from_millis(50));
-
     // Should get the latest value with timeout
     let latest = consumer
         .get_latest_with_timeout(Duration::from_secs(1))
@@ -517,14 +397,11 @@ fn test_error_propagation() {
     // Create a producer for an unregistered type/key
     // Note: producer creation succeeds, but set() should fail
     let producer = handle
-        .producer_with_capacity::<TestData>("test.data", 10)
+        .producer::<TestData>("test.data")
         .expect("Failed to create producer");
 
     // Try to produce a value - this should fail because the key is not registered
-    let test_value = TestData {
-        id: 1,
-        value: "test".to_string(),
-    };
+    let test_value = test_value();
 
     let result = producer.set(test_value.clone());
 
@@ -542,12 +419,5 @@ fn test_error_propagation() {
         other => panic!("Expected RecordKeyNotFound error, got: {:?}", other),
     }
 
-    // Test set_with_timeout also propagates errors
-    let result = producer.set_with_timeout(test_value, Duration::from_millis(100));
-    assert!(
-        result.is_err(),
-        "Expected produce to fail for unregistered key (with timeout)"
-    );
-
     handle.detach().expect("Failed to detach");
 }
diff --git a/aimdb-sync/tests/settable_integration.rs b/aimdb-sync/tests/settable_integration.rs
index eb15052f..0b496c35 100644
--- a/aimdb-sync/tests/settable_integration.rs
+++ b/aimdb-sync/tests/settable_integration.rs
@@ -47,7 +47,7 @@ fn set_value_constructs_produces_and_is_consumed() {
     let producer = handle
         .producer::<Temperature>("temperature")
         .expect("failed to create producer");
-    let consumer = handle
+    let mut consumer = handle
         .consumer::<Temperature>("temperature")
         .expect("failed to create consumer");
 
@@ -77,7 +77,7 @@ fn try_set_value_is_non_blocking_and_produces() {
     let producer = handle
         .producer::<Temperature>("temperature")
         .expect("failed to create producer");
-    let consumer = handle
+    let mut consumer = handle
         .consumer::<Temperature>("temperature")
         .expect("failed to create consumer");
 
@@ -109,7 +109,7 @@ fn set_value_at_stamps_the_explicit_timestamp() {
     let producer = handle
         .producer::<Temperature>("temperature")
         .expect("failed to create producer");
-    let consumer = handle
+    let mut consumer = handle
         .consumer::<Temperature>("temperature")
         .expect("failed to create consumer");
 
diff --git a/examples/sync-api-demo/src/main.rs b/examples/sync-api-demo/src/main.rs
index d29dbb6a..78f1bb1d 100644
--- a/examples/sync-api-demo/src/main.rs
+++ b/examples/sync-api-demo/src/main.rs
@@ -69,10 +69,8 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     // Step 2: Create consumers before producing
     println!("2. Creating consumers for Temperature...");
-    let consumer1 = handle.consumer::<Temperature>("sensor.temperature")?;
-    let consumer2 = handle.consumer::<Temperature>("sensor.temperature")?;
-    // Alternative with custom capacity for high-frequency data:
-    // let consumer1 = handle.consumer_with_capacity::<Temperature>("sensor.temperature", 1000)?;
+    let mut consumer1 = handle.consumer::<Temperature>("sensor.temperature")?;
+    let mut consumer2 = handle.consumer::<Temperature>("sensor.temperature")?;
     println!("   ✓ Two consumers created\n");
 
     // Step 3: Spawn consumer threads
@@ -126,8 +124,6 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
     // Step 4: Create a synchronous producer
     println!("4. Creating producer and producing values...");
     let producer = handle.producer::<Temperature>("sensor.temperature")?;
-    // Alternative with custom capacity:
-    // let producer = handle.producer_with_capacity::<Temperature>("sensor.temperature", 500)?;
     println!("   ✓ Producer created\n");
 
     // Step 5: Produce values
@@ -171,8 +167,8 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
     println!("\nThis example demonstrated:");
     println!("  • Pure synchronous context (no #[tokio::main])");
     println!("  • Multiple independent consumers");
-    println!("  • Blocking (get), timeout (get_timeout), and non-blocking (try_get) reads");
-    println!("  • Blocking (set), timeout (set_timeout), and non-blocking (try_set) writes");
+    println!("  • Blocking (get), timeout (get_with_timeout), and non-blocking (try_get) reads");
+    println!("  • Blocking (set) and non-blocking (try_set) writes");
     println!("  • Multi-threaded producer-consumer patterns");
 
     Ok(())