pamoja_ffi/
sync.rs

1//! The C ABI for store-and-forward buffers.
2//!
3//! These functions wrap [`pamoja_sync`] for callers that reach the SDK through
4//! the flat C boundary: the queue a node writes into while it has nowhere to
5//! send, and the drain that empties it once a link comes back.
6//!
7//! Two stores cross, behind one handle. An in-memory queue is the right choice
8//! for a test or a process that will not outlive its buffer; a file-backed one
9//! survives a reboot, which is what a node in a place with no reliable power
10//! actually needs. The kind is chosen when the store is created and nothing
11//! afterwards has to care which it is.
12
13use std::ffi::c_char;
14use std::ptr;
15
16use pamoja_core::{Result, Store};
17use pamoja_sync::{drain_to, FileStore, MemoryStore};
18
19use crate::transport::PamojaTransport;
20use crate::{read_bytes, read_str, runtime, set_last_error, PamojaBuffer, PamojaStatus};
21
22/// One buffer, whichever kind it was created as.
23pub(crate) enum StoreKind {
24    /// A queue held in memory, lost when the process ends.
25    Memory(MemoryStore),
26    /// A queue on disk, which survives a restart.
27    File(FileStore),
28}
29
30impl Store for StoreKind {
31    async fn append(&mut self, record: &[u8]) -> Result<()> {
32        match self {
33            StoreKind::Memory(store) => store.append(record).await,
34            StoreKind::File(store) => store.append(record).await,
35        }
36    }
37
38    async fn peek(&self) -> Result<Option<Vec<u8>>> {
39        match self {
40            StoreKind::Memory(store) => store.peek().await,
41            StoreKind::File(store) => store.peek().await,
42        }
43    }
44
45    async fn pop(&mut self) -> Result<Option<Vec<u8>>> {
46        match self {
47            StoreKind::Memory(store) => store.pop().await,
48            StoreKind::File(store) => store.pop().await,
49        }
50    }
51
52    async fn len(&self) -> Result<usize> {
53        match self {
54            StoreKind::Memory(store) => store.len().await,
55            StoreKind::File(store) => store.len().await,
56        }
57    }
58}
59
60/// An opaque handle to a store-and-forward buffer.
61pub struct PamojaStore {
62    pub(crate) kind: StoreKind,
63}
64
65/// Creates a buffer held in memory.
66///
67/// # Arguments
68///
69/// * `capacity` - the most records to hold, or 0 for no bound. A full store
70///   refuses the next append rather than dropping anything, so a record is
71///   never lost without the caller being told.
72///
73/// # Returns
74///
75/// A handle the caller must release with [`pamoja_store_free`] or hand to a call
76/// that consumes it.
77#[no_mangle]
78pub extern "C" fn pamoja_store_memory(capacity: usize) -> *mut PamojaStore {
79    let store = if capacity == 0 {
80        MemoryStore::new()
81    } else {
82        MemoryStore::with_capacity(capacity)
83    };
84    Box::into_raw(Box::new(PamojaStore {
85        kind: StoreKind::Memory(store),
86    }))
87}
88
89/// Opens a buffer backed by a directory, so it survives a restart.
90///
91/// # Arguments
92///
93/// * `dir` - the directory to hold records in, as null-terminated UTF-8. It is
94///   created if it does not exist.
95/// * `capacity` - the most records to hold, or 0 for no bound. A full store
96///   refuses the next append, which keeps a long outage from filling the disk.
97///
98/// # Returns
99///
100/// A handle the caller must release with [`pamoja_store_free`] or hand to a call
101/// that consumes it, or null if the directory cannot be opened.
102///
103/// # Safety
104///
105/// `dir` must be a valid null-terminated UTF-8 string for the duration of the
106/// call.
107#[no_mangle]
108pub unsafe extern "C" fn pamoja_store_file(
109    dir: *const c_char,
110    capacity: usize,
111) -> *mut PamojaStore {
112    let Some(dir) = read_str(dir, "dir") else {
113        return ptr::null_mut();
114    };
115    let opened = if capacity == 0 {
116        FileStore::open(dir)
117    } else {
118        FileStore::open_with_capacity(dir, capacity)
119    };
120    match opened {
121        Ok(store) => Box::into_raw(Box::new(PamojaStore {
122            kind: StoreKind::File(store),
123        })),
124        Err(error) => {
125            set_last_error(error.to_string());
126            ptr::null_mut()
127        }
128    }
129}
130
131/// Adds a record to the end of a buffer.
132///
133/// # Arguments
134///
135/// * `store` - the buffer.
136/// * `record` - the bytes to hold.
137/// * `record_len` - the length of `record`.
138///
139/// # Returns
140///
141/// [`PamojaStatus::Ok`] once the record is held.
142///
143/// # Safety
144///
145/// `store` must be a live handle, and `record` must point to at least
146/// `record_len` readable bytes or be null when that length is 0.
147#[no_mangle]
148pub unsafe extern "C" fn pamoja_store_append(
149    store: *mut PamojaStore,
150    record: *const u8,
151    record_len: usize,
152) -> PamojaStatus {
153    let Some(store) = store_handle(store) else {
154        return PamojaStatus::InvalidArgument;
155    };
156    let record = match read_bytes(record, record_len) {
157        Ok(record) => record,
158        Err(status) => return status,
159    };
160    match runtime().block_on(store.kind.append(&record)) {
161        Ok(()) => PamojaStatus::Ok,
162        Err(error) => fail(error),
163    }
164}
165
166/// Reads the oldest record without removing it.
167///
168/// # Arguments
169///
170/// * `store` - the buffer.
171/// * `out_record` - receives a buffer handle, or null when the store is empty.
172///
173/// # Returns
174///
175/// [`PamojaStatus::Ok`] on success. A null `out_record` with an `Ok` status
176/// means the buffer is empty.
177///
178/// # Safety
179///
180/// `store` must be a live handle and `out_record` must be writable.
181#[no_mangle]
182pub unsafe extern "C" fn pamoja_store_peek(
183    store: *mut PamojaStore,
184    out_record: *mut *mut PamojaBuffer,
185) -> PamojaStatus {
186    let Some(store) = store_handle(store) else {
187        return PamojaStatus::InvalidArgument;
188    };
189    if out_record.is_null() {
190        set_last_error("out_record must not be null".to_owned());
191        return PamojaStatus::InvalidArgument;
192    }
193    *out_record = ptr::null_mut();
194    match runtime().block_on(store.kind.peek()) {
195        Ok(Some(record)) => {
196            *out_record = PamojaBuffer::into_raw(record);
197            PamojaStatus::Ok
198        }
199        Ok(None) => PamojaStatus::Ok,
200        Err(error) => fail(error),
201    }
202}
203
204/// Removes and returns the oldest record.
205///
206/// # Arguments
207///
208/// * `store` - the buffer.
209/// * `out_record` - receives a buffer handle, or null when the store is empty.
210///
211/// # Returns
212///
213/// [`PamojaStatus::Ok`] on success. A null `out_record` with an `Ok` status
214/// means the buffer is empty.
215///
216/// # Safety
217///
218/// `store` must be a live handle and `out_record` must be writable.
219#[no_mangle]
220pub unsafe extern "C" fn pamoja_store_pop(
221    store: *mut PamojaStore,
222    out_record: *mut *mut PamojaBuffer,
223) -> PamojaStatus {
224    let Some(store) = store_handle(store) else {
225        return PamojaStatus::InvalidArgument;
226    };
227    if out_record.is_null() {
228        set_last_error("out_record must not be null".to_owned());
229        return PamojaStatus::InvalidArgument;
230    }
231    *out_record = ptr::null_mut();
232    match runtime().block_on(store.kind.pop()) {
233        Ok(Some(record)) => {
234            *out_record = PamojaBuffer::into_raw(record);
235            PamojaStatus::Ok
236        }
237        Ok(None) => PamojaStatus::Ok,
238        Err(error) => fail(error),
239    }
240}
241
242/// Reports how many records a buffer holds.
243///
244/// # Arguments
245///
246/// * `store` - the buffer.
247/// * `out_len` - receives the count.
248///
249/// # Returns
250///
251/// [`PamojaStatus::Ok`] on success.
252///
253/// # Safety
254///
255/// `store` must be a live handle and `out_len` must be writable.
256#[no_mangle]
257pub unsafe extern "C" fn pamoja_store_len(
258    store: *mut PamojaStore,
259    out_len: *mut usize,
260) -> PamojaStatus {
261    let Some(store) = store_handle(store) else {
262        return PamojaStatus::InvalidArgument;
263    };
264    if out_len.is_null() {
265        set_last_error("out_len must not be null".to_owned());
266        return PamojaStatus::InvalidArgument;
267    }
268    match runtime().block_on(store.kind.len()) {
269        Ok(len) => {
270            *out_len = len;
271            PamojaStatus::Ok
272        }
273        Err(error) => fail(error),
274    }
275}
276
277/// Sends every held record over a transport, oldest first.
278///
279/// A record is removed only once the transport has taken it, so a link that
280/// fails part-way leaves the rest of the queue intact for the next attempt.
281///
282/// # Arguments
283///
284/// * `store` - the buffer to drain.
285/// * `transport` - the transport to send over, borrowed rather than consumed.
286/// * `topic` - the topic to send to, as null-terminated UTF-8.
287/// * `out_sent` - receives how many records went out, or may be null.
288///
289/// # Returns
290///
291/// [`PamojaStatus::Ok`] if the whole buffer drained, or a transport error with
292/// `out_sent` holding how many got through before it stopped.
293///
294/// # Safety
295///
296/// `store` and `transport` must be live handles, `topic` a valid
297/// null-terminated UTF-8 string, and `out_sent` writable or null.
298#[no_mangle]
299pub unsafe extern "C" fn pamoja_store_drain_to(
300    store: *mut PamojaStore,
301    transport: *mut PamojaTransport,
302    topic: *const c_char,
303    out_sent: *mut usize,
304) -> PamojaStatus {
305    let Some(store) = store_handle(store) else {
306        return PamojaStatus::InvalidArgument;
307    };
308    if transport.is_null() {
309        set_last_error("transport must not be null".to_owned());
310        return PamojaStatus::InvalidArgument;
311    }
312    let Some(topic) = read_str(topic, "topic") else {
313        return PamojaStatus::InvalidArgument;
314    };
315
316    let transport = &mut (*transport).kind;
317    match runtime().block_on(drain_to(&mut store.kind, transport, topic)) {
318        Ok(sent) => {
319            if !out_sent.is_null() {
320                *out_sent = sent;
321            }
322            PamojaStatus::Ok
323        }
324        Err(error) => fail(error),
325    }
326}
327
328/// Releases a store handle.
329///
330/// Passing null is a no-op.
331///
332/// # Safety
333///
334/// `store` must be a handle from a call that produced one and that has not
335/// already been freed or consumed, or null. After this call it must not be used
336/// again.
337#[no_mangle]
338pub unsafe extern "C" fn pamoja_store_free(store: *mut PamojaStore) {
339    if !store.is_null() {
340        drop(Box::from_raw(store));
341    }
342}
343
344/// Borrows a store handle, rejecting a null pointer.
345///
346/// # Safety
347///
348/// `store` must be a live handle from a call that produced one, or null.
349unsafe fn store_handle<'a>(store: *mut PamojaStore) -> Option<&'a mut PamojaStore> {
350    if store.is_null() {
351        set_last_error("store must not be null".to_owned());
352        return None;
353    }
354    Some(&mut *store)
355}
356
357/// Takes ownership of a store handle, leaving the caller nothing to free.
358///
359/// # Safety
360///
361/// `store` must be a live handle that has not been freed or consumed, or null.
362/// After this call the caller must not use it again.
363pub(crate) unsafe fn take_store(store: *mut PamojaStore) -> Option<StoreKind> {
364    if store.is_null() {
365        set_last_error("store must not be null".to_owned());
366        return None;
367    }
368    Some(Box::from_raw(store).kind)
369}
370
371/// Records an error and maps it onto a status.
372fn fail(error: pamoja_core::Error) -> PamojaStatus {
373    let status = PamojaStatus::from_error(&error);
374    set_last_error(error.to_string());
375    status
376}
377
378#[cfg(test)]
379mod tests {
380    use super::*;
381    use crate::{pamoja_buffer_data, pamoja_buffer_free, pamoja_buffer_len};
382
383    /// Reads a buffer handle out and releases it.
384    unsafe fn take(buffer: *mut PamojaBuffer) -> Vec<u8> {
385        assert!(!buffer.is_null());
386        let bytes =
387            std::slice::from_raw_parts(pamoja_buffer_data(buffer), pamoja_buffer_len(buffer))
388                .to_vec();
389        pamoja_buffer_free(buffer);
390        bytes
391    }
392
393    #[test]
394    fn a_buffer_returns_records_oldest_first() {
395        unsafe {
396            let store = pamoja_store_memory(0);
397            assert_eq!(
398                pamoja_store_append(store, b"one".as_ptr(), 3),
399                PamojaStatus::Ok
400            );
401            assert_eq!(
402                pamoja_store_append(store, b"two".as_ptr(), 3),
403                PamojaStatus::Ok
404            );
405
406            let mut len = 0;
407            assert_eq!(pamoja_store_len(store, &mut len), PamojaStatus::Ok);
408            assert_eq!(len, 2);
409
410            let mut record = ptr::null_mut();
411            assert_eq!(pamoja_store_peek(store, &mut record), PamojaStatus::Ok);
412            assert_eq!(take(record), b"one", "peek leaves the record in place");
413
414            assert_eq!(pamoja_store_pop(store, &mut record), PamojaStatus::Ok);
415            assert_eq!(take(record), b"one");
416            assert_eq!(pamoja_store_pop(store, &mut record), PamojaStatus::Ok);
417            assert_eq!(take(record), b"two");
418
419            assert_eq!(pamoja_store_pop(store, &mut record), PamojaStatus::Ok);
420            assert!(record.is_null(), "an empty store yields nothing");
421
422            pamoja_store_free(store);
423        }
424    }
425
426    #[test]
427    fn a_full_buffer_refuses_rather_than_losing_a_record() {
428        unsafe {
429            let store = pamoja_store_memory(2);
430            assert_eq!(
431                pamoja_store_append(store, b"one".as_ptr(), 3),
432                PamojaStatus::Ok
433            );
434            assert_eq!(
435                pamoja_store_append(store, b"two".as_ptr(), 3),
436                PamojaStatus::Ok
437            );
438            assert_ne!(
439                pamoja_store_append(store, b"raw".as_ptr(), 3),
440                PamojaStatus::Ok,
441                "a full store tells the caller rather than dropping something"
442            );
443
444            let mut len = 0;
445            pamoja_store_len(store, &mut len);
446            assert_eq!(len, 2, "and what it already held is untouched");
447
448            let mut record = ptr::null_mut();
449            pamoja_store_pop(store, &mut record);
450            assert_eq!(take(record), b"one");
451
452            pamoja_store_free(store);
453        }
454    }
455
456    #[test]
457    fn a_null_handle_is_refused_rather_than_dereferenced() {
458        unsafe {
459            assert_eq!(
460                pamoja_store_append(ptr::null_mut(), b"x".as_ptr(), 1),
461                PamojaStatus::InvalidArgument
462            );
463            assert_eq!(
464                pamoja_store_len(ptr::null_mut(), ptr::null_mut()),
465                PamojaStatus::InvalidArgument
466            );
467            pamoja_store_free(ptr::null_mut());
468        }
469    }
470}
pamoja_ffi/sync.rs

pamoja_ffi/
sync.rs
