commonware_storage/queue/
mod.rs

1//! A durable, at-least-once delivery queue backed by a [`variable::Journal`](crate::journal::contiguous::variable).
2//!
3//! [Queue] provides a persistent message queue with at-least-once delivery semantics.
4//! Items are durably stored in a journal and will survive crashes. [Queue::init] returns the
5//! queue, which writes items, and its [Reader], which delivers committed items. The reader must
6//! explicitly acknowledge each item after processing. On restart, all non-pruned items are
7//! re-delivered (acknowledged or not).
8//!
9//! # Ownership
10//!
11//! Methods that write to storage (`append`, `enqueue`, `commit`, `sync`) take the queue by value
12//! and return it on success. If one returns an error, or its future is dropped before it finishes,
13//! the queue is gone: state that was not yet durable is discarded, but everything already on disk
14//! stays recoverable. The reader then delivers every published item and returns `None`. Reader
15//! methods borrow the reader. A failed read does not invalidate it.
16//!
17//! # Concurrency
18//!
19//! The queue and its reader can run in separate tasks. The reader never waits on the queue, and
20//! [Reader::recv] integrates with `select!` for multiplexing with other futures.
21//!
22//! ```rust,ignore
23//! use commonware_storage::queue::Queue;
24//! use commonware_macros::select;
25//!
26//! let (queue, mut reader) = Queue::init(context, config).await?;
27//!
28//! // Writer task
29//! let (queue, position) = queue.enqueue(&item).await?;
30//!
31//! // Reader task
32//! loop {
33//!     select! {
34//!         result = reader.recv() => {
35//!             let Some((pos, item)) = result? else { break };
36//!             // Process item...
37//!             reader.ack(pos)?;
38//!         }
39//!         _ = shutdown => break,
40//!     }
41//! }
42//! ```
43//!
44//! # Example
45//!
46//! ```rust
47//! use commonware_codec::RangeCfg;
48//! use commonware_runtime::{Spawner, Runner, deterministic, buffer::paged::CacheRef};
49//! use commonware_storage::{queue::{Queue, Config}};
50//! use std::num::{NonZeroU16, NonZeroU64, NonZeroUsize};
51//!
52//! let executor = deterministic::Runner::default();
53//! executor.start(|context| async move {
54//!     // Create a page cache
55//!     let page_cache = CacheRef::new(
56//!         NonZeroU16::new(1024).unwrap(),
57//!         NonZeroUsize::new(10).unwrap(),
58//!     );
59//!
60//!     // Create a queue and its reader
61//!     let (mut queue, mut reader) = Queue::<_, Vec<u8>>::init(context, Config {
62//!         partition: "my-queue".into(),
63//!         items_per_section: NonZeroU64::new(1000).unwrap(),
64//!         compression: None,
65//!         codec_config: ((0..).into(), ()), // RangeCfg for Vec length, () for u8
66//!         page_cache,
67//!         write_buffer: NonZeroUsize::new(4096).unwrap(),
68//!         replay_buffer: NonZeroUsize::new(4096).unwrap(),
69//!     }).await.unwrap();
70//!
71//!     // Enqueue items
72//!     (queue, _) = queue.enqueue(&b"task1".to_vec()).await.unwrap();
73//!     (queue, _) = queue.enqueue(&b"task2".to_vec()).await.unwrap();
74//!
75//!     // Receive and process items (can be acknowledged out of order)
76//!     while let Some((position, item)) = reader.try_recv().await.unwrap() {
77//!         // Process the item...
78//!         println!("Processing item at position {}", position);
79//!
80//!         // Acknowledge after successful processing
81//!         reader.ack(position).unwrap();
82//!     }
83//!
84//!     // Persist the queue and prune acknowledged items
85//!     queue.sync().await.unwrap();
86//! });
87//! ```
88
89#[cfg(all(test, feature = "arbitrary"))]
90mod conformance;
91mod cursor;
92mod metrics;
93mod storage;
94
95pub use storage::{Config, Queue, Reader};
96use thiserror::Error;
97
98/// Errors that can occur when interacting with [Queue].
99#[derive(Debug, Error)]
100pub enum Error {
101    #[error("journal error: {0}")]
102    Journal(#[from] crate::journal::Error),
103    #[error("position out of range: {0} (queue size is {1})")]
104    PositionOutOfRange(u64, u64),
105}
commonware_storage/queue/mod.rs

commonware_storage/queue/
mod.rs
