appcore_sync/sync/
outbox.rs

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

appcore_sync/sync/
outbox.rs
