appcore_sync/sync/
log.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: log.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//! Replication log contracts and local implementations.
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::snapshot::{snapshot_from_records, validate_snapshot, ReplicationSnapshot};
19use sha2::{Digest, Sha256};
20use std::collections::HashMap;
21use std::fs;
22use std::io::Write;
23use std::path::{Path, PathBuf};
24
25/// Stable on-disk format marker for hash-chained replication logs.
26pub const REPLICATION_LOG_FORMAT_V1: &str = "# appcore-replication-log-v1";
27const MAX_REPLICATION_LOG_BYTES: u64 = 256 * 1024 * 1024;
28const MAX_REPLICATION_RECORD_BYTES: usize = 1024 * 1024;
29
30/// Replication log contract.
31pub trait ReplicationLog {
32    /// Appends an unsequenced record and returns its one-based log index.
33    fn append(&mut self, record: Vec<u8>) -> SyncResult<usize>;
34    /// Idempotently appends `record` at a source sequence.
35    fn append_with_sequence(&mut self, record: Vec<u8>, sequence: u64) -> SyncResult<usize>;
36    /// Returns the payload at a source sequence when sequence lookup is supported.
37    fn event_at_sequence(&self, _sequence: u64) -> SyncResult<Option<Vec<u8>>> {
38        Ok(None)
39    }
40    /// Returns payloads after the supplied zero-based log offset.
41    fn events_since(&self, index: usize) -> SyncResult<Vec<Vec<u8>>>;
42    /// Returns the one-based final log index, or zero for an empty log.
43    fn last_index(&self) -> usize;
44    /// Returns the number of records in the log.
45    fn len(&self) -> usize;
46    /// Reports whether the log contains no records.
47    fn is_empty(&self) -> bool;
48    /// Creates a validated portable snapshot when supported.
49    fn create_snapshot(&self) -> SyncResult<ReplicationSnapshot> {
50        Err(SyncError::SnapshotUnsupported)
51    }
52    /// Atomically replaces log contents from a validated snapshot when supported.
53    fn restore_snapshot(&mut self, _snapshot: &ReplicationSnapshot) -> SyncResult<()> {
54        Err(SyncError::SnapshotUnsupported)
55    }
56}
57
58/// In-memory replication log for local sync scenarios.
59#[derive(Debug, Clone, Default, PartialEq, Eq)]
60pub struct InMemoryReplicationLog {
61    events: Vec<ReplicationRecord>,
62    sequence_indices: HashMap<u64, usize>,
63}
64
65#[derive(Debug, Clone, PartialEq, Eq)]
66pub(super) struct ReplicationRecord {
67    pub(super) index: usize,
68    pub(super) sequence: u64,
69    pub(super) payload: Vec<u8>,
70    pub(super) previous_hash: String,
71    pub(super) record_hash: String,
72}
73
74impl InMemoryReplicationLog {
75    /// Creates an empty process-local replication log.
76    pub fn new() -> Self {
77        Self {
78            events: Vec::new(),
79            sequence_indices: HashMap::new(),
80        }
81    }
82
83    /// Idempotently appends a record at a source sequence.
84    pub fn append_with_sequence(&mut self, record: Vec<u8>, sequence: u64) -> SyncResult<usize> {
85        validate_record_size(&record)?;
86        if sequence > 0 {
87            if let Some(existing) = self
88                .sequence_indices
89                .get(&sequence)
90                .and_then(|offset| self.events.get(*offset))
91            {
92                return if existing.payload == record {
93                    Ok(existing.index)
94                } else {
95                    Err(SyncError::SequenceConflict(sequence))
96                };
97            }
98        }
99        let index = self.events.len() + 1;
100        let previous_hash = self
101            .events
102            .last()
103            .map(|record| record.record_hash.clone())
104            .unwrap_or_default();
105        let record_hash = replication_record_hash(&previous_hash, sequence, &record);
106        self.events.push(ReplicationRecord {
107            index,
108            sequence,
109            payload: record,
110            previous_hash,
111            record_hash,
112        });
113        self.sequence_indices.insert(sequence, index - 1);
114        Ok(index)
115    }
116
117    /// Returns the one-based final log index, or zero when empty.
118    pub fn last_index(&self) -> usize {
119        self.events.len()
120    }
121
122    /// Reports whether a source sequence is present.
123    pub fn contains_sequence(&self, sequence: u64) -> bool {
124        self.sequence_indices.contains_key(&sequence)
125    }
126}
127
128impl ReplicationLog for InMemoryReplicationLog {
129    fn append(&mut self, record: Vec<u8>) -> SyncResult<usize> {
130        self.append_with_sequence(record, 0)
131    }
132
133    fn append_with_sequence(&mut self, record: Vec<u8>, sequence: u64) -> SyncResult<usize> {
134        self.append_with_sequence(record, sequence)
135    }
136
137    fn event_at_sequence(&self, sequence: u64) -> SyncResult<Option<Vec<u8>>> {
138        Ok(self
139            .sequence_indices
140            .get(&sequence)
141            .filter(|_| sequence > 0)
142            .and_then(|offset| self.events.get(*offset))
143            .map(|event| event.payload.clone()))
144    }
145
146    fn events_since(&self, index: usize) -> SyncResult<Vec<Vec<u8>>> {
147        if index > self.events.len() {
148            return Err(SyncError::LogIndexOutOfBounds {
149                index,
150                len: self.events.len(),
151            });
152        }
153        Ok(self.events[index..]
154            .iter()
155            .map(|record| record.payload.clone())
156            .collect::<Vec<_>>())
157    }
158
159    fn len(&self) -> usize {
160        self.events.len()
161    }
162
163    fn last_index(&self) -> usize {
164        self.last_index()
165    }
166
167    fn is_empty(&self) -> bool {
168        self.events.is_empty()
169    }
170
171    fn create_snapshot(&self) -> SyncResult<ReplicationSnapshot> {
172        Ok(snapshot_from_records(&self.events))
173    }
174
175    fn restore_snapshot(&mut self, snapshot: &ReplicationSnapshot) -> SyncResult<()> {
176        let records = validate_snapshot(snapshot)?;
177        self.sequence_indices = sequence_indices(&records);
178        self.events = records;
179        Ok(())
180    }
181}
182
183/// File-backed append-only replication log for local runtime sync.
184#[derive(Debug, Clone, PartialEq, Eq)]
185pub struct FileReplicationLog {
186    file_path: PathBuf,
187    events: Vec<ReplicationRecord>,
188    sequence_indices: HashMap<u64, usize>,
189}
190
191impl FileReplicationLog {
192    /// Opens a relative append-only log below `storage_path`.
193    pub fn new(storage_path: impl AsRef<Path>, relative_path: &str) -> SyncResult<Self> {
194        let storage_root = storage_path.as_ref();
195        let relative = PathBuf::from(relative_path);
196        if relative.as_os_str().is_empty()
197            || relative.is_absolute()
198            || relative
199                .components()
200                .any(|component| matches!(component, std::path::Component::ParentDir))
201        {
202            return Err(SyncError::ReplicationFailed(
203                "invalid replication log path".to_string(),
204            ));
205        }
206        let file_path = storage_root.join(&relative);
207        if let Some(parent) = file_path.parent() {
208            fs::create_dir_all(parent)
209                .map_err(|err| SyncError::ReplicationFailed(err.to_string()))?;
210        }
211        let _process_lock = acquire_persistence_lock(&file_path)?;
212        if !file_path.exists() {
213            atomic_write(
214                &file_path,
215                format!("{REPLICATION_LOG_FORMAT_V1}\n").as_bytes(),
216            )?;
217        }
218        let mut log = Self {
219            file_path,
220            events: Vec::new(),
221            sequence_indices: HashMap::new(),
222        };
223        log.reload_unlocked()?;
224        Ok(log)
225    }
226
227    /// Returns the durable log file path.
228    pub fn file_path(&self) -> &Path {
229        &self.file_path
230    }
231
232    /// Re-reads and validates all durable records from disk.
233    pub fn reload(&mut self) -> SyncResult<()> {
234        let _process_lock = acquire_persistence_lock(&self.file_path)?;
235        self.reload_unlocked()
236    }
237
238    fn reload_unlocked(&mut self) -> SyncResult<()> {
239        let text = read_bounded_text(&self.file_path, MAX_REPLICATION_LOG_BYTES)?;
240        let formatted = split_format(&text, REPLICATION_LOG_FORMAT_V1)?;
241        let (records, recovered_tail) = parse_replication_records(formatted.body)?;
242        if recovered_tail {
243            write_replication_records(&self.file_path, &records)?;
244        }
245        self.sequence_indices = sequence_indices(&records);
246        self.events = records;
247        Ok(())
248    }
249}
250
251impl ReplicationLog for FileReplicationLog {
252    fn append(&mut self, record: Vec<u8>) -> SyncResult<usize> {
253        self.append_with_sequence(record, 0)
254    }
255
256    fn append_with_sequence(&mut self, record: Vec<u8>, sequence: u64) -> SyncResult<usize> {
257        let _process_lock = acquire_persistence_lock(&self.file_path)?;
258        self.reload_unlocked()?;
259        validate_record_size(&record)?;
260        if sequence > 0 {
261            if let Some(existing) = self
262                .sequence_indices
263                .get(&sequence)
264                .and_then(|offset| self.events.get(*offset))
265            {
266                return if existing.payload == record {
267                    Ok(existing.index)
268                } else {
269                    Err(SyncError::SequenceConflict(sequence))
270                };
271            }
272        }
273        let previous_hash = self
274            .events
275            .last()
276            .map(|entry| entry.record_hash.clone())
277            .unwrap_or_default();
278        let record_hash = replication_record_hash(&previous_hash, sequence, &record);
279        let line = format!(
280            "{}\t{}\t{}\t{}\n",
281            sequence,
282            bytes_to_hex(&record),
283            previous_hash,
284            record_hash
285        );
286        let current_bytes = fs::metadata(&self.file_path)
287            .map_err(|err| SyncError::ReplicationFailed(err.to_string()))?
288            .len();
289        if current_bytes
290            .checked_add(line.len() as u64)
291            .is_none_or(|bytes| bytes > MAX_REPLICATION_LOG_BYTES)
292        {
293            return Err(SyncError::ReplicationFailed(
294                "replication log exceeds configured limit".to_string(),
295            ));
296        }
297        let mut file = fs::OpenOptions::new()
298            .append(true)
299            .open(&self.file_path)
300            .map_err(|err| SyncError::ReplicationFailed(err.to_string()))?;
301        file.write_all(line.as_bytes())
302            .map_err(|err| SyncError::ReplicationFailed(err.to_string()))?;
303        file.sync_data()
304            .map_err(|err| SyncError::ReplicationFailed(err.to_string()))?;
305        let index = self.events.len() + 1;
306        self.events.push(ReplicationRecord {
307            index,
308            sequence,
309            payload: record,
310            previous_hash,
311            record_hash,
312        });
313        self.sequence_indices.insert(sequence, index - 1);
314        Ok(index)
315    }
316
317    fn event_at_sequence(&self, sequence: u64) -> SyncResult<Option<Vec<u8>>> {
318        Ok(self
319            .sequence_indices
320            .get(&sequence)
321            .filter(|_| sequence > 0)
322            .and_then(|offset| self.events.get(*offset))
323            .map(|event| event.payload.clone()))
324    }
325
326    fn events_since(&self, index: usize) -> SyncResult<Vec<Vec<u8>>> {
327        if index > self.events.len() {
328            return Err(SyncError::LogIndexOutOfBounds {
329                index,
330                len: self.events.len(),
331            });
332        }
333        Ok(self.events[index..]
334            .iter()
335            .map(|record| record.payload.clone())
336            .collect::<Vec<_>>())
337    }
338
339    fn last_index(&self) -> usize {
340        self.events.len()
341    }
342
343    fn len(&self) -> usize {
344        self.events.len()
345    }
346
347    fn is_empty(&self) -> bool {
348        self.events.is_empty()
349    }
350
351    fn create_snapshot(&self) -> SyncResult<ReplicationSnapshot> {
352        Ok(snapshot_from_records(&self.events))
353    }
354
355    fn restore_snapshot(&mut self, snapshot: &ReplicationSnapshot) -> SyncResult<()> {
356        let _process_lock = acquire_persistence_lock(&self.file_path)?;
357        let records = validate_snapshot(snapshot)?;
358        write_replication_records(&self.file_path, &records)?;
359        self.sequence_indices = sequence_indices(&records);
360        self.events = records;
361        Ok(())
362    }
363}
364
365fn sequence_indices(records: &[ReplicationRecord]) -> HashMap<u64, usize> {
366    records
367        .iter()
368        .enumerate()
369        .map(|(offset, record)| (record.sequence, offset))
370        .collect()
371}
372
373fn parse_replication_records(body: &str) -> SyncResult<(Vec<ReplicationRecord>, bool)> {
374    let (complete, recovered_tail) = complete_record_prefix(body);
375    let mut records = Vec::<ReplicationRecord>::new();
376    let mut sequences = HashMap::<u64, usize>::new();
377    let mut chain_head = String::new();
378    for (line_number, line) in complete.lines().enumerate() {
379        if line.trim().is_empty() {
380            continue;
381        }
382        let parts = line.split('\t').collect::<Vec<_>>();
383        if parts.len() != 4 {
384            let reason = if line.contains('\t') {
385                "invalid field count"
386            } else {
387                "missing separator"
388            };
389            return Err(corrupt_log(line_number, reason));
390        }
391        let sequence = parts[0]
392            .parse::<u64>()
393            .map_err(|_| corrupt_log(line_number, "invalid sequence"))?;
394        let payload =
395            hex_to_bytes(parts[1]).map_err(|_| corrupt_log(line_number, "invalid event hex"))?;
396        validate_record_size(&payload)?;
397        let record_hash = replication_record_hash(&chain_head, sequence, &payload);
398        if parts[2] != chain_head || parts[3] != record_hash {
399            return Err(corrupt_log(line_number, "hash chain mismatch"));
400        }
401        if sequence > 0 {
402            if let Some(offset) = sequences.get(&sequence) {
403                if records[*offset].payload != payload {
404                    return Err(SyncError::SequenceConflict(sequence));
405                }
406                return Err(corrupt_log(line_number, "duplicate sequence"));
407            }
408        }
409        let index = records.len() + 1;
410        records.push(ReplicationRecord {
411            index,
412            sequence,
413            payload,
414            previous_hash: chain_head,
415            record_hash: record_hash.clone(),
416        });
417        sequences.insert(sequence, index - 1);
418        chain_head = record_hash;
419    }
420    Ok((records, recovered_tail))
421}
422
423fn complete_record_prefix(body: &str) -> (&str, bool) {
424    if body.is_empty() || body.ends_with('\n') {
425        return (body, false);
426    }
427    match body.rfind('\n') {
428        Some(last_newline) => (&body[..=last_newline], true),
429        None => ("", true),
430    }
431}
432
433fn corrupt_log(line_number: usize, reason: &'static str) -> SyncError {
434    SyncError::CorruptReplicationLog {
435        line: line_number + 1,
436        reason,
437    }
438}
439
440fn write_replication_records(path: &Path, records: &[ReplicationRecord]) -> SyncResult<()> {
441    let mut output = format!("{REPLICATION_LOG_FORMAT_V1}\n");
442    let mut previous_hash = String::new();
443    for record in records {
444        validate_record_size(&record.payload)?;
445        let hash = replication_record_hash(&previous_hash, record.sequence, &record.payload);
446        output.push_str(&format!(
447            "{}\t{}\t{}\t{}\n",
448            record.sequence,
449            bytes_to_hex(&record.payload),
450            previous_hash,
451            hash
452        ));
453        if output.len() as u64 > MAX_REPLICATION_LOG_BYTES {
454            return Err(SyncError::ReplicationFailed(
455                "replication log exceeds configured limit".to_string(),
456            ));
457        }
458        previous_hash = hash;
459    }
460    atomic_write(path, output.as_bytes())
461}
462
463pub(super) fn validate_record_size(payload: &[u8]) -> SyncResult<()> {
464    if payload.len() > MAX_REPLICATION_RECORD_BYTES {
465        return Err(SyncError::ReplicationFailed(
466            "replication record exceeds size limit".to_string(),
467        ));
468    }
469    Ok(())
470}
471
472pub(super) fn replication_record_hash(
473    previous_hash: &str,
474    sequence: u64,
475    payload: &[u8],
476) -> String {
477    let mut hasher = Sha256::new();
478    hasher.update(REPLICATION_LOG_FORMAT_V1.as_bytes());
479    hasher.update((previous_hash.len() as u64).to_be_bytes());
480    hasher.update(previous_hash.as_bytes());
481    hasher.update(sequence.to_be_bytes());
482    hasher.update((payload.len() as u64).to_be_bytes());
483    hasher.update(payload);
484    bytes_to_hex(&hasher.finalize())
485}
appcore_sync/sync/log.rs

appcore_sync/sync/
log.rs
