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///
96/// # Returns
97///
98/// A handle the caller must release with [`pamoja_store_free`] or hand to a call
99/// that consumes it, or null if the directory cannot be opened.
100///
101/// # Safety
102///
103/// `dir` must be a valid null-terminated UTF-8 string for the duration of the
104/// call.
105#[no_mangle]
106pub unsafe extern "C" fn pamoja_store_file(dir: *const c_char) -> *mut PamojaStore {
107    let Some(dir) = read_str(dir, "dir") else {
108        return ptr::null_mut();
109    };
110    match FileStore::open(dir) {
111        Ok(store) => Box::into_raw(Box::new(PamojaStore {
112            kind: StoreKind::File(store),
113        })),
114        Err(error) => {
115            set_last_error(error.to_string());
116            ptr::null_mut()
117        }
118    }
119}
120
121/// Adds a record to the end of a buffer.
122///
123/// # Arguments
124///
125/// * `store` - the buffer.
126/// * `record` - the bytes to hold.
127/// * `record_len` - the length of `record`.
128///
129/// # Returns
130///
131/// [`PamojaStatus::Ok`] once the record is held.
132///
133/// # Safety
134///
135/// `store` must be a live handle, and `record` must point to at least
136/// `record_len` readable bytes or be null when that length is 0.
137#[no_mangle]
138pub unsafe extern "C" fn pamoja_store_append(
139    store: *mut PamojaStore,
140    record: *const u8,
141    record_len: usize,
142) -> PamojaStatus {
143    let Some(store) = store_handle(store) else {
144        return PamojaStatus::InvalidArgument;
145    };
146    let record = match read_bytes(record, record_len) {
147        Ok(record) => record,
148        Err(status) => return status,
149    };
150    match runtime().block_on(store.kind.append(&record)) {
151        Ok(()) => PamojaStatus::Ok,
152        Err(error) => fail(error),
153    }
154}
155
156/// Reads the oldest record without removing it.
157///
158/// # Arguments
159///
160/// * `store` - the buffer.
161/// * `out_record` - receives a buffer handle, or null when the store is empty.
162///
163/// # Returns
164///
165/// [`PamojaStatus::Ok`] on success. A null `out_record` with an `Ok` status
166/// means the buffer is empty.
167///
168/// # Safety
169///
170/// `store` must be a live handle and `out_record` must be writable.
171#[no_mangle]
172pub unsafe extern "C" fn pamoja_store_peek(
173    store: *mut PamojaStore,
174    out_record: *mut *mut PamojaBuffer,
175) -> PamojaStatus {
176    let Some(store) = store_handle(store) else {
177        return PamojaStatus::InvalidArgument;
178    };
179    if out_record.is_null() {
180        set_last_error("out_record must not be null".to_owned());
181        return PamojaStatus::InvalidArgument;
182    }
183    *out_record = ptr::null_mut();
184    match runtime().block_on(store.kind.peek()) {
185        Ok(Some(record)) => {
186            *out_record = PamojaBuffer::into_raw(record);
187            PamojaStatus::Ok
188        }
189        Ok(None) => PamojaStatus::Ok,
190        Err(error) => fail(error),
191    }
192}
193
194/// Removes and returns the oldest record.
195///
196/// # Arguments
197///
198/// * `store` - the buffer.
199/// * `out_record` - receives a buffer handle, or null when the store is empty.
200///
201/// # Returns
202///
203/// [`PamojaStatus::Ok`] on success. A null `out_record` with an `Ok` status
204/// means the buffer is empty.
205///
206/// # Safety
207///
208/// `store` must be a live handle and `out_record` must be writable.
209#[no_mangle]
210pub unsafe extern "C" fn pamoja_store_pop(
211    store: *mut PamojaStore,
212    out_record: *mut *mut PamojaBuffer,
213) -> PamojaStatus {
214    let Some(store) = store_handle(store) else {
215        return PamojaStatus::InvalidArgument;
216    };
217    if out_record.is_null() {
218        set_last_error("out_record must not be null".to_owned());
219        return PamojaStatus::InvalidArgument;
220    }
221    *out_record = ptr::null_mut();
222    match runtime().block_on(store.kind.pop()) {
223        Ok(Some(record)) => {
224            *out_record = PamojaBuffer::into_raw(record);
225            PamojaStatus::Ok
226        }
227        Ok(None) => PamojaStatus::Ok,
228        Err(error) => fail(error),
229    }
230}
231
232/// Reports how many records a buffer holds.
233///
234/// # Arguments
235///
236/// * `store` - the buffer.
237/// * `out_len` - receives the count.
238///
239/// # Returns
240///
241/// [`PamojaStatus::Ok`] on success.
242///
243/// # Safety
244///
245/// `store` must be a live handle and `out_len` must be writable.
246#[no_mangle]
247pub unsafe extern "C" fn pamoja_store_len(
248    store: *mut PamojaStore,
249    out_len: *mut usize,
250) -> PamojaStatus {
251    let Some(store) = store_handle(store) else {
252        return PamojaStatus::InvalidArgument;
253    };
254    if out_len.is_null() {
255        set_last_error("out_len must not be null".to_owned());
256        return PamojaStatus::InvalidArgument;
257    }
258    match runtime().block_on(store.kind.len()) {
259        Ok(len) => {
260            *out_len = len;
261            PamojaStatus::Ok
262        }
263        Err(error) => fail(error),
264    }
265}
266
267/// Sends every held record over a transport, oldest first.
268///
269/// A record is removed only once the transport has taken it, so a link that
270/// fails part-way leaves the rest of the queue intact for the next attempt.
271///
272/// # Arguments
273///
274/// * `store` - the buffer to drain.
275/// * `transport` - the transport to send over, borrowed rather than consumed.
276/// * `topic` - the topic to send to, as null-terminated UTF-8.
277/// * `out_sent` - receives how many records went out, or may be null.
278///
279/// # Returns
280///
281/// [`PamojaStatus::Ok`] if the whole buffer drained, or a transport error with
282/// `out_sent` holding how many got through before it stopped.
283///
284/// # Safety
285///
286/// `store` and `transport` must be live handles, `topic` a valid
287/// null-terminated UTF-8 string, and `out_sent` writable or null.
288#[no_mangle]
289pub unsafe extern "C" fn pamoja_store_drain_to(
290    store: *mut PamojaStore,
291    transport: *mut PamojaTransport,
292    topic: *const c_char,
293    out_sent: *mut usize,
294) -> PamojaStatus {
295    let Some(store) = store_handle(store) else {
296        return PamojaStatus::InvalidArgument;
297    };
298    if transport.is_null() {
299        set_last_error("transport must not be null".to_owned());
300        return PamojaStatus::InvalidArgument;
301    }
302    let Some(topic) = read_str(topic, "topic") else {
303        return PamojaStatus::InvalidArgument;
304    };
305
306    let transport = &mut (*transport).kind;
307    match runtime().block_on(drain_to(&mut store.kind, transport, topic)) {
308        Ok(sent) => {
309            if !out_sent.is_null() {
310                *out_sent = sent;
311            }
312            PamojaStatus::Ok
313        }
314        Err(error) => fail(error),
315    }
316}
317
318/// Releases a store handle.
319///
320/// Passing null is a no-op.
321///
322/// # Safety
323///
324/// `store` must be a handle from a call that produced one and that has not
325/// already been freed or consumed, or null. After this call it must not be used
326/// again.
327#[no_mangle]
328pub unsafe extern "C" fn pamoja_store_free(store: *mut PamojaStore) {
329    if !store.is_null() {
330        drop(Box::from_raw(store));
331    }
332}
333
334/// Borrows a store handle, rejecting a null pointer.
335///
336/// # Safety
337///
338/// `store` must be a live handle from a call that produced one, or null.
339unsafe fn store_handle<'a>(store: *mut PamojaStore) -> Option<&'a mut PamojaStore> {
340    if store.is_null() {
341        set_last_error("store must not be null".to_owned());
342        return None;
343    }
344    Some(&mut *store)
345}
346
347/// Takes ownership of a store handle, leaving the caller nothing to free.
348///
349/// # Safety
350///
351/// `store` must be a live handle that has not been freed or consumed, or null.
352/// After this call the caller must not use it again.
353pub(crate) unsafe fn take_store(store: *mut PamojaStore) -> Option<StoreKind> {
354    if store.is_null() {
355        set_last_error("store must not be null".to_owned());
356        return None;
357    }
358    Some(Box::from_raw(store).kind)
359}
360
361/// Records an error and maps it onto a status.
362fn fail(error: pamoja_core::Error) -> PamojaStatus {
363    let status = PamojaStatus::from_error(&error);
364    set_last_error(error.to_string());
365    status
366}
367
368#[cfg(test)]
369mod tests {
370    use super::*;
371    use crate::{pamoja_buffer_data, pamoja_buffer_free, pamoja_buffer_len};
372
373    /// Reads a buffer handle out and releases it.
374    unsafe fn take(buffer: *mut PamojaBuffer) -> Vec<u8> {
375        assert!(!buffer.is_null());
376        let bytes =
377            std::slice::from_raw_parts(pamoja_buffer_data(buffer), pamoja_buffer_len(buffer))
378                .to_vec();
379        pamoja_buffer_free(buffer);
380        bytes
381    }
382
383    #[test]
384    fn a_buffer_returns_records_oldest_first() {
385        unsafe {
386            let store = pamoja_store_memory(0);
387            assert_eq!(
388                pamoja_store_append(store, b"one".as_ptr(), 3),
389                PamojaStatus::Ok
390            );
391            assert_eq!(
392                pamoja_store_append(store, b"two".as_ptr(), 3),
393                PamojaStatus::Ok
394            );
395
396            let mut len = 0;
397            assert_eq!(pamoja_store_len(store, &mut len), PamojaStatus::Ok);
398            assert_eq!(len, 2);
399
400            let mut record = ptr::null_mut();
401            assert_eq!(pamoja_store_peek(store, &mut record), PamojaStatus::Ok);
402            assert_eq!(take(record), b"one", "peek leaves the record in place");
403
404            assert_eq!(pamoja_store_pop(store, &mut record), PamojaStatus::Ok);
405            assert_eq!(take(record), b"one");
406            assert_eq!(pamoja_store_pop(store, &mut record), PamojaStatus::Ok);
407            assert_eq!(take(record), b"two");
408
409            assert_eq!(pamoja_store_pop(store, &mut record), PamojaStatus::Ok);
410            assert!(record.is_null(), "an empty store yields nothing");
411
412            pamoja_store_free(store);
413        }
414    }
415
416    #[test]
417    fn a_full_buffer_refuses_rather_than_losing_a_record() {
418        unsafe {
419            let store = pamoja_store_memory(2);
420            assert_eq!(
421                pamoja_store_append(store, b"one".as_ptr(), 3),
422                PamojaStatus::Ok
423            );
424            assert_eq!(
425                pamoja_store_append(store, b"two".as_ptr(), 3),
426                PamojaStatus::Ok
427            );
428            assert_ne!(
429                pamoja_store_append(store, b"raw".as_ptr(), 3),
430                PamojaStatus::Ok,
431                "a full store tells the caller rather than dropping something"
432            );
433
434            let mut len = 0;
435            pamoja_store_len(store, &mut len);
436            assert_eq!(len, 2, "and what it already held is untouched");
437
438            let mut record = ptr::null_mut();
439            pamoja_store_pop(store, &mut record);
440            assert_eq!(take(record), b"one");
441
442            pamoja_store_free(store);
443        }
444    }
445
446    #[test]
447    fn a_null_handle_is_refused_rather_than_dereferenced() {
448        unsafe {
449            assert_eq!(
450                pamoja_store_append(ptr::null_mut(), b"x".as_ptr(), 1),
451                PamojaStatus::InvalidArgument
452            );
453            assert_eq!(
454                pamoja_store_len(ptr::null_mut(), ptr::null_mut()),
455                PamojaStatus::InvalidArgument
456            );
457            pamoja_store_free(ptr::null_mut());
458        }
459    }
460}
pamoja_ffi/sync.rs

pamoja_ffi/
sync.rs
