appcore_sync/sync/
checkpoint.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: checkpoint.rs
4//    ##   ## ##   ##    P: AppCore-Runtime
5//         ## ##
6//                       C: 2026/06/02 13:08:16 by dnettoRaw
7//    ##   ## ##   ##    U: 2026/08/02 13:24:05 by dnettoRaw
8//      ###########      S: 1.0.1-rc.8
9// =============================================================================
10
11//! Per-peer checkpoint contracts and local implementations.
12
13use crate::sync::error::{SyncError, SyncResult, UPDATE_REQUIRED_MESSAGE};
14use crate::sync::persistence::{
15    acquire_persistence_lock, atomic_write, atomic_write_with, reject_symlink,
16};
17use parking_lot::Mutex;
18use std::collections::BTreeMap;
19use std::fs::{self, File};
20use std::io::{BufRead, BufReader, BufWriter, Read, Write};
21use std::path::{Path, PathBuf};
22use std::sync::Arc;
23
24/// Stable on-disk format marker for peer checkpoints.
25pub const SYNC_CHECKPOINT_FORMAT_V1: &str = "# appcore-sync-checkpoint-v1";
26/// Maximum bytes accepted in one checkpoint file.
27pub const MAX_CHECKPOINT_FILE_BYTES: u64 = 8 * 1024 * 1024;
28/// Maximum UTF-8 bytes accepted in one checkpoint peer identifier.
29pub const MAX_CHECKPOINT_PEER_ID_BYTES: usize = 256;
30/// Maximum non-empty records accepted in one checkpoint file.
31pub const MAX_CHECKPOINT_RECORDS: usize = 65_536;
32const MAX_CHECKPOINT_LINE_BYTES: usize = MAX_CHECKPOINT_PEER_ID_BYTES + 87;
33const CHECKPOINT_BUFFER_BYTES: usize = 16 * 1024;
34type Checkpoint = (u64, String);
35type CheckpointMap = BTreeMap<String, Checkpoint>;
36
37#[derive(Clone, Copy, Debug, Eq, PartialEq)]
38enum LineStatus {
39    Complete,
40    Partial,
41    End,
42}
43
44/// Sync checkpoint storage contract by peer id.
45pub trait SyncCheckpointStore: Send + Sync {
46    /// Returns the last accepted sequence and batch hash for `peer_id`.
47    fn get_checkpoint(&self, peer_id: &str) -> SyncResult<Option<(u64, String)>>;
48    /// Atomically replaces the sequence and batch hash for `peer_id`.
49    fn set_checkpoint(&self, peer_id: &str, sequence: u64, hash: &str) -> SyncResult<()>;
50
51    /// Returns the last accepted sequence, or zero when no checkpoint exists.
52    fn get_last_sequence(&self, peer_id: &str) -> SyncResult<u64> {
53        Ok(self
54            .get_checkpoint(peer_id)?
55            .map(|(seq, _)| seq)
56            .unwrap_or(0))
57    }
58
59    /// Updates only the accepted sequence while preserving the stored hash.
60    fn set_last_sequence(&self, peer_id: &str, sequence: u64) -> SyncResult<()> {
61        let hash = self
62            .get_checkpoint(peer_id)?
63            .map(|(_, h)| h)
64            .unwrap_or_default();
65        self.set_checkpoint(peer_id, sequence, &hash)
66    }
67}
68
69/// In-memory checkpoint store for tests/local runtime.
70#[derive(Debug, Clone, Default)]
71pub struct InMemorySyncCheckpointStore {
72    checkpoints: Arc<Mutex<CheckpointMap>>,
73}
74
75impl InMemorySyncCheckpointStore {
76    /// Creates an empty process-local checkpoint store.
77    pub fn new() -> Self {
78        Self {
79            checkpoints: Arc::new(Mutex::new(BTreeMap::new())),
80        }
81    }
82}
83
84impl SyncCheckpointStore for InMemorySyncCheckpointStore {
85    fn get_checkpoint(&self, peer_id: &str) -> SyncResult<Option<(u64, String)>> {
86        validate_peer_id(peer_id)?;
87        let guard = self.checkpoints.lock();
88        Ok(guard.get(peer_id).cloned())
89    }
90
91    fn set_checkpoint(&self, peer_id: &str, sequence: u64, hash: &str) -> SyncResult<()> {
92        validate_peer_id(peer_id)?;
93        validate_checkpoint_hash(hash)?;
94        let mut guard = self.checkpoints.lock();
95        guard.insert(peer_id.to_string(), (sequence, hash.to_string()));
96        Ok(())
97    }
98}
99
100/// File-backed checkpoint store (line-based `peer=sequence,hash`).
101#[derive(Debug, Clone)]
102pub struct FileSyncCheckpointStore {
103    file_path: PathBuf,
104    lock: Arc<Mutex<()>>,
105}
106
107impl FileSyncCheckpointStore {
108    /// Opens or creates an atomic line-based checkpoint file.
109    pub fn new(file_path: impl Into<PathBuf>) -> SyncResult<Self> {
110        let file_path = file_path.into();
111        if let Some(parent) = file_path.parent() {
112            fs::create_dir_all(parent)
113                .map_err(|err| crate::sync::error::SyncError::ReplicationFailed(err.to_string()))?;
114        }
115        let _process_lock = acquire_persistence_lock(&file_path)?;
116        if !file_path.exists() {
117            atomic_write(
118                &file_path,
119                format!("{SYNC_CHECKPOINT_FORMAT_V1}\n").as_bytes(),
120            )?;
121        }
122        let store = Self {
123            file_path,
124            lock: Arc::new(Mutex::new(())),
125        };
126        store.validate_state()?;
127        Ok(store)
128    }
129
130    /// Returns the durable checkpoint file path.
131    pub fn file_path(&self) -> &Path {
132        &self.file_path
133    }
134
135    fn validate_state(&self) -> SyncResult<()> {
136        scan_checkpoint_file(&self.file_path, |_, _, _| Ok(()))
137    }
138
139    fn read_map(&self) -> SyncResult<CheckpointMap> {
140        let mut map = BTreeMap::new();
141        scan_checkpoint_file(&self.file_path, |peer_id, sequence, hash| {
142            map.insert(peer_id.to_string(), (sequence, hash.to_string()));
143            Ok(())
144        })?;
145        Ok(map)
146    }
147
148    fn write_map(&self, map: &CheckpointMap) -> SyncResult<()> {
149        if map.len() > MAX_CHECKPOINT_RECORDS {
150            return Err(checkpoint_failure("checkpoint record limit exceeded"));
151        }
152        let encoded_bytes = encoded_checkpoint_bytes(map)?;
153        if encoded_bytes > MAX_CHECKPOINT_FILE_BYTES {
154            return Err(checkpoint_failure(
155                "persistent file exceeds configured limit",
156            ));
157        }
158        atomic_write_with(&self.file_path, |file| write_checkpoint_map(file, map))
159    }
160}
161
162impl SyncCheckpointStore for FileSyncCheckpointStore {
163    fn get_checkpoint(&self, peer_id: &str) -> SyncResult<Option<(u64, String)>> {
164        validate_peer_id(peer_id)?;
165        let _guard = self.lock.lock();
166        let _process_lock = acquire_persistence_lock(&self.file_path)?;
167        let mut checkpoint = None;
168        scan_checkpoint_file(&self.file_path, |stored_peer, sequence, hash| {
169            if stored_peer == peer_id {
170                checkpoint = Some((sequence, hash.to_string()));
171            }
172            Ok(())
173        })?;
174        Ok(checkpoint)
175    }
176
177    fn set_checkpoint(&self, peer_id: &str, sequence: u64, hash: &str) -> SyncResult<()> {
178        validate_peer_id(peer_id)?;
179        validate_checkpoint_hash(hash)?;
180        let _guard = self.lock.lock();
181        let _process_lock = acquire_persistence_lock(&self.file_path)?;
182        let mut map = self.read_map()?;
183        map.insert(peer_id.to_string(), (sequence, hash.to_string()));
184        self.write_map(&map)
185    }
186}
187
188fn validate_peer_id(peer_id: &str) -> SyncResult<()> {
189    if peer_id.is_empty() || peer_id.len() > MAX_CHECKPOINT_PEER_ID_BYTES {
190        return Err(crate::sync::error::SyncError::InvalidPeerId);
191    }
192    if !peer_id
193        .chars()
194        .all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '.' | '_' | ':' | '-'))
195    {
196        return Err(crate::sync::error::SyncError::InvalidPeerId);
197    }
198    Ok(())
199}
200
201fn validate_checkpoint_hash(hash: &str) -> SyncResult<()> {
202    if hash.is_empty() || (hash.len() == 64 && hash.bytes().all(|byte| byte.is_ascii_hexdigit())) {
203        return Ok(());
204    }
205    Err(crate::sync::error::SyncError::ReplicationFailed(
206        "invalid checkpoint hash".to_string(),
207    ))
208}
209
210fn scan_checkpoint_file(
211    path: &Path,
212    mut visit: impl FnMut(&str, u64, &str) -> SyncResult<()>,
213) -> SyncResult<()> {
214    reject_symlink(path)?;
215    let file = File::open(path).map_err(checkpoint_io)?;
216    if file.metadata().map_err(checkpoint_io)?.len() > MAX_CHECKPOINT_FILE_BYTES {
217        return Err(checkpoint_failure(
218            "persistent file exceeds configured limit",
219        ));
220    }
221    let limited = file.take(MAX_CHECKPOINT_FILE_BYTES.saturating_add(1));
222    let mut reader = BufReader::with_capacity(CHECKPOINT_BUFFER_BYTES, limited);
223    let mut line = Vec::with_capacity(MAX_CHECKPOINT_LINE_BYTES);
224    let mut consumed = 0u64;
225    let marker = read_checkpoint_line(&mut reader, &mut line, &mut consumed)?;
226    validate_marker(marker, &line)?;
227    if marker == LineStatus::Partial {
228        return Ok(());
229    }
230    let mut records = 0usize;
231    loop {
232        let status = read_checkpoint_line(&mut reader, &mut line, &mut consumed)?;
233        if status == LineStatus::End {
234            return Ok(());
235        }
236        let text = checkpoint_line_text(&line)?;
237        if !text.trim().is_empty() {
238            records = records
239                .checked_add(1)
240                .filter(|count| *count <= MAX_CHECKPOINT_RECORDS)
241                .ok_or_else(|| checkpoint_failure("checkpoint record limit exceeded"))?;
242            let (peer_id, sequence, hash) = parse_checkpoint_line(text)?;
243            visit(peer_id, sequence, hash)?;
244        }
245        if status == LineStatus::Partial {
246            return Ok(());
247        }
248    }
249}
250
251fn read_checkpoint_line<R: BufRead>(
252    reader: &mut R,
253    line: &mut Vec<u8>,
254    consumed: &mut u64,
255) -> SyncResult<LineStatus> {
256    line.clear();
257    loop {
258        let available = reader.fill_buf().map_err(checkpoint_io)?;
259        if available.is_empty() {
260            return Ok(if line.is_empty() {
261                LineStatus::End
262            } else {
263                LineStatus::Partial
264            });
265        }
266        let newline = available.iter().position(|byte| *byte == b'\n');
267        let used = newline.map_or(available.len(), |index| index + 1);
268        let body_bytes = newline.unwrap_or(available.len());
269        if line.len().saturating_add(body_bytes) > MAX_CHECKPOINT_LINE_BYTES {
270            return Err(checkpoint_failure("checkpoint line exceeds size limit"));
271        }
272        line.extend_from_slice(&available[..body_bytes]);
273        reader.consume(used);
274        *consumed = consumed.saturating_add(used as u64);
275        if *consumed > MAX_CHECKPOINT_FILE_BYTES {
276            return Err(checkpoint_failure(
277                "persistent file exceeds configured limit",
278            ));
279        }
280        if newline.is_some() {
281            return Ok(LineStatus::Complete);
282        }
283    }
284}
285
286fn validate_marker(status: LineStatus, line: &[u8]) -> SyncResult<()> {
287    let marker =
288        std::str::from_utf8(line).map_err(|_| checkpoint_failure("invalid checkpoint UTF-8"))?;
289    if status == LineStatus::End || marker != SYNC_CHECKPOINT_FORMAT_V1 {
290        return Err(checkpoint_failure(UPDATE_REQUIRED_MESSAGE));
291    }
292    Ok(())
293}
294
295fn checkpoint_line_text(line: &[u8]) -> SyncResult<&str> {
296    let line = line.strip_suffix(b"\r").unwrap_or(line);
297    std::str::from_utf8(line).map_err(|_| checkpoint_failure("invalid checkpoint UTF-8"))
298}
299
300fn parse_checkpoint_line(line: &str) -> SyncResult<(&str, u64, &str)> {
301    let (peer_id, rest) = line
302        .split_once('=')
303        .ok_or_else(|| checkpoint_failure("invalid checkpoint line"))?;
304    validate_peer_id(peer_id)?;
305    let (sequence_text, hash) = rest.split_once(',').unwrap_or((rest, ""));
306    let sequence = sequence_text
307        .parse::<u64>()
308        .map_err(|_| checkpoint_failure("invalid checkpoint sequence"))?;
309    validate_checkpoint_hash(hash)?;
310    Ok((peer_id, sequence, hash))
311}
312
313fn encoded_checkpoint_bytes(map: &CheckpointMap) -> SyncResult<u64> {
314    let mut bytes = (SYNC_CHECKPOINT_FORMAT_V1.len() + 1) as u64;
315    for (peer_id, (sequence, hash)) in map {
316        let record_bytes = peer_id
317            .len()
318            .checked_add(decimal_digits(*sequence))
319            .and_then(|size| size.checked_add(hash.len() + 3))
320            .ok_or_else(|| checkpoint_failure("checkpoint size overflow"))?;
321        bytes = bytes
322            .checked_add(record_bytes as u64)
323            .ok_or_else(|| checkpoint_failure("checkpoint size overflow"))?;
324    }
325    Ok(bytes)
326}
327
328fn write_checkpoint_map(file: &mut File, map: &CheckpointMap) -> SyncResult<()> {
329    let mut writer = BufWriter::with_capacity(CHECKPOINT_BUFFER_BYTES, file);
330    writeln!(writer, "{SYNC_CHECKPOINT_FORMAT_V1}").map_err(checkpoint_io)?;
331    for (peer_id, (sequence, hash)) in map {
332        writeln!(writer, "{peer_id}={sequence},{hash}").map_err(checkpoint_io)?;
333    }
334    writer.flush().map_err(checkpoint_io)
335}
336
337fn checkpoint_io(error: std::io::Error) -> SyncError {
338    SyncError::ReplicationFailed(error.to_string())
339}
340
341fn checkpoint_failure(message: &str) -> SyncError {
342    SyncError::ReplicationFailed(message.to_string())
343}
344
345fn decimal_digits(mut value: u64) -> usize {
346    let mut digits = 1;
347    while value >= 10 {
348        value /= 10;
349        digits += 1;
350    }
351    digits
352}
appcore_sync/sync/checkpoint.rs

appcore_sync/sync/
checkpoint.rs
