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/02 13:24:05 by dnettoRaw
8//      ###########      S: 1.0.1-rc.8
9// =============================================================================
10
11//! Durable bounded outbox contracts for follower replication pushes.
12
13use crate::sync::codec::{bytes_to_hex, hex_to_bytes};
14use crate::sync::error::{SyncError, SyncResult};
15use crate::sync::persistence::{
16    acquire_persistence_lock, atomic_write, read_bounded_text, split_format,
17};
18use crate::sync::types::SyncMessage;
19use parking_lot::Mutex;
20use std::collections::VecDeque;
21use std::path::{Path, PathBuf};
22
23/// Stable on-disk format marker for durable sync outboxes.
24pub const SYNC_OUTBOX_FORMAT_V1: &str = "# appcore-sync-outbox-v1";
25const MAX_OUTBOX_FILE_BYTES: u64 = 64 * 1024 * 1024;
26
27/// Ordered bounded queue that retains replication batches until acknowledgement.
28pub trait SyncOutbox: Send + Sync {
29    /// Enqueues a batch if the current length is below `max_len`.
30    fn try_enqueue(&self, message: SyncMessage, max_len: usize) -> SyncResult<bool>;
31    /// Returns the oldest pending batch.
32    fn front(&self) -> SyncResult<Option<SyncMessage>>;
33    /// Removes the oldest batch only when its identifier matches `batch_id`.
34    fn acknowledge_front(&self, batch_id: &str) -> SyncResult<()>;
35    /// Returns all pending batches in delivery order.
36    fn messages(&self) -> SyncResult<Vec<SyncMessage>>;
37    /// Returns the number of pending batches.
38    fn len(&self) -> SyncResult<usize>;
39    /// Reports whether no batches are pending.
40    fn is_empty(&self) -> SyncResult<bool> {
41        Ok(self.len()? == 0)
42    }
43}
44
45#[derive(Debug, Default)]
46/// Process-local synchronization outbox.
47pub struct InMemorySyncOutbox {
48    messages: Mutex<VecDeque<SyncMessage>>,
49}
50
51impl InMemorySyncOutbox {
52    /// Creates an empty in-memory outbox.
53    pub fn new() -> Self {
54        Self::default()
55    }
56}
57
58impl SyncOutbox for InMemorySyncOutbox {
59    fn try_enqueue(&self, message: SyncMessage, max_len: usize) -> SyncResult<bool> {
60        let mut messages = self.messages.lock();
61        if messages.len() >= max_len {
62            return Ok(false);
63        }
64        messages.push_back(message);
65        Ok(true)
66    }
67
68    fn front(&self) -> SyncResult<Option<SyncMessage>> {
69        Ok(self.messages.lock().front().cloned())
70    }
71
72    fn acknowledge_front(&self, batch_id: &str) -> SyncResult<()> {
73        let mut messages = self.messages.lock();
74        if messages.front().map(|message| message.batch_id.as_str()) != Some(batch_id) {
75            return Err(SyncError::InvalidSyncMessage(
76                "outbox acknowledgement mismatch",
77            ));
78        }
79        messages.pop_front();
80        Ok(())
81    }
82
83    fn messages(&self) -> SyncResult<Vec<SyncMessage>> {
84        Ok(self.messages.lock().iter().cloned().collect())
85    }
86
87    fn len(&self) -> SyncResult<usize> {
88        Ok(self.messages.lock().len())
89    }
90}
91
92#[derive(Debug)]
93/// Crash-consistent file-backed synchronization outbox.
94pub struct FileSyncOutbox {
95    file_path: PathBuf,
96    messages: Mutex<VecDeque<SyncMessage>>,
97}
98
99impl FileSyncOutbox {
100    /// Opens or creates an outbox and validates all existing records.
101    pub fn new(file_path: impl Into<PathBuf>) -> SyncResult<Self> {
102        let file_path = file_path.into();
103        if let Some(parent) = file_path.parent() {
104            std::fs::create_dir_all(parent)
105                .map_err(|error| SyncError::ReplicationFailed(error.to_string()))?;
106        }
107        let _process_lock = acquire_persistence_lock(&file_path)?;
108        let existed = file_path.exists();
109        let messages = if existed {
110            load_messages(&file_path)?
111        } else {
112            VecDeque::new()
113        };
114        let outbox = Self {
115            file_path,
116            messages: Mutex::new(messages),
117        };
118        if !existed {
119            outbox.replace(&outbox.messages.lock())?;
120        }
121        Ok(outbox)
122    }
123
124    /// Returns the durable outbox file path.
125    pub fn file_path(&self) -> &Path {
126        &self.file_path
127    }
128
129    fn replace(&self, messages: &VecDeque<SyncMessage>) -> SyncResult<()> {
130        let mut encoded_file = format!("{SYNC_OUTBOX_FORMAT_V1}\n");
131        for message in messages {
132            let encoded = serde_json::to_vec(message)
133                .map_err(|error| SyncError::ReplicationFailed(error.to_string()))?;
134            encoded_file.push_str(&bytes_to_hex(&encoded));
135            encoded_file.push('\n');
136            if encoded_file.len() as u64 > MAX_OUTBOX_FILE_BYTES {
137                return Err(SyncError::ReplicationFailed(
138                    "sync outbox exceeds configured limit".to_string(),
139                ));
140            }
141        }
142        atomic_write(&self.file_path, encoded_file.as_bytes())
143    }
144}
145
146impl SyncOutbox for FileSyncOutbox {
147    fn try_enqueue(&self, message: SyncMessage, max_len: usize) -> SyncResult<bool> {
148        let mut messages = self.messages.lock();
149        let _process_lock = acquire_persistence_lock(&self.file_path)?;
150        *messages = load_messages(&self.file_path)?;
151        if messages.len() >= max_len {
152            return Ok(false);
153        }
154        let mut updated = messages.clone();
155        updated.push_back(message);
156        self.replace(&updated)?;
157        *messages = updated;
158        Ok(true)
159    }
160
161    fn front(&self) -> SyncResult<Option<SyncMessage>> {
162        let mut messages = self.messages.lock();
163        let _process_lock = acquire_persistence_lock(&self.file_path)?;
164        *messages = load_messages(&self.file_path)?;
165        Ok(messages.front().cloned())
166    }
167
168    fn acknowledge_front(&self, batch_id: &str) -> SyncResult<()> {
169        let mut messages = self.messages.lock();
170        let _process_lock = acquire_persistence_lock(&self.file_path)?;
171        *messages = load_messages(&self.file_path)?;
172        if messages.front().map(|message| message.batch_id.as_str()) != Some(batch_id) {
173            return Err(SyncError::InvalidSyncMessage(
174                "outbox acknowledgement mismatch",
175            ));
176        }
177        let mut updated = messages.clone();
178        updated.pop_front();
179        self.replace(&updated)?;
180        *messages = updated;
181        Ok(())
182    }
183
184    fn messages(&self) -> SyncResult<Vec<SyncMessage>> {
185        let mut messages = self.messages.lock();
186        let _process_lock = acquire_persistence_lock(&self.file_path)?;
187        *messages = load_messages(&self.file_path)?;
188        Ok(messages.iter().cloned().collect())
189    }
190
191    fn len(&self) -> SyncResult<usize> {
192        let mut messages = self.messages.lock();
193        let _process_lock = acquire_persistence_lock(&self.file_path)?;
194        *messages = load_messages(&self.file_path)?;
195        Ok(messages.len())
196    }
197}
198
199fn load_messages(path: &Path) -> SyncResult<VecDeque<SyncMessage>> {
200    let contents = read_bounded_text(path, MAX_OUTBOX_FILE_BYTES)?;
201    let formatted = split_format(&contents, SYNC_OUTBOX_FORMAT_V1)?;
202    let mut messages = VecDeque::new();
203    for (line_number, line) in formatted.body.lines().enumerate() {
204        if line.is_empty() {
205            continue;
206        }
207        let bytes = hex_to_bytes(line).map_err(|_| SyncError::CorruptOutbox {
208            line: line_number + 1,
209        })?;
210        if !bytes.starts_with(b"{") {
211            return Err(SyncError::ReplicationFailed(
212                crate::sync::error::UPDATE_REQUIRED_MESSAGE.to_string(),
213            ));
214        }
215        let message = serde_json::from_slice(&bytes).map_err(|_| SyncError::CorruptOutbox {
216            line: line_number + 1,
217        })?;
218        messages.push_back(message);
219    }
220    Ok(messages)
221}
appcore_sync/sync/outbox.rs

appcore_sync/sync/
outbox.rs
