appcore_sync/sync/
outbox.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: outbox.rs
4//    ##   ## ##   ##    P: AppCore-Runtime
5//         ## ##
6//                       C: 2026/07/21 10:48:21 by dnettoRaw
7//    ##   ## ##   ##    U: 2026/08/26 00:00:00 by dnettoRaw
8//      ###########      S: 2.0.0
9// =============================================================================
10
11//! Durable bounded outbox contracts and process-local implementation.
12
13use crate::sync::error::{SyncError, SyncResult};
14use crate::sync::types::SyncMessage;
15use parking_lot::Mutex;
16use std::collections::VecDeque;
17
18pub use crate::sync::outbox_journal::{FileSyncOutbox, SYNC_OUTBOX_FORMAT_V2};
19
20/// Ordered bounded queue that retains replication batches until acknowledgement.
21pub trait SyncOutbox: Send + Sync {
22    /// Enqueues a batch if the current length is below `max_len`.
23    fn try_enqueue(&self, message: SyncMessage, max_len: usize) -> SyncResult<bool>;
24    /// Returns the oldest pending batch.
25    fn front(&self) -> SyncResult<Option<SyncMessage>>;
26    /// Removes the oldest batch only when its identifier matches `batch_id`.
27    fn acknowledge_front(&self, batch_id: &str) -> SyncResult<()>;
28    /// Returns all pending batches in delivery order.
29    fn messages(&self) -> SyncResult<Vec<SyncMessage>>;
30    /// Returns the number of pending batches.
31    fn len(&self) -> SyncResult<usize>;
32    /// Reports whether no batches are pending.
33    fn is_empty(&self) -> SyncResult<bool> {
34        Ok(self.len()? == 0)
35    }
36}
37
38#[derive(Debug, Default)]
39/// Process-local synchronization outbox.
40pub struct InMemorySyncOutbox {
41    messages: Mutex<VecDeque<SyncMessage>>,
42}
43
44impl InMemorySyncOutbox {
45    /// Creates an empty in-memory outbox.
46    pub fn new() -> Self {
47        Self::default()
48    }
49}
50
51impl SyncOutbox for InMemorySyncOutbox {
52    fn try_enqueue(&self, message: SyncMessage, max_len: usize) -> SyncResult<bool> {
53        let mut messages = self.messages.lock();
54        if messages.len() >= max_len {
55            return Ok(false);
56        }
57        messages.push_back(message);
58        Ok(true)
59    }
60
61    fn front(&self) -> SyncResult<Option<SyncMessage>> {
62        Ok(self.messages.lock().front().cloned())
63    }
64
65    fn acknowledge_front(&self, batch_id: &str) -> SyncResult<()> {
66        let mut messages = self.messages.lock();
67        if messages.front().map(|message| message.batch_id.as_str()) != Some(batch_id) {
68            return Err(SyncError::InvalidSyncMessage(
69                "outbox acknowledgement mismatch",
70            ));
71        }
72        messages.pop_front();
73        Ok(())
74    }
75
76    fn messages(&self) -> SyncResult<Vec<SyncMessage>> {
77        Ok(self.messages.lock().iter().cloned().collect())
78    }
79
80    fn len(&self) -> SyncResult<usize> {
81        Ok(self.messages.lock().len())
82    }
83}
appcore_sync/sync/outbox.rs

appcore_sync/sync/
outbox.rs
