pamoja_core/
store.rs

1//! Durable local storage backing the offline-first synchronization layer.
2
3use alloc::string::String;
4use alloc::vec::Vec;
5use core::future::Future;
6
7use crate::error::Result;
8
9/// A durable, first-in first-out queue for store-and-forward buffering.
10///
11/// Records are appended while a device is offline and drained in order when a
12/// link becomes available, letting applications tolerate intermittent
13/// connectivity without losing data.
14///
15/// The returned futures are `Send`, for the same reason a [`Transport`]'s are: a
16/// transport ladder buffers into a store and is itself a transport, so a task on a
17/// multi-threaded runtime can drive it. An implementation written as `async fn`
18/// satisfies this as long as everything it holds across an await is `Send`.
19///
20/// [`Transport`]: crate::Transport
21pub trait Store {
22    /// Appends a record to the back of the queue.
23    ///
24    /// # Arguments
25    ///
26    /// * `record` - the raw bytes to persist.
27    ///
28    /// # Returns
29    ///
30    /// `Ok(())` once the record is durably stored.
31    ///
32    /// # Errors
33    ///
34    /// Returns [`Error::Io`](crate::Error::Io) if the record cannot be written to
35    /// durable storage.
36    fn append(&mut self, record: &[u8]) -> impl Future<Output = Result<()>> + Send;
37
38    /// Appends text to the back of the queue: a reading or a line written out.
39    ///
40    /// This is [`append`](Self::append) with the text's UTF-8 bytes, so a queue of
41    /// readings needs no encoding step; [`peek_text`](Self::peek_text) and
42    /// [`pop_text`](Self::pop_text) read them back.
43    ///
44    /// # Arguments
45    ///
46    /// * `text` - the text to persist.
47    ///
48    /// # Returns
49    ///
50    /// `Ok(())` once the record is durably stored.
51    ///
52    /// # Errors
53    ///
54    /// Whatever [`append`](Self::append) returns.
55    fn append_text(&mut self, text: &str) -> impl Future<Output = Result<()>> + Send {
56        self.append(text.as_bytes())
57    }
58
59    /// Returns the oldest record without removing it.
60    ///
61    /// This lets a forwarder send a record before committing to its removal, so a
62    /// failed send can leave the record buffered in order rather than dropping it.
63    ///
64    /// # Returns
65    ///
66    /// `Some(record)` containing the oldest buffered bytes, or `None` if the queue
67    /// is empty.
68    ///
69    /// # Errors
70    ///
71    /// Returns [`Error::Io`](crate::Error::Io) if the queue cannot be read.
72    fn peek(&self) -> impl Future<Output = Result<Option<Vec<u8>>>> + Send;
73
74    /// Returns the oldest record as text, without removing it.
75    ///
76    /// It borrows the store across an await, so it asks for `Sync`, which a store
77    /// whose own futures are `Send` already is.
78    ///
79    /// # Returns
80    ///
81    /// `Some(text)` for the oldest record, or `None` if the queue is empty.
82    ///
83    /// # Errors
84    ///
85    /// Whatever [`peek`](Self::peek) returns, or
86    /// [`Error::Codec`](crate::Error::Codec) if the record is not UTF-8 text.
87    fn peek_text(&self) -> impl Future<Output = Result<Option<String>>> + Send
88    where
89        Self: Sync,
90    {
91        async { as_text(self.peek().await?) }
92    }
93
94    /// Removes and returns the oldest record in the queue.
95    ///
96    /// # Returns
97    ///
98    /// `Some(record)` containing the oldest buffered bytes, or `None` if the queue
99    /// is empty.
100    ///
101    /// # Errors
102    ///
103    /// Returns [`Error::Io`](crate::Error::Io) if the queue cannot be read.
104    fn pop(&mut self) -> impl Future<Output = Result<Option<Vec<u8>>>> + Send;
105
106    /// Removes and returns the oldest record as text.
107    ///
108    /// # Returns
109    ///
110    /// `Some(text)` for the oldest record, or `None` if the queue is empty.
111    ///
112    /// # Errors
113    ///
114    /// Whatever [`pop`](Self::pop) returns, or
115    /// [`Error::Codec`](crate::Error::Codec) if the record is not UTF-8 text.
116    fn pop_text(&mut self) -> impl Future<Output = Result<Option<String>>> + Send
117    where
118        Self: Send,
119    {
120        async { as_text(self.pop().await?) }
121    }
122
123    /// Returns the number of records currently buffered.
124    ///
125    /// # Returns
126    ///
127    /// The count of records waiting to be drained.
128    ///
129    /// # Errors
130    ///
131    /// Returns [`Error::Io`](crate::Error::Io) if the queue length cannot be
132    /// determined.
133    fn len(&self) -> impl Future<Output = Result<usize>> + Send;
134
135    /// Returns whether the queue currently holds no records.
136    ///
137    /// The default implementation reports whether [`len`](Self::len) is zero. It
138    /// borrows the store across an await, so it asks for `Sync`, which a store whose
139    /// own futures are `Send` already is.
140    ///
141    /// # Returns
142    ///
143    /// `true` if the queue is empty, `false` otherwise.
144    ///
145    /// # Errors
146    ///
147    /// Returns [`Error::Io`](crate::Error::Io) if the queue length cannot be
148    /// determined.
149    fn is_empty(&self) -> impl Future<Output = Result<bool>> + Send
150    where
151        Self: Sync,
152    {
153        async { Ok(self.len().await? == 0) }
154    }
155}
156
157// A buffered record read as text, for the `_text` helpers above.
158fn as_text(record: Option<Vec<u8>>) -> Result<Option<String>> {
159    match record {
160        Some(bytes) => String::from_utf8(bytes)
161            .map(Some)
162            .map_err(|_| crate::Error::Codec("the record is not UTF-8 text".into())),
163        None => Ok(None),
164    }
165}
pamoja_core/store.rs

pamoja_core/
store.rs
