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. The reader must
5//! explicitly acknowledge each item after processing. On restart, all non-pruned
6//! items are re-delivered (acknowledged or not).
7//!
8//! # Ownership
9//!
10//! Methods that write to storage (`append`, `enqueue`, `commit`, `sync`) take the queue by
11//! value and return it on success. If one returns an error, or its future is dropped before
12//! it finishes, the queue is gone: state that was not yet durable is discarded, but
13//! everything already on disk stays recoverable. Reads and
14//! in-memory bookkeeping (`dequeue`, `ack`, `ack_up_to`, `reset`) borrow the queue; a failed
15//! `dequeue` read does not invalidate it.
16//!
17//! # Concurrent Access
18//!
19//! For concurrent access from separate writer and reader tasks, use the [shared] module.
20//! Writers can be cloned for multiple producer tasks.
21//!
22//! ```rust,ignore
23//! use commonware_storage::queue::shared;
24//! use commonware_macros::select;
25//!
26//! let (writer, mut reader) = shared::init(context, config).await?;
27//!
28//! // Writer task (can clone for multiple producers)
29//! writer.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).await?;
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::from_pooler(
56//!         &context,
57//!         NonZeroU16::new(1024).unwrap(),
58//!         NonZeroUsize::new(10).unwrap(),
59//!     );
60//!
61//!     // Create a queue
62//!     let mut queue = Queue::<_, Vec<u8>>::init(context, Config {
63//!         partition: "my-queue".into(),
64//!         items_per_section: NonZeroU64::new(1000).unwrap(),
65//!         compression: None,
66//!         codec_config: ((0..).into(), ()), // RangeCfg for Vec length, () for u8
67//!         page_cache,
68//!         write_buffer: NonZeroUsize::new(4096).unwrap(),
69//!         replay_buffer: NonZeroUsize::new(4096).unwrap(),
70//!     }).await.unwrap();
71//!
72//!     // Enqueue items
73//!     (queue, _) = queue.enqueue(b"task1".to_vec()).await.unwrap();
74//!     (queue, _) = queue.enqueue(b"task2".to_vec()).await.unwrap();
75//!
76//!     // Dequeue and process items (can be done out of order)
77//!     while let Some((position, item)) = queue.dequeue().await.unwrap() {
78//!         // Process the item...
79//!         println!("Processing item at position {}", position);
80//!
81//!         // Acknowledge after successful processing
82//!         queue.ack(position).unwrap();
83//!     }
84//! });
85//! ```
86
87#[cfg(all(test, feature = "arbitrary"))]
88mod conformance;
89mod metrics;
90pub mod shared;
91mod storage;
92
93pub use shared::{Reader, Writer};
94pub use storage::{Config, Queue};
95use thiserror::Error;
96
97/// Errors that can occur when interacting with [Queue].
98#[derive(Debug, Error)]
99pub enum Error {
100    #[error("journal error: {0}")]
101    Journal(#[from] crate::journal::Error),
102    #[error("position out of range: {0} (queue size is {1})")]
103    PositionOutOfRange(u64, u64),
104    #[error(
105        "queue is no longer usable: a previous operation failed or was interrupted; reopen it to recover"
106    )]
107    Unavailable,
108}
commonware_storage/queue/mod.rs

commonware_storage/queue/
mod.rs
