From e9b26eaacbdde6559e97f2c4ad268af1ad3bd82a Mon Sep 17 00:00:00 2001 From: fresheed Date: Sat, 25 Jul 2026 19:47:16 +0200 Subject: [PATCH 01/25] extracted background code; to be refactored further --- aimdb-sync/src/handle.rs | 90 ++++++++++++++++++++-------------------- 1 file changed, 45 insertions(+), 45 deletions(-) diff --git a/aimdb-sync/src/handle.rs b/aimdb-sync/src/handle.rs index f9963f2..522ff7c 100644 --- a/aimdb-sync/src/handle.rs +++ b/aimdb-sync/src/handle.rs @@ -151,51 +151,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), })?; @@ -632,6 +588,50 @@ impl AimDbHandle { Ok(()) } + + fn setup_background(builder: AimDbBuilder, mut shutdown_rx: mpsc::Receiver, db_tx: mpsc::Sender>, handle_tx: mpsc::Sender) { + // 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() => {} + } + }); +} + } impl Drop for AimDbHandle { From 94c31d70f2a0e64cfae5e0cc11dd6a649ca0f9fc Mon Sep 17 00:00:00 2001 From: fresheed Date: Sat, 25 Jul 2026 21:52:04 +0200 Subject: [PATCH 02/25] wip: simplifying producer --- aimdb-sync/src/producer.rs | 67 ++++++++++---------------------------- 1 file changed, 17 insertions(+), 50 deletions(-) diff --git a/aimdb-sync/src/producer.rs b/aimdb-sync/src/producer.rs index 033fc0b..cb78a58 100644 --- a/aimdb-sync/src/producer.rs +++ b/aimdb-sync/src/producer.rs @@ -1,11 +1,11 @@ //! Synchronous producer for typed records. -use crate::{SyncError, SyncResult}; -use aimdb_core::DbResult; -use alloc::sync::Arc; +use crate::{AimDbHandle, SyncError, SyncResult}; +use aimdb_core::{AimDb, DbResult}; +use alloc::sync::{Arc, Weak}; use core::fmt::Debug; +use core::marker::PhantomData; use core::time::Duration; -use tokio::sync::{mpsc, oneshot}; /// Synchronous producer for records of type `T`. /// @@ -47,13 +47,10 @@ pub struct SyncProducer 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>)>>, - - /// Runtime handle for executing async operations with timeout - runtime_handle: tokio::runtime::Handle, + db: Weak, + key: String, + // same reasons as for Producer in aimdb-core/src/typed_api.rs + _phantom: PhantomData T>, } impl SyncProducer @@ -61,48 +58,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>)>, - runtime_handle: tokio::runtime::Handle, - ) -> Self { + pub(crate) fn new(db: Weak, key: impl AsRef) -> 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) -> 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,7 +97,11 @@ where /// # } /// ``` pub fn set(&self, value: T) -> SyncResult<()> { - self.send_internal(value, None) + if let Some(db) = self.db.upgrade() { + db.produce(&self.key, value).map_err(|e| SyncError::Db(e)) + } else { + Err(SyncError::RuntimeShutdown) + } } /// Set the value with a timeout. From e95b99cd703254737ec41d19d891293532b8853a Mon Sep 17 00:00:00 2001 From: fresheed Date: Mon, 27 Jul 2026 00:07:25 +0200 Subject: [PATCH 03/25] very wip --- aimdb-sync/src/handle.rs | 34 ++++++------------- aimdb-sync/src/producer.rs | 68 ++++++++------------------------------ 2 files changed, 24 insertions(+), 78 deletions(-) diff --git a/aimdb-sync/src/handle.rs b/aimdb-sync/src/handle.rs index 522ff7c..651a85b 100644 --- a/aimdb-sync/src/handle.rs +++ b/aimdb-sync/src/handle.rs @@ -2,7 +2,7 @@ use crate::{SyncError, SyncResult}; use aimdb_core::{log_error, log_warn, AimDb, AimDbBuilder, DbError, DbResult}; -use alloc::sync::Arc; +use alloc::sync::{Arc, Weak}; use core::fmt::Debug; use core::time::Duration; use std::thread::{self, JoinHandle}; @@ -347,25 +347,7 @@ impl AimDbHandle { 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>)>(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())) + Ok(crate::SyncProducer::new(Arc::downgrade(&self.db), key)) } /// Create a synchronous consumer with custom channel capacity. @@ -589,7 +571,12 @@ impl AimDbHandle { Ok(()) } - fn setup_background(builder: AimDbBuilder, mut shutdown_rx: mpsc::Receiver, db_tx: mpsc::Sender>, handle_tx: mpsc::Sender) { + fn setup_background( + builder: AimDbBuilder, + mut shutdown_rx: mpsc::Receiver, + db_tx: mpsc::Sender>, + handle_tx: mpsc::Sender, + ) { // Create a new Tokio runtime for this thread let runtime = match tokio::runtime::Runtime::new() { Ok(rt) => rt, @@ -629,9 +616,8 @@ impl AimDbHandle { _ = runner.run() => { let _ = shutdown_rx.recv().await; } _ = shutdown_rx.recv() => {} } - }); -} - + }); + } } impl Drop for AimDbHandle { diff --git a/aimdb-sync/src/producer.rs b/aimdb-sync/src/producer.rs index cb78a58..59061e8 100644 --- a/aimdb-sync/src/producer.rs +++ b/aimdb-sync/src/producer.rs @@ -1,7 +1,7 @@ //! Synchronous producer for typed records. use crate::{AimDbHandle, SyncError, SyncResult}; -use aimdb_core::{AimDb, DbResult}; +use aimdb_core::{AimDb, DbResult, TryProduceError}; use alloc::sync::{Arc, Weak}; use core::fmt::Debug; use core::marker::PhantomData; @@ -28,13 +28,6 @@ use core::time::Duration; /// // 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"), @@ -104,49 +97,13 @@ where } } - /// 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::("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)) - } - /// 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 + /// wait for the produce operation to complete. Use `set()` if /// you need to know whether the produce operation succeeded. /// /// # Errors @@ -177,13 +134,15 @@ where /// # } /// ``` 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) + } } } @@ -268,8 +227,9 @@ where /// Multiple clones can set values concurrently. fn clone(&self) -> Self { Self { - tx: self.tx.clone(), - runtime_handle: self.runtime_handle.clone(), + db: self.db.clone(), + key: self.key.clone(), + _phantom: PhantomData, } } } From 8f1942800ecfd80fd9cb7da21c64bdf71be3822b Mon Sep 17 00:00:00 2001 From: fresheed Date: Mon, 27 Jul 2026 17:33:28 +0200 Subject: [PATCH 04/25] extracted buffer forwarding of consumer --- aimdb-sync/src/handle.rs | 83 ++++++++++++++++++++++------------------ 1 file changed, 45 insertions(+), 38 deletions(-) diff --git a/aimdb-sync/src/handle.rs b/aimdb-sync/src/handle.rs index 651a85b..59984ba 100644 --- a/aimdb-sync/src/handle.rs +++ b/aimdb-sync/src/handle.rs @@ -403,45 +403,9 @@ impl AimDbHandle { self.runtime_handle.spawn(async move { // Subscribe to the database buffer for type T match db.subscribe::(&record_key) { - Ok(mut reader) => { - // Signal that subscription succeeded + Ok(reader) => { 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::(), - 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::(), - e - ); - break; - } - } - } + Self::forward_buffered(std_tx, reader).await; } Err(e) => { log_error!( @@ -618,6 +582,49 @@ impl AimDbHandle { } }); } + + async fn forward_buffered( + std_tx: std::sync::mpsc::SyncSender, + mut reader: aimdb_core::Reader, + ) where + T: Send + Clone, + { + // 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::(), + 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::(), + e + ); + break; + } + } + } + } } impl Drop for AimDbHandle { From 2789b8db6cb3e87fca61f05a369107d73e96f44f Mon Sep 17 00:00:00 2001 From: fresheed Date: Mon, 27 Jul 2026 23:07:20 +0200 Subject: [PATCH 05/25] wip: updating consumer --- aimdb-sync/src/consumer.rs | 14 +++++++------- aimdb-sync/src/lib.rs | 1 + aimdb-sync/src/waiter.rs | 15 +++++++++++++++ 3 files changed, 23 insertions(+), 7 deletions(-) create mode 100644 aimdb-sync/src/waiter.rs diff --git a/aimdb-sync/src/consumer.rs b/aimdb-sync/src/consumer.rs index 2b519da..b69c383 100644 --- a/aimdb-sync/src/consumer.rs +++ b/aimdb-sync/src/consumer.rs @@ -1,5 +1,8 @@ //! Synchronous consumer for typed records. +use aimdb_core::buffer::BufferReader; + +use crate::waiter::{BlockingBridge, Waiter}; use crate::{SyncError, SyncResult}; use alloc::sync::Arc; use core::fmt::Debug; @@ -49,9 +52,8 @@ pub struct SyncConsumer where T: Send + Sync + 'static + Debug + Clone, { - /// Channel receiver for consumer data - /// Wrapped in `Arc` so it can be shared but only one thread receives at a time - rx: Arc>>, + waiter: Waiter, + reader: Box + Send>, } impl SyncConsumer @@ -59,10 +61,8 @@ where T: Send + Sync + 'static + Debug + Clone, { /// Create a new sync consumer (internal use only) - pub(crate) fn new(rx: mpsc::Receiver) -> Self { - Self { - rx: Arc::new(Mutex::new(rx)), - } + pub(crate) fn new(waiter: Waiter, reader: Box + Send>) -> Self { + Self { waiter, reader } } /// Get a value, blocking until one is available. diff --git a/aimdb-sync/src/lib.rs b/aimdb-sync/src/lib.rs index 82f00b1..f9931ea 100644 --- a/aimdb-sync/src/lib.rs +++ b/aimdb-sync/src/lib.rs @@ -261,6 +261,7 @@ mod error; mod handle; #[cfg(feature = "std")] mod producer; +mod waiter; #[cfg(feature = "std")] pub use consumer::SyncConsumer; diff --git a/aimdb-sync/src/waiter.rs b/aimdb-sync/src/waiter.rs new file mode 100644 index 0000000..6f119ae --- /dev/null +++ b/aimdb-sync/src/waiter.rs @@ -0,0 +1,15 @@ +/// Runtime-specific implementations of Waiter define how to +/// running the given future on the current thread until completion +use std::future::Future; + +#[cfg(feature = "std")] +pub struct Waiter { + handle: tokio::runtime::Handle, +} + +#[cfg(feature = "std")] +impl Waiter { + pub fn block_on(&self, fut: F) -> F::Output { + self.handle.block_on(fut) + } +} From f8587026e75cbf3577d9bb07be278cf978f9879b Mon Sep 17 00:00:00 2001 From: fresheed Date: Tue, 28 Jul 2026 23:04:36 +0200 Subject: [PATCH 06/25] wip: rewriting consumer, need clarifications from maintainer --- aimdb-sync/src/consumer.rs | 76 +++++++++++++++++++------------------- 1 file changed, 37 insertions(+), 39 deletions(-) diff --git a/aimdb-sync/src/consumer.rs b/aimdb-sync/src/consumer.rs index b69c383..62b0215 100644 --- a/aimdb-sync/src/consumer.rs +++ b/aimdb-sync/src/consumer.rs @@ -1,6 +1,7 @@ //! Synchronous consumer for typed records. use aimdb_core::buffer::BufferReader; +use aimdb_core::{DbError, Reader}; use crate::waiter::{BlockingBridge, Waiter}; use crate::{SyncError, SyncResult}; @@ -53,7 +54,7 @@ where T: Send + Sync + 'static + Debug + Clone, { waiter: Waiter, - reader: Box + Send>, + reader: Reader, } impl SyncConsumer @@ -61,10 +62,18 @@ where T: Send + Sync + 'static + Debug + Clone, { /// Create a new sync consumer (internal use only) - pub(crate) fn new(waiter: Waiter, reader: Box + Send>) -> Self { + pub(crate) fn new(waiter: Waiter, reader: Reader) -> Self { Self { waiter, reader } } + async fn get_impl(reader: &mut Reader) -> SyncResult { + 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. /// /// Blocks indefinitely until a value is available from the @@ -98,9 +107,8 @@ where /// # Ok(()) /// # } /// ``` - pub fn get(&self) -> SyncResult { - let rx = self.rx.lock().unwrap(); - rx.recv().map_err(|_| SyncError::RuntimeShutdown) + pub fn get(&mut self) -> SyncResult { + self.waiter.block_on(Self::get_impl(&mut self.reader)) } /// Get a value with a timeout. @@ -139,12 +147,10 @@ where /// # Ok(()) /// # } /// ``` - pub fn get_with_timeout(&self, timeout: Duration) -> SyncResult { - 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 { + let fut = tokio::time::timeout(timeout, Self::get_impl(&mut self.reader)); + let res = self.waiter.block_on(fut); + res.unwrap_or_else(|_| Err(SyncError::GetTimeout)) } /// Try to get a value without blocking. @@ -179,11 +185,12 @@ where /// # Ok(()) /// # } /// ``` - pub fn try_get(&self) -> SyncResult { - 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 { + let res = self.reader.try_recv(); + res.map_err(|e| match e { + DbError::BufferClosed { .. } => SyncError::RuntimeShutdown, + DbError::BufferEmpty => SyncError::GetTimeout, + e => SyncError::Db(e), }) } @@ -226,17 +233,16 @@ where /// # Ok(()) /// # } /// ``` - pub fn get_latest(&self) -> SyncResult { - 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 { + // 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) } @@ -280,20 +286,12 @@ where /// # Ok(()) /// # } /// ``` - pub fn get_latest_with_timeout(&self, timeout: Duration) -> SyncResult { - 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 { + // 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) } } From 899aa24046317ae96d6ba68af929e6ae2024db17 Mon Sep 17 00:00:00 2001 From: fresheed Date: Wed, 29 Jul 2026 22:27:28 +0200 Subject: [PATCH 07/25] code seems to be fixed --- aimdb-sync/src/consumer.rs | 40 ++++++------- aimdb-sync/src/handle.rs | 111 +++++++++++-------------------------- aimdb-sync/src/producer.rs | 17 +++++- aimdb-sync/src/waiter.rs | 4 ++ 4 files changed, 67 insertions(+), 105 deletions(-) diff --git a/aimdb-sync/src/consumer.rs b/aimdb-sync/src/consumer.rs index 62b0215..9f20ad0 100644 --- a/aimdb-sync/src/consumer.rs +++ b/aimdb-sync/src/consumer.rs @@ -1,9 +1,8 @@ //! Synchronous consumer for typed records. -use aimdb_core::buffer::BufferReader; use aimdb_core::{DbError, Reader}; -use crate::waiter::{BlockingBridge, Waiter}; +use crate::waiter::Waiter; use crate::{SyncError, SyncResult}; use alloc::sync::Arc; use core::fmt::Debug; @@ -13,6 +12,8 @@ use std::sync::Mutex; /// Synchronous consumer for records of type `T`. /// +/// TODO: **doc below is wrong, update it** +/// /// Thread-safe, can be cloned and shared across threads. /// Each clone receives data independently according to buffer semantics (SPMC, etc.). /// @@ -51,15 +52,26 @@ use std::sync::Mutex; /// ``` pub struct SyncConsumer where - T: Send + Sync + 'static + Debug + Clone, + T: Send + 'static + Debug + Clone, { waiter: Waiter, reader: Reader, } +// TODO: remove or replace with static_assertions +const _: () = { + fn assert_send() {} + // fn assert_sync() {} + + fn check() { + assert_send::>(); + // assert_sync::(); + } +}; + impl SyncConsumer where - T: Send + Sync + 'static + Debug + Clone, + T: Send + 'static + Debug + Clone, { /// Create a new sync consumer (internal use only) pub(crate) fn new(waiter: Waiter, reader: Reader) -> Self { @@ -296,26 +308,6 @@ where } } -impl Clone for SyncConsumer -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 Send for SyncConsumer where T: Send + Sync + 'static + Debug + Clone {} -unsafe impl Sync for SyncConsumer where T: Send + Sync + 'static + Debug + Clone {} - #[cfg(test)] mod tests { #[test] diff --git a/aimdb-sync/src/handle.rs b/aimdb-sync/src/handle.rs index 59984ba..7615d8d 100644 --- a/aimdb-sync/src/handle.rs +++ b/aimdb-sync/src/handle.rs @@ -1,5 +1,6 @@ //! 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 alloc::sync::{Arc, Weak}; @@ -391,42 +392,39 @@ impl AimDbHandle { 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::(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::(&record_key) { - Ok(reader) => { - let _ = ready_tx.send(()); - Self::forward_buffered(std_tx, reader).await; - } - Err(e) => { - log_error!( - "Failed to subscribe to record type {}: {}", - std::any::type_name::(), - 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::()), - })?; - - Ok(crate::SyncConsumer::new(std_rx)) + let reader = self.db.subscribe::(&record_key).map_err(SyncError::Db)?; + let waiter = Waiter::new(self.runtime_handle.clone()); + Ok(crate::SyncConsumer::new(waiter, reader)) + + // match { + // Ok(reader) => { + // let _ = ready_tx.send(()); + // Self::forward_buffered(std_tx, reader).await; + // } + // Err(e) => { + // log_error!( + // "Failed to subscribe to record type {}: {}", + // std::any::type_name::(), + // e + // ); + // // Signal failure (will be ignored if receiver dropped) + // let _ = ready_tx.send(()); + // } + // } + + // self.runtime_handle.spawn(async move { + // // Subscribe to the database buffer for type T + // }); + + // // 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::()), + // })?; + + // Ok(crate::SyncConsumer::new(std_rx)) } /// Gracefully shut down the runtime thread. @@ -582,49 +580,6 @@ impl AimDbHandle { } }); } - - async fn forward_buffered( - std_tx: std::sync::mpsc::SyncSender, - mut reader: aimdb_core::Reader, - ) where - T: Send + Clone, - { - // 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::(), - 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::(), - e - ); - break; - } - } - } - } } impl Drop for AimDbHandle { diff --git a/aimdb-sync/src/producer.rs b/aimdb-sync/src/producer.rs index 59061e8..22b03ad 100644 --- a/aimdb-sync/src/producer.rs +++ b/aimdb-sync/src/producer.rs @@ -234,9 +234,20 @@ where } } -// Safety: SyncProducer uses Arc internally and is safe to send/share -unsafe impl Send for SyncProducer where T: Send + 'static + Debug + Clone {} -unsafe impl Sync for SyncProducer where T: Send + 'static + Debug + Clone {} +// // Safety: SyncProducer uses Arc internally and is safe to send/share +// unsafe impl Send for SyncProducer where T: Send + 'static + Debug + Clone {} +// unsafe impl Sync for SyncProducer where T: Send + 'static + Debug + Clone {} + +// TODO: remove or replace with static_assertions +const _: () = { + fn assert_send() {} + fn assert_sync() {} + + fn check() { + assert_send::>(); + assert_sync::>(); + } +}; #[cfg(test)] mod tests { diff --git a/aimdb-sync/src/waiter.rs b/aimdb-sync/src/waiter.rs index 6f119ae..c7c405f 100644 --- a/aimdb-sync/src/waiter.rs +++ b/aimdb-sync/src/waiter.rs @@ -9,6 +9,10 @@ pub struct Waiter { #[cfg(feature = "std")] impl Waiter { + pub fn new(handle: tokio::runtime::Handle) -> Self { + Self { handle } + } + pub fn block_on(&self, fut: F) -> F::Output { self.handle.block_on(fut) } From 44bbb1210ac3c7c939b662f54b120fc78886f5cb Mon Sep 17 00:00:00 2001 From: fresheed Date: Wed, 29 Jul 2026 23:26:08 +0200 Subject: [PATCH 08/25] integration tests passing; units are failing --- aimdb-sync/src/consumer.rs | 2 +- aimdb-sync/tests/integration_test.rs | 29 +++++++++++----------------- 2 files changed, 12 insertions(+), 19 deletions(-) diff --git a/aimdb-sync/src/consumer.rs b/aimdb-sync/src/consumer.rs index 9f20ad0..23201da 100644 --- a/aimdb-sync/src/consumer.rs +++ b/aimdb-sync/src/consumer.rs @@ -160,7 +160,7 @@ where /// # } /// ``` pub fn get_with_timeout(&mut self, timeout: Duration) -> SyncResult { - let fut = tokio::time::timeout(timeout, Self::get_impl(&mut self.reader)); + 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)) } diff --git a/aimdb-sync/tests/integration_test.rs b/aimdb-sync/tests/integration_test.rs index 52dee83..7c41395 100644 --- a/aimdb-sync/tests/integration_test.rs +++ b/aimdb-sync/tests/integration_test.rs @@ -38,7 +38,7 @@ fn test_basic_producer_consumer() { let producer = handle .producer::("test.data") .expect("Failed to create producer"); - let consumer = handle + let mut consumer = handle .consumer::("test.data") .expect("Failed to create consumer"); @@ -77,10 +77,10 @@ fn test_multi_threaded_producer_consumer() { let handle = builder.attach().expect("Failed to attach"); // Create multiple consumers - let consumer1 = handle + let mut consumer1 = handle .consumer::("test.data") .expect("Failed to create consumer 1"); - let consumer2 = handle + let mut consumer2 = handle .consumer::("test.data") .expect("Failed to create consumer 2"); @@ -165,7 +165,7 @@ fn test_timeout_operations() { let producer = handle .producer::("test.data") .expect("Failed to create producer"); - let consumer = handle + let mut consumer = handle .consumer::("test.data") .expect("Failed to create consumer"); @@ -179,7 +179,7 @@ fn test_timeout_operations() { value: "test".to_string(), }; 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 @@ -212,7 +212,7 @@ fn test_non_blocking_operations() { let producer = handle .producer::("test.data") .expect("Failed to create producer"); - let consumer = handle + let mut consumer = handle .consumer::("test.data") .expect("Failed to create consumer"); @@ -314,7 +314,7 @@ fn test_runtime_shutdown_error() { let producer = handle .producer::("test.data") .expect("Failed to create producer"); - let consumer = handle + let mut consumer = handle .consumer::("test.data") .expect("Failed to create consumer"); @@ -355,10 +355,10 @@ fn test_spmc_ring_semantics() { let producer = handle .producer::("test.data") .expect("Failed to create producer"); - let consumer1 = handle + let mut consumer1 = handle .consumer::("test.data") .expect("Failed to create consumer 1"); - let consumer2 = handle + let mut consumer2 = handle .consumer::("test.data") .expect("Failed to create consumer 2"); @@ -407,7 +407,7 @@ fn test_single_latest_semantics() { .producer::("test.data") .expect("Failed to create producer"); - let consumer = handle + let mut consumer = handle .consumer::("test.data") .expect("Failed to create consumer"); @@ -475,7 +475,7 @@ fn test_get_latest_with_timeout() { .producer::("test.data") .expect("Failed to create producer"); - let consumer = handle + let mut consumer = handle .consumer::("test.data") .expect("Failed to create consumer"); @@ -542,12 +542,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"); } From fc5e1f6e9fcea92cfdb7a6624949dca1a81c0c38 Mon Sep 17 00:00:00 2001 From: fresheed Date: Sat, 1 Aug 2026 10:37:33 +0200 Subject: [PATCH 09/25] moved compile-time checks for Send/Sync --- aimdb-sync/src/consumer.rs | 32 +++++++++++--------------------- aimdb-sync/src/handle.rs | 35 +++-------------------------------- aimdb-sync/src/lib.rs | 17 +++++++---------- aimdb-sync/src/producer.rs | 28 ++++++++-------------------- 4 files changed, 29 insertions(+), 83 deletions(-) diff --git a/aimdb-sync/src/consumer.rs b/aimdb-sync/src/consumer.rs index 23201da..bfba484 100644 --- a/aimdb-sync/src/consumer.rs +++ b/aimdb-sync/src/consumer.rs @@ -30,7 +30,7 @@ use std::sync::Mutex; /// # use serde::{Serialize, Deserialize}; /// # #[derive(Debug, Clone, Serialize, Deserialize)] /// # struct Temperature { celsius: f32 } -/// # fn example(consumer: &SyncConsumer) -> SyncResult<()> { +/// # fn example(consumer: &mut SyncConsumer) -> SyncResult<()> { /// // Get value (blocks until available) /// let temp = consumer.get()?; /// println!("Temperature: {}°C", temp.celsius); @@ -58,17 +58,6 @@ where reader: Reader, } -// TODO: remove or replace with static_assertions -const _: () = { - fn assert_send() {} - // fn assert_sync() {} - - fn check() { - assert_send::>(); - // assert_sync::(); - } -}; - impl SyncConsumer where T: Send + 'static + Debug + Clone, @@ -113,7 +102,7 @@ where /// let handle = AimDbBuilder::new() /// .runtime(Arc::new(TokioAdapter)) /// .attach()?; - /// let consumer = handle.consumer::("my_data")?; + /// let mut consumer = handle.consumer::("my_data")?; /// let data = consumer.get()?; // blocks until value available /// println!("Got: {:?}", data); /// # Ok(()) @@ -151,7 +140,7 @@ where /// let handle = AimDbBuilder::new() /// .runtime(Arc::new(TokioAdapter)) /// .attach()?; - /// let consumer = handle.consumer::("my_data")?; + /// let mut consumer = handle.consumer::("my_data")?; /// match consumer.get_with_timeout(Duration::from_millis(100)) { /// Ok(data) => println!("Got: {:?}", data), /// Err(_) => println!("No data available"), @@ -189,7 +178,7 @@ where /// let handle = AimDbBuilder::new() /// .runtime(Arc::new(TokioAdapter)) /// .attach()?; - /// let consumer = handle.consumer::("my_data")?; + /// let mut consumer = handle.consumer::("my_data")?; /// match consumer.try_get() { /// Ok(data) => println!("Got: {:?}", data), /// Err(_) => println!("No data yet"), @@ -237,7 +226,7 @@ where /// let handle = AimDbBuilder::new() /// .runtime(Arc::new(TokioAdapter)) /// .attach()?; - /// let consumer = handle.consumer::("my_data")?; + /// let mut consumer = handle.consumer::("my_data")?; /// /// // Get the latest value, skipping any queued intermediate values /// let latest = consumer.get_latest()?; @@ -288,7 +277,7 @@ where /// let handle = AimDbBuilder::new() /// .runtime(Arc::new(TokioAdapter)) /// .attach()?; - /// let consumer = handle.consumer::("my_data")?; + /// let mut consumer = handle.consumer::("my_data")?; /// /// // Get the latest value within 100ms /// match consumer.get_latest_with_timeout(Duration::from_millis(100)) { @@ -310,9 +299,10 @@ where #[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() {} + #[allow(dead_code)] + fn check() { + assert_send::>(); } } diff --git a/aimdb-sync/src/handle.rs b/aimdb-sync/src/handle.rs index 7615d8d..d95305f 100644 --- a/aimdb-sync/src/handle.rs +++ b/aimdb-sync/src/handle.rs @@ -143,7 +143,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 { // Create shutdown channel - let (shutdown_tx, mut shutdown_rx) = mpsc::channel::(1); + let (shutdown_tx, shutdown_rx) = mpsc::channel::(1); // Create channels for passing the built database and runtime handle back let (db_tx, mut db_rx) = mpsc::channel::>(1); @@ -295,7 +295,7 @@ impl AimDbHandle { /// # #[derive(Clone, Debug, Serialize, Deserialize)] /// # struct Temperature { celsius: f32 } /// # fn example(handle: &AimDbHandle) -> SyncResult<()> { - /// let consumer = handle.consumer::("sensor::temp")?; + /// let mut consumer = handle.consumer::("sensor::temp")?; /// let temp = consumer.get()?; /// # Ok(()) /// # } @@ -379,7 +379,7 @@ impl AimDbHandle { /// # struct RareEvent { id: u32 } /// # fn example(handle: &AimDbHandle) -> SyncResult<()> { /// // Rare events need smaller buffer - /// let consumer = handle.consumer_with_capacity::("events::rare", 10)?; + /// let mut consumer = handle.consumer_with_capacity::("events::rare", 10)?; /// let event = consumer.get()?; /// # Ok(()) /// # } @@ -396,35 +396,6 @@ impl AimDbHandle { let reader = self.db.subscribe::(&record_key).map_err(SyncError::Db)?; let waiter = Waiter::new(self.runtime_handle.clone()); Ok(crate::SyncConsumer::new(waiter, reader)) - - // match { - // Ok(reader) => { - // let _ = ready_tx.send(()); - // Self::forward_buffered(std_tx, reader).await; - // } - // Err(e) => { - // log_error!( - // "Failed to subscribe to record type {}: {}", - // std::any::type_name::(), - // e - // ); - // // Signal failure (will be ignored if receiver dropped) - // let _ = ready_tx.send(()); - // } - // } - - // self.runtime_handle.spawn(async move { - // // Subscribe to the database buffer for type T - // }); - - // // 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::()), - // })?; - - // Ok(crate::SyncConsumer::new(std_rx)) } /// Gracefully shut down the runtime thread. diff --git a/aimdb-sync/src/lib.rs b/aimdb-sync/src/lib.rs index f9931ea..f307d82 100644 --- a/aimdb-sync/src/lib.rs +++ b/aimdb-sync/src/lib.rs @@ -67,7 +67,7 @@ //! //! // Create producer and consumer //! let producer = handle.producer::("sensor.temp")?; -//! let consumer = handle.consumer::("sensor.temp")?; +//! let mut consumer = handle.consumer::("sensor.temp")?; //! //! // Producer: blocking operations //! producer.set(Temperature { celsius: 25.0 })?; @@ -84,28 +84,25 @@ //! //! ## 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, consumer: SyncConsumer) { +//! # fn demo(producer: SyncProducer, mut consumer: SyncConsumer) { //! //! // 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); +//! }; //! # } //! ``` //! @@ -173,7 +170,7 @@ #![cfg_attr(not(feature = "std"), doc = "```ignore")] //! # use aimdb_sync::SyncResult; //! # #[derive(Debug, Clone)] struct Temperature { celsius: f32 } -//! # fn demo(consumer: &aimdb_sync::SyncConsumer) -> SyncResult<()> { +//! # fn demo(consumer: &mut aimdb_sync::SyncConsumer) -> SyncResult<()> { //! // Always get the latest value, skipping queued intermediates //! let latest = consumer.get_latest()?; //! # Ok(()) diff --git a/aimdb-sync/src/producer.rs b/aimdb-sync/src/producer.rs index 22b03ad..7952f88 100644 --- a/aimdb-sync/src/producer.rs +++ b/aimdb-sync/src/producer.rs @@ -91,7 +91,7 @@ where /// ``` pub fn set(&self, value: T) -> SyncResult<()> { if let Some(db) = self.db.upgrade() { - db.produce(&self.key, value).map_err(|e| SyncError::Db(e)) + db.produce(&self.key, value).map_err(SyncError::Db) } else { Err(SyncError::RuntimeShutdown) } @@ -234,26 +234,14 @@ where } } -// // Safety: SyncProducer uses Arc internally and is safe to send/share -// unsafe impl Send for SyncProducer where T: Send + 'static + Debug + Clone {} -// unsafe impl Sync for SyncProducer where T: Send + 'static + Debug + Clone {} - -// TODO: remove or replace with static_assertions -const _: () = { - fn assert_send() {} - fn assert_sync() {} - - fn check() { - assert_send::>(); - assert_sync::>(); - } -}; - #[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 + // TODO: is it possible with static_assertions? + fn assert_send() {} + fn assert_sync() {} + #[allow(dead_code)] + fn check() { + assert_send::>(); + assert_sync::>(); } } From f334bc7614b37ff05e20a5c9c7b3f217c3e74252 Mon Sep 17 00:00:00 2001 From: fresheed Date: Sat, 1 Aug 2026 10:45:53 +0200 Subject: [PATCH 10/25] removed everything capacity-related --- aimdb-sync/src/handle.rs | 102 +-------------------------------------- aimdb-sync/src/lib.rs | 74 +--------------------------- 2 files changed, 3 insertions(+), 173 deletions(-) diff --git a/aimdb-sync/src/handle.rs b/aimdb-sync/src/handle.rs index d95305f..be43c54 100644 --- a/aimdb-sync/src/handle.rs +++ b/aimdb-sync/src/handle.rs @@ -9,18 +9,6 @@ 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 @@ -269,7 +257,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`. @@ -301,94 +289,6 @@ impl AimDbHandle { /// # } /// ``` pub fn consumer(&self, key: impl AsRef) -> SyncResult> - 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::("sensor::high_freq", 1000)?; - /// producer.set(HighFrequencySensor { value: 42.0 })?; - /// # Ok(()) - /// # } - /// ``` - pub fn producer_with_capacity( - &self, - key: impl AsRef, - capacity: usize, - ) -> SyncResult> - where - T: Send + 'static + Debug + Clone, - { - Ok(crate::SyncProducer::new(Arc::downgrade(&self.db), key)) - } - - /// 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 mut consumer = handle.consumer_with_capacity::("events::rare", 10)?; - /// let event = consumer.get()?; - /// # Ok(()) - /// # } - /// ``` - pub fn consumer_with_capacity( - &self, - key: impl AsRef, - capacity: usize, - ) -> SyncResult> where T: Send + Sync + 'static + Debug + Clone, { diff --git a/aimdb-sync/src/lib.rs b/aimdb-sync/src/lib.rs index f307d82..987593d 100644 --- a/aimdb-sync/src/lib.rs +++ b/aimdb-sync/src/lib.rs @@ -108,8 +108,7 @@ //! //! ## 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")] @@ -124,74 +123,6 @@ //! # } //! ``` //! -//! ## 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::("sensor.fast", 1000)?; -//! -//! // Rare events can use smaller buffer -//! let consumer = handle.consumer_with_capacity::("events.rare", 10)?; -//! -//! // SingleLatest-like behavior: use capacity=1 to minimize queueing -//! let consumer = handle.consumer_with_capacity::("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: &mut aimdb_sync::SyncConsumer) -> 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::("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 @@ -201,7 +132,6 @@ //! ## 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 @@ -263,7 +193,7 @@ 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; From cd1f9944afd8d979f66b6741458932e2dd804970 Mon Sep 17 00:00:00 2001 From: fresheed Date: Sat, 1 Aug 2026 10:49:40 +0200 Subject: [PATCH 11/25] `make check` passes --- aimdb-sync/tests/integration_test.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/aimdb-sync/tests/integration_test.rs b/aimdb-sync/tests/integration_test.rs index 7c41395..80b6d05 100644 --- a/aimdb-sync/tests/integration_test.rs +++ b/aimdb-sync/tests/integration_test.rs @@ -517,7 +517,7 @@ 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::("test.data", 10) + .producer::("test.data") .expect("Failed to create producer"); // Try to produce a value - this should fail because the key is not registered From d804b2df53650c57d2ebbe03fdcf2fd3d666d6ac Mon Sep 17 00:00:00 2001 From: fresheed Date: Sat, 1 Aug 2026 10:58:26 +0200 Subject: [PATCH 12/25] clarified shutdown test --- aimdb-sync/tests/integration_test.rs | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/aimdb-sync/tests/integration_test.rs b/aimdb-sync/tests/integration_test.rs index 80b6d05..5492d74 100644 --- a/aimdb-sync/tests/integration_test.rs +++ b/aimdb-sync/tests/integration_test.rs @@ -330,11 +330,8 @@ fn test_runtime_shutdown_error() { let result = producer.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.get(); + assert!(matches!(result, Err(SyncError::RuntimeShutdown))); } /// Test buffer semantics - SPMC Ring From 3aa1fdfdc26515c9b6229835e9e28fea7e81aa6c Mon Sep 17 00:00:00 2001 From: fresheed Date: Sat, 1 Aug 2026 13:06:55 +0200 Subject: [PATCH 13/25] made settable_integration run upon tests --- aimdb-sync/Cargo.toml | 6 ++++++ aimdb-sync/tests/settable_integration.rs | 6 +++--- 2 files changed, 9 insertions(+), 3 deletions(-) diff --git a/aimdb-sync/Cargo.toml b/aimdb-sync/Cargo.toml index 429416e..2757af9 100644 --- a/aimdb-sync/Cargo.toml +++ b/aimdb-sync/Cargo.toml @@ -41,6 +41,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"] diff --git a/aimdb-sync/tests/settable_integration.rs b/aimdb-sync/tests/settable_integration.rs index eb15052..0b496c3 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") .expect("failed to create producer"); - let consumer = handle + let mut consumer = handle .consumer::("temperature") .expect("failed to create consumer"); @@ -77,7 +77,7 @@ fn try_set_value_is_non_blocking_and_produces() { let producer = handle .producer::("temperature") .expect("failed to create producer"); - let consumer = handle + let mut consumer = handle .consumer::("temperature") .expect("failed to create consumer"); @@ -109,7 +109,7 @@ fn set_value_at_stamps_the_explicit_timestamp() { let producer = handle .producer::("temperature") .expect("failed to create producer"); - let consumer = handle + let mut consumer = handle .consumer::("temperature") .expect("failed to create consumer"); From 7335f77cef882363b50c384e92346b76cf330235 Mon Sep 17 00:00:00 2001 From: fresheed Date: Sat, 1 Aug 2026 13:09:18 +0200 Subject: [PATCH 14/25] updated cargo.lock (see previous commit) --- Cargo.lock | 1 + 1 file changed, 1 insertion(+) diff --git a/Cargo.lock b/Cargo.lock index 98aaedc..a07543c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -327,6 +327,7 @@ version = "0.6.0" dependencies = [ "aimdb-core", "aimdb-data-contracts", + "aimdb-sync", "aimdb-tokio-adapter", "serde", "serde_json", From a20f2b41cff6c47a853f35c6a05305beaf1e3c64 Mon Sep 17 00:00:00 2001 From: fresheed Date: Sat, 1 Aug 2026 13:09:58 +0200 Subject: [PATCH 15/25] added test for shutting down after producing --- aimdb-sync/tests/integration_test.rs | 41 ++++++++++++++++++++++++++++ 1 file changed, 41 insertions(+) diff --git a/aimdb-sync/tests/integration_test.rs b/aimdb-sync/tests/integration_test.rs index 5492d74..cbc2d60 100644 --- a/aimdb-sync/tests/integration_test.rs +++ b/aimdb-sync/tests/integration_test.rs @@ -334,6 +334,47 @@ fn test_runtime_shutdown_error() { assert!(matches!(result, Err(SyncError::RuntimeShutdown))); } +/// Test error handling - reading messages sent before the shutdown +#[test] +fn test_runtime_shutdown_after_produce_read_error() { + let adapter = Arc::new(TokioAdapter); + let mut builder = AimDbBuilder::new().runtime(adapter); + + builder.configure::("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::("test.data") + .expect("Failed to create producer"); + let mut consumer = handle + .consumer::("test.data") + .expect("Failed to create consumer"); + + let test_value = TestData { + id: 1, + value: "test".to_string(), + }; + + 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() { From 3d794f0d2d1144a40892649202dc21fce911ecc3 Mon Sep 17 00:00:00 2001 From: fresheed Date: Sat, 1 Aug 2026 13:28:08 +0200 Subject: [PATCH 16/25] refactored integration tests --- aimdb-sync/tests/integration_test.rs | 264 ++++++--------------------- 1 file changed, 53 insertions(+), 211 deletions(-) diff --git a/aimdb-sync/tests/integration_test.rs b/aimdb-sync/tests/integration_test.rs index cbc2d60..589a073 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,34 +19,51 @@ 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::("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, SyncConsumer) { + let handle = attach(cfg); let producer = handle .producer::("test.data") .expect("Failed to create producer"); - let mut consumer = handle + let consumer = handle .consumer::("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 @@ -64,17 +81,7 @@ 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::("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 mut consumer1 = handle @@ -150,34 +157,14 @@ 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::("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::("test.data") - .expect("Failed to create producer"); - let mut consumer = handle - .consumer::("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(test_value.clone()) .expect("Failed to produce with timeout"); @@ -197,34 +184,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::("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::("test.data") - .expect("Failed to create producer"); - let mut consumer = handle - .consumer::("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 +213,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::("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::("test.data") @@ -264,11 +221,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 +231,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::("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,33 +242,13 @@ 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); - - builder.configure::("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::("test.data") - .expect("Failed to create producer"); - let mut consumer = handle - .consumer::("test.data") - .expect("Failed to create consumer"); + 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(), - }; + let test_value = test_value(); let result = producer.set(test_value); assert!(matches!(result, Err(SyncError::RuntimeShutdown))); @@ -337,29 +260,9 @@ fn test_runtime_shutdown_error() { /// Test error handling - reading messages sent before the shutdown #[test] fn test_runtime_shutdown_after_produce_read_error() { - let adapter = Arc::new(TokioAdapter); - let mut builder = AimDbBuilder::new().runtime(adapter); - - builder.configure::("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::("test.data") - .expect("Failed to create producer"); - let mut consumer = handle - .consumer::("test.data") - .expect("Failed to create consumer"); + let (handle, producer, mut consumer) = setup(BufferCfg::SpmcRing { capacity: 10 }); - let test_value = TestData { - id: 1, - value: "test".to_string(), - }; + let test_value = test_value(); let result = producer.set(test_value.clone()); assert!(matches!(result, Ok(()))); @@ -378,17 +281,7 @@ fn test_runtime_shutdown_after_produce_read_error() { /// Test buffer semantics - SPMC Ring #[test] fn test_spmc_ring_semantics() { - let adapter = Arc::new(TokioAdapter); - let mut builder = AimDbBuilder::new().runtime(adapter); - - builder.configure::("test.data", |reg| { - reg.buffer(BufferCfg::SpmcRing { capacity: 5 }) - .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: 5 }); let producer = handle .producer::("test.data") @@ -402,11 +295,7 @@ fn test_spmc_ring_semantics() { // 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 @@ -429,32 +318,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::("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::("test.data") - .expect("Failed to create producer"); - - let mut consumer = handle - .consumer::("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"); + producer.set(initial_value).expect("Failed to produce"); thread::sleep(Duration::from_millis(100)); // Consume first value to establish the subscription @@ -463,11 +334,7 @@ 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"); + producer.set(data(i)).expect("Failed to produce value"); thread::sleep(Duration::from_millis(5)); } @@ -497,25 +364,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::("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::("test.data") - .expect("Failed to create producer"); - - let mut consumer = handle - .consumer::("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)); @@ -523,11 +372,7 @@ 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)); @@ -559,10 +404,7 @@ fn test_error_propagation() { .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()); From 1c43f920e8e5b89c2018ede4b9475202169b7674 Mon Sep 17 00:00:00 2001 From: fresheed Date: Sat, 1 Aug 2026 13:44:06 +0200 Subject: [PATCH 17/25] more detailed error handling for subscribing --- aimdb-sync/src/handle.rs | 23 ++++++++++++++++++++++- 1 file changed, 22 insertions(+), 1 deletion(-) diff --git a/aimdb-sync/src/handle.rs b/aimdb-sync/src/handle.rs index be43c54..e0f49e4 100644 --- a/aimdb-sync/src/handle.rs +++ b/aimdb-sync/src/handle.rs @@ -293,7 +293,10 @@ impl AimDbHandle { T: Send + Sync + 'static + Debug + Clone, { let record_key = key.as_ref().to_string(); - let reader = self.db.subscribe::(&record_key).map_err(SyncError::Db)?; + let reader = self + .db + .subscribe::(&record_key) + .map_err(lift_subscribe_error)?; let waiter = Waiter::new(self.runtime_handle.clone()); Ok(crate::SyncConsumer::new(waiter, reader)) } @@ -453,6 +456,24 @@ impl AimDbHandle { } } +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 { /// Attempts graceful shutdown if `detach()` was not called. /// From ccb981d29d536eadce6ea964d76e5b74c42727ad Mon Sep 17 00:00:00 2001 From: fresheed Date: Sat, 1 Aug 2026 13:58:25 +0200 Subject: [PATCH 18/25] fixed example (mut; called detach) --- examples/sync-api-demo/src/main.rs | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/examples/sync-api-demo/src/main.rs b/examples/sync-api-demo/src/main.rs index d29dbb6..4344291 100644 --- a/examples/sync-api-demo/src/main.rs +++ b/examples/sync-api-demo/src/main.rs @@ -69,8 +69,8 @@ fn main() -> Result<(), Box> { // Step 2: Create consumers before producing println!("2. Creating consumers for Temperature..."); - let consumer1 = handle.consumer::("sensor.temperature")?; - let consumer2 = handle.consumer::("sensor.temperature")?; + let mut consumer1 = handle.consumer::("sensor.temperature")?; + let mut consumer2 = handle.consumer::("sensor.temperature")?; // Alternative with custom capacity for high-frequency data: // let consumer1 = handle.consumer_with_capacity::("sensor.temperature", 1000)?; println!(" ✓ Two consumers created\n"); @@ -160,6 +160,7 @@ fn main() -> Result<(), Box> { println!("7. Shutting down..."); // Give async tasks time to process remaining values thread::sleep(Duration::from_millis(200)); + handle.detach().expect("Failed to detach"); // Detach the handle to gracefully shut down the runtime thread #[cfg(feature = "graceful-shutdown")] From a8db4a272a2a4a4037fa3a519c111c66f58a918b Mon Sep 17 00:00:00 2001 From: fresheed Date: Sat, 1 Aug 2026 15:47:12 +0200 Subject: [PATCH 19/25] cleaning up docs; removed `unsafe impl`s from handle --- aimdb-sync/src/consumer.rs | 26 ++++++++++++-------------- aimdb-sync/src/handle.rs | 16 ++++++++-------- aimdb-sync/src/producer.rs | 17 +---------------- 3 files changed, 21 insertions(+), 38 deletions(-) diff --git a/aimdb-sync/src/consumer.rs b/aimdb-sync/src/consumer.rs index bfba484..8666c59 100644 --- a/aimdb-sync/src/consumer.rs +++ b/aimdb-sync/src/consumer.rs @@ -12,16 +12,9 @@ use std::sync::Mutex; /// Synchronous consumer for records of type `T`. /// -/// TODO: **doc below is wrong, update it** -/// -/// 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` 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 reading from the same producer +/// receives data independently according to buffer semantics (SPMC, etc.). /// /// # Example /// @@ -52,7 +45,7 @@ use std::sync::Mutex; /// ``` pub struct SyncConsumer where - T: Send + 'static + Debug + Clone, + T: Send + Debug + Clone, { waiter: Waiter, reader: Reader, @@ -60,7 +53,7 @@ where impl SyncConsumer where - T: Send + 'static + Debug + Clone, + T: Send + Debug + Clone, { /// Create a new sync consumer (internal use only) pub(crate) fn new(waiter: Waiter, reader: Reader) -> Self { @@ -87,6 +80,7 @@ where /// # Errors /// /// - `SyncError::RuntimeShutdown` if the runtime thread has stopped + /// - `SyncError::Db` for other errors occured during read /// /// # Example /// @@ -124,6 +118,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 /// @@ -163,6 +158,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 /// @@ -209,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 /// @@ -302,7 +300,7 @@ mod tests { // TODO: is it possible with static_assertions? fn assert_send() {} #[allow(dead_code)] - fn check() { + fn check() { assert_send::>(); } } diff --git a/aimdb-sync/src/handle.rs b/aimdb-sync/src/handle.rs index e0f49e4..9384041 100644 --- a/aimdb-sync/src/handle.rs +++ b/aimdb-sync/src/handle.rs @@ -274,6 +274,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 /// @@ -492,15 +493,14 @@ 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 + // TODO: is it possible with static_assertions? + fn assert_send() {} + fn assert_sync() {} + #[allow(dead_code)] + fn check() { + assert_send::(); + assert_sync::(); } } diff --git a/aimdb-sync/src/producer.rs b/aimdb-sync/src/producer.rs index 7952f88..b12974b 100644 --- a/aimdb-sync/src/producer.rs +++ b/aimdb-sync/src/producer.rs @@ -36,6 +36,7 @@ use core::time::Duration; /// # Ok(()) /// # } /// ``` +#[derive(Clone)] pub struct SyncProducer where T: Send + 'static + Debug + Clone, @@ -218,22 +219,6 @@ fn unix_now_ms() -> u64 { .as_millis() as u64 } -impl Clone for SyncProducer -where - T: Send + 'static + Debug + Clone, -{ - /// Clone the producer to share across threads. - /// - /// Multiple clones can set values concurrently. - fn clone(&self) -> Self { - Self { - db: self.db.clone(), - key: self.key.clone(), - _phantom: PhantomData, - } - } -} - #[cfg(test)] mod tests { // TODO: is it possible with static_assertions? From 2a41f1da46182da838d96c8436103af41398913d Mon Sep 17 00:00:00 2001 From: fresheed Date: Sat, 1 Aug 2026 16:01:18 +0200 Subject: [PATCH 20/25] some cleanup in integratino tests --- aimdb-sync/tests/integration_test.rs | 25 ++----------------------- 1 file changed, 2 insertions(+), 23 deletions(-) diff --git a/aimdb-sync/tests/integration_test.rs b/aimdb-sync/tests/integration_test.rs index 589a073..759fdcc 100644 --- a/aimdb-sync/tests/integration_test.rs +++ b/aimdb-sync/tests/integration_test.rs @@ -66,13 +66,8 @@ fn test_basic_producer_consumer() { 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"); @@ -112,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::("test.data") @@ -169,9 +161,6 @@ fn test_timeout_operations() { .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)) @@ -298,9 +287,6 @@ fn test_spmc_ring_semantics() { 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"); @@ -326,7 +312,6 @@ fn test_single_latest_semantics() { value: "initial".to_string(), }; producer.set(initial_value).expect("Failed to produce"); - thread::sleep(Duration::from_millis(100)); // Consume first value to establish the subscription let first = consumer.get().expect("Failed to consume initial value"); @@ -335,12 +320,8 @@ fn test_single_latest_semantics() { // Now produce multiple values rapidly for i in 1..=5 { producer.set(data(i)).expect("Failed to produce value"); - thread::sleep(Duration::from_millis(5)); } - // 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"); @@ -375,8 +356,6 @@ fn test_get_latest_with_timeout() { 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)) From 4e65534d323ff63d7e59507830500c9bf2bfb6a5 Mon Sep 17 00:00:00 2001 From: fresheed Date: Sat, 1 Aug 2026 16:37:54 +0200 Subject: [PATCH 21/25] cleaned up docs; removed previously added .detach from example --- aimdb-sync/src/lib.rs | 17 ++++++++++------- examples/sync-api-demo/src/main.rs | 9 ++------- 2 files changed, 12 insertions(+), 14 deletions(-) diff --git a/aimdb-sync/src/lib.rs b/aimdb-sync/src/lib.rs index 987593d..ce097eb 100644 --- a/aimdb-sync/src/lib.rs +++ b/aimdb-sync/src/lib.rs @@ -13,16 +13,16 @@ //! //! ### 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()` //! @@ -150,9 +150,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")] @@ -171,7 +172,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)] diff --git a/examples/sync-api-demo/src/main.rs b/examples/sync-api-demo/src/main.rs index 4344291..78f1bb1 100644 --- a/examples/sync-api-demo/src/main.rs +++ b/examples/sync-api-demo/src/main.rs @@ -71,8 +71,6 @@ fn main() -> Result<(), Box> { println!("2. Creating consumers for Temperature..."); let mut consumer1 = handle.consumer::("sensor.temperature")?; let mut consumer2 = handle.consumer::("sensor.temperature")?; - // Alternative with custom capacity for high-frequency data: - // let consumer1 = handle.consumer_with_capacity::("sensor.temperature", 1000)?; println!(" ✓ Two consumers created\n"); // Step 3: Spawn consumer threads @@ -126,8 +124,6 @@ fn main() -> Result<(), Box> { // Step 4: Create a synchronous producer println!("4. Creating producer and producing values..."); let producer = handle.producer::("sensor.temperature")?; - // Alternative with custom capacity: - // let producer = handle.producer_with_capacity::("sensor.temperature", 500)?; println!(" ✓ Producer created\n"); // Step 5: Produce values @@ -160,7 +156,6 @@ fn main() -> Result<(), Box> { println!("7. Shutting down..."); // Give async tasks time to process remaining values thread::sleep(Duration::from_millis(200)); - handle.detach().expect("Failed to detach"); // Detach the handle to gracefully shut down the runtime thread #[cfg(feature = "graceful-shutdown")] @@ -172,8 +167,8 @@ fn main() -> Result<(), Box> { 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(()) From 6a4d1359c0bd7440c6bc7b2186ccaac3a9c7e999 Mon Sep 17 00:00:00 2001 From: fresheed Date: Sat, 1 Aug 2026 16:58:05 +0200 Subject: [PATCH 22/25] removed mentions of channels and removed methods; removed claim about overhead --- aimdb-sync/src/consumer.rs | 2 +- aimdb-sync/src/error.rs | 6 +++--- aimdb-sync/src/lib.rs | 6 ++---- aimdb-sync/src/producer.rs | 17 +++++++---------- 4 files changed, 13 insertions(+), 18 deletions(-) diff --git a/aimdb-sync/src/consumer.rs b/aimdb-sync/src/consumer.rs index 8666c59..5cac151 100644 --- a/aimdb-sync/src/consumer.rs +++ b/aimdb-sync/src/consumer.rs @@ -193,7 +193,7 @@ where /// 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. /// diff --git a/aimdb-sync/src/error.rs b/aimdb-sync/src/error.rs index 6583410..e647ea2 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/lib.rs b/aimdb-sync/src/lib.rs index ce097eb..b85b225 100644 --- a/aimdb-sync/src/lib.rs +++ b/aimdb-sync/src/lib.rs @@ -6,8 +6,8 @@ //! ## 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 //! @@ -127,11 +127,9 @@ //! //! - **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) //! - **Latency**: Excellent for <50ms target, not suitable for hard low-latency requirements //! //! ## Error Handling diff --git a/aimdb-sync/src/producer.rs b/aimdb-sync/src/producer.rs index b12974b..19c440f 100644 --- a/aimdb-sync/src/producer.rs +++ b/aimdb-sync/src/producer.rs @@ -10,7 +10,7 @@ use core::time::Duration; /// 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 /// @@ -31,7 +31,7 @@ use core::time::Duration; /// // 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(()) /// # } @@ -100,16 +100,13 @@ where /// 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()` 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 @@ -129,7 +126,7 @@ where /// let producer = handle.producer::("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(()) /// # } From 2d38d81211957e9515481c0b30907fb0b5437d04 Mon Sep 17 00:00:00 2001 From: fresheed Date: Sat, 1 Aug 2026 17:05:16 +0200 Subject: [PATCH 23/25] added test for failing non-blocking methods --- aimdb-sync/tests/integration_test.rs | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/aimdb-sync/tests/integration_test.rs b/aimdb-sync/tests/integration_test.rs index 759fdcc..2e528af 100644 --- a/aimdb-sync/tests/integration_test.rs +++ b/aimdb-sync/tests/integration_test.rs @@ -246,6 +246,24 @@ fn test_runtime_shutdown_error() { 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"); + + // Non-blocking operations should now fail with RuntimeShutdown too + let test_value = test_value(); + + let result = producer.try_set(test_value); + assert!(matches!(result, Err(SyncError::RuntimeShutdown))); + + let result = consumer.try_get(); + assert!(matches!(result, Err(SyncError::RuntimeShutdown))); +} + /// Test error handling - reading messages sent before the shutdown #[test] fn test_runtime_shutdown_after_produce_read_error() { From 94fdd92958f354d01130a6af3c3668e1970b6232 Mon Sep 17 00:00:00 2001 From: fresheed Date: Sun, 2 Aug 2026 16:55:27 +0200 Subject: [PATCH 24/25] pass after the initial review --- Cargo.lock | 1 - aimdb-sync/Cargo.toml | 6 ------ aimdb-sync/src/consumer.rs | 3 --- aimdb-sync/src/handle.rs | 6 ++---- aimdb-sync/src/lib.rs | 1 + aimdb-sync/src/producer.rs | 9 ++++----- aimdb-sync/src/waiter.rs | 6 ++---- 7 files changed, 9 insertions(+), 23 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index a07543c..98aaedc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -327,7 +327,6 @@ version = "0.6.0" dependencies = [ "aimdb-core", "aimdb-data-contracts", - "aimdb-sync", "aimdb-tokio-adapter", "serde", "serde_json", diff --git a/aimdb-sync/Cargo.toml b/aimdb-sync/Cargo.toml index 2757af9..429416e 100644 --- a/aimdb-sync/Cargo.toml +++ b/aimdb-sync/Cargo.toml @@ -41,12 +41,6 @@ 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"] diff --git a/aimdb-sync/src/consumer.rs b/aimdb-sync/src/consumer.rs index 5cac151..7ff27f5 100644 --- a/aimdb-sync/src/consumer.rs +++ b/aimdb-sync/src/consumer.rs @@ -4,11 +4,8 @@ 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`. /// diff --git a/aimdb-sync/src/handle.rs b/aimdb-sync/src/handle.rs index 9384041..7d4a9e9 100644 --- a/aimdb-sync/src/handle.rs +++ b/aimdb-sync/src/handle.rs @@ -2,8 +2,8 @@ use crate::waiter::Waiter; use crate::{SyncError, SyncResult}; -use aimdb_core::{log_error, log_warn, AimDb, AimDbBuilder, DbError, DbResult}; -use alloc::sync::{Arc, Weak}; +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}; @@ -167,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 { // Create shutdown channel let (shutdown_tx, mut shutdown_rx) = mpsc::channel::(1); diff --git a/aimdb-sync/src/lib.rs b/aimdb-sync/src/lib.rs index b85b225..d853183 100644 --- a/aimdb-sync/src/lib.rs +++ b/aimdb-sync/src/lib.rs @@ -189,6 +189,7 @@ mod error; mod handle; #[cfg(feature = "std")] mod producer; +#[cfg(feature = "std")] mod waiter; #[cfg(feature = "std")] diff --git a/aimdb-sync/src/producer.rs b/aimdb-sync/src/producer.rs index 19c440f..ea822e2 100644 --- a/aimdb-sync/src/producer.rs +++ b/aimdb-sync/src/producer.rs @@ -1,11 +1,10 @@ //! Synchronous producer for typed records. -use crate::{AimDbHandle, SyncError, SyncResult}; -use aimdb_core::{AimDb, DbResult, TryProduceError}; -use alloc::sync::{Arc, Weak}; +use crate::{SyncError, SyncResult}; +use aimdb_core::{AimDb, TryProduceError}; +use alloc::sync::Weak; use core::fmt::Debug; use core::marker::PhantomData; -use core::time::Duration; /// Synchronous producer for records of type `T`. /// @@ -222,7 +221,7 @@ mod tests { fn assert_send() {} fn assert_sync() {} #[allow(dead_code)] - fn check() { + fn check() { assert_send::>(); assert_sync::>(); } diff --git a/aimdb-sync/src/waiter.rs b/aimdb-sync/src/waiter.rs index c7c405f..f400326 100644 --- a/aimdb-sync/src/waiter.rs +++ b/aimdb-sync/src/waiter.rs @@ -1,13 +1,11 @@ -/// Runtime-specific implementations of Waiter define how to -/// running the given future on the current thread until completion +/// tokio-specific implementation of running the given future +/// on the current thread until completion use std::future::Future; -#[cfg(feature = "std")] pub struct Waiter { handle: tokio::runtime::Handle, } -#[cfg(feature = "std")] impl Waiter { pub fn new(handle: tokio::runtime::Handle) -> Self { Self { handle } From 4be9c2d6a9678945d09556761afa87fb0bc032c1 Mon Sep 17 00:00:00 2001 From: fresheed Date: Sun, 2 Aug 2026 17:58:13 +0200 Subject: [PATCH 25/25] removed mentions of static_assertions --- aimdb-sync/src/consumer.rs | 1 - aimdb-sync/src/handle.rs | 1 - aimdb-sync/src/producer.rs | 1 - 3 files changed, 3 deletions(-) diff --git a/aimdb-sync/src/consumer.rs b/aimdb-sync/src/consumer.rs index 7ff27f5..c8678e4 100644 --- a/aimdb-sync/src/consumer.rs +++ b/aimdb-sync/src/consumer.rs @@ -294,7 +294,6 @@ where #[cfg(test)] mod tests { - // TODO: is it possible with static_assertions? fn assert_send() {} #[allow(dead_code)] fn check() { diff --git a/aimdb-sync/src/handle.rs b/aimdb-sync/src/handle.rs index 7d4a9e9..98ed504 100644 --- a/aimdb-sync/src/handle.rs +++ b/aimdb-sync/src/handle.rs @@ -493,7 +493,6 @@ impl Drop for AimDbHandle { #[cfg(test)] mod tests { - // TODO: is it possible with static_assertions? fn assert_send() {} fn assert_sync() {} #[allow(dead_code)] diff --git a/aimdb-sync/src/producer.rs b/aimdb-sync/src/producer.rs index ea822e2..61265ae 100644 --- a/aimdb-sync/src/producer.rs +++ b/aimdb-sync/src/producer.rs @@ -217,7 +217,6 @@ fn unix_now_ms() -> u64 { #[cfg(test)] mod tests { - // TODO: is it possible with static_assertions? fn assert_send() {} fn assert_sync() {} #[allow(dead_code)]