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;
14use crate::sync::error::{SyncError, SyncResult};
15use crate::sync::snapshot::{
16    into_replication_records, snapshot_from_records, validate_snapshot, ReplicationSnapshot,
17};
18use sha2::{Digest, Sha256};
19use std::borrow::Cow;
20
21pub use crate::sync::log_file::FileReplicationLog;
22
23/// Stable on-disk format marker for hash-chained replication logs.
24pub const REPLICATION_LOG_FORMAT_V1: &str = "# appcore-replication-log-v1";
25pub(super) const MAX_REPLICATION_LOG_BYTES: u64 = 256 * 1024 * 1024;
26pub(super) const MAX_REPLICATION_RECORD_BYTES: usize = 1024 * 1024;
27pub(super) const MAX_REPLICATION_RECORDS: usize = 262_144;
28/// Maximum records returned by one bounded replication-log page.
29pub const MAX_REPLICATION_PAGE_RECORDS: usize = 1024;
30/// Maximum aggregate payload bytes returned by one replication-log page.
31pub const MAX_REPLICATION_PAGE_BYTES: usize = 48 * 1024 * 1024;
32/// Maximum raw event bytes grouped into one Runtime HTTP sync batch.
33pub const MAX_SYNC_BATCH_PAYLOAD_BYTES: usize = 1024 * 1024;
34
35/// Replication log contract.
36pub trait ReplicationLog {
37    /// Appends an unsequenced record and returns its one-based log index.
38    fn append(&mut self, record: Vec<u8>) -> SyncResult<usize>;
39    /// Idempotently appends `record` at a source sequence.
40    fn append_with_sequence(&mut self, record: Vec<u8>, sequence: u64) -> SyncResult<usize>;
41    /// Returns the payload at a source sequence when sequence lookup is supported.
42    fn event_at_sequence(&self, _sequence: u64) -> SyncResult<Option<Vec<u8>>> {
43        Ok(None)
44    }
45    /// Returns payloads after the supplied zero-based log offset.
46    fn events_since(&self, index: usize) -> SyncResult<Vec<Vec<u8>>>;
47    /// Returns one payload page with bounded record count and payload bytes.
48    ///
49    /// Compatibility providers may use the default full-read adapter. Durable
50    /// providers should override this method to enforce both limits before
51    /// materializing payloads.
52    /// The default moves selected payloads without another deep copy, but cannot
53    /// bound the allocation performed by `events_since` before selection.
54    fn events_page(
55        &self,
56        index: usize,
57        max_records: usize,
58        max_bytes: usize,
59    ) -> SyncResult<Vec<Vec<u8>>> {
60        validate_page_limits(max_records, max_bytes)?;
61        let events = self.events_since(index)?;
62        bounded_page(events.into_iter().map(Cow::Owned), max_records, max_bytes)
63    }
64    /// Returns the one-based final log index, or zero for an empty log.
65    fn last_index(&self) -> SyncResult<usize>;
66    /// Returns the number of records in the log.
67    fn len(&self) -> SyncResult<usize>;
68    /// Reports whether the log contains no records.
69    fn is_empty(&self) -> SyncResult<bool>;
70    /// Creates a validated portable snapshot when supported.
71    fn create_snapshot(&self) -> SyncResult<ReplicationSnapshot> {
72        Err(SyncError::SnapshotUnsupported)
73    }
74    /// Atomically replaces log contents from a validated snapshot when supported.
75    fn restore_snapshot(&mut self, _snapshot: &ReplicationSnapshot) -> SyncResult<()> {
76        Err(SyncError::SnapshotUnsupported)
77    }
78}
79
80/// In-memory replication log for local sync scenarios.
81#[derive(Debug, Clone, Default, PartialEq, Eq)]
82pub struct InMemoryReplicationLog {
83    events: Vec<ReplicationRecord>,
84    /// Sorted sequence-to-event offsets; a flat index avoids hash-bucket
85    /// overhead while retaining logarithmic lookup.
86    sequence_indices: Vec<(u64, usize)>,
87}
88
89#[derive(Debug, Clone, PartialEq, Eq)]
90pub(super) struct ReplicationRecord {
91    pub(super) index: usize,
92    pub(super) sequence: u64,
93    pub(super) payload: Vec<u8>,
94    pub(super) previous_hash: String,
95    pub(super) record_hash: String,
96}
97
98impl InMemoryReplicationLog {
99    /// Creates an empty process-local replication log.
100    pub fn new() -> Self {
101        Self {
102            events: Vec::new(),
103            sequence_indices: Vec::new(),
104        }
105    }
106
107    /// Idempotently appends a record at a source sequence.
108    pub fn append_with_sequence(&mut self, record: Vec<u8>, sequence: u64) -> SyncResult<usize> {
109        validate_record_size(&record)?;
110        validate_record_count(self.events.len().saturating_add(1))?;
111        if sequence > 0 {
112            if let Some(existing) = self
113                .sequence_offset(sequence)
114                .and_then(|offset| self.events.get(offset))
115            {
116                return if existing.payload == record {
117                    Ok(existing.index)
118                } else {
119                    Err(SyncError::SequenceConflict(sequence))
120                };
121            }
122        }
123        let index = self.events.len() + 1;
124        let previous_hash = self
125            .events
126            .last()
127            .map(|record| record.record_hash.clone())
128            .unwrap_or_default();
129        let record_hash = replication_record_hash(&previous_hash, sequence, &record);
130        self.events.push(ReplicationRecord {
131            index,
132            sequence,
133            payload: record,
134            previous_hash,
135            record_hash,
136        });
137        insert_sequence_index(&mut self.sequence_indices, sequence, index - 1);
138        Ok(index)
139    }
140
141    /// Returns the one-based final log index, or zero when empty.
142    pub fn last_index(&self) -> usize {
143        self.events.len()
144    }
145
146    /// Reports whether a source sequence is present.
147    pub fn contains_sequence(&self, sequence: u64) -> bool {
148        self.sequence_offset(sequence).is_some()
149    }
150
151    fn sequence_offset(&self, sequence: u64) -> Option<usize> {
152        self.sequence_indices
153            .binary_search_by_key(&sequence, |(value, _)| *value)
154            .ok()
155            .map(|position| self.sequence_indices[position].1)
156    }
157
158    /// Restores a snapshot by moving its payload allocations into the log.
159    ///
160    /// This consuming variant avoids retaining the source snapshot and the
161    /// destination record payloads at the same time. Use the trait method
162    /// [`ReplicationLog::restore_snapshot`] when the caller must keep a
163    /// borrowed snapshot for compatibility.
164    pub fn restore_snapshot_owned(&mut self, snapshot: ReplicationSnapshot) -> SyncResult<()> {
165        let records = into_replication_records(snapshot)?;
166        self.sequence_indices = sequence_indices(&records);
167        self.events = records;
168        Ok(())
169    }
170}
171
172impl ReplicationLog for InMemoryReplicationLog {
173    fn append(&mut self, record: Vec<u8>) -> SyncResult<usize> {
174        self.append_with_sequence(record, 0)
175    }
176
177    fn append_with_sequence(&mut self, record: Vec<u8>, sequence: u64) -> SyncResult<usize> {
178        self.append_with_sequence(record, sequence)
179    }
180
181    fn event_at_sequence(&self, sequence: u64) -> SyncResult<Option<Vec<u8>>> {
182        Ok(self
183            .sequence_offset(sequence)
184            .filter(|_| sequence > 0)
185            .and_then(|offset| self.events.get(offset))
186            .map(|event| event.payload.clone()))
187    }
188
189    fn events_since(&self, index: usize) -> SyncResult<Vec<Vec<u8>>> {
190        if index > self.events.len() {
191            return Err(SyncError::LogIndexOutOfBounds {
192                index,
193                len: self.events.len(),
194            });
195        }
196        Ok(self.events[index..]
197            .iter()
198            .map(|record| record.payload.clone())
199            .collect::<Vec<_>>())
200    }
201
202    fn events_page(
203        &self,
204        index: usize,
205        max_records: usize,
206        max_bytes: usize,
207    ) -> SyncResult<Vec<Vec<u8>>> {
208        validate_log_index(index, self.events.len())?;
209        validate_page_limits(max_records, max_bytes)?;
210        bounded_page(
211            self.events[index..]
212                .iter()
213                .map(|record| Cow::Borrowed(record.payload.as_slice())),
214            max_records,
215            max_bytes,
216        )
217    }
218
219    fn len(&self) -> SyncResult<usize> {
220        Ok(self.events.len())
221    }
222
223    fn last_index(&self) -> SyncResult<usize> {
224        Ok(self.last_index())
225    }
226
227    fn is_empty(&self) -> SyncResult<bool> {
228        Ok(self.events.is_empty())
229    }
230
231    fn create_snapshot(&self) -> SyncResult<ReplicationSnapshot> {
232        Ok(snapshot_from_records(&self.events))
233    }
234
235    fn restore_snapshot(&mut self, snapshot: &ReplicationSnapshot) -> SyncResult<()> {
236        let records = validate_snapshot(snapshot)?;
237        self.sequence_indices = sequence_indices(&records);
238        self.events = records;
239        Ok(())
240    }
241}
242
243fn sequence_indices(records: &[ReplicationRecord]) -> Vec<(u64, usize)> {
244    let mut indices = Vec::with_capacity(records.len());
245    for (offset, record) in records.iter().enumerate() {
246        insert_sequence_index(&mut indices, record.sequence, offset);
247    }
248    indices
249}
250
251fn insert_sequence_index(indices: &mut Vec<(u64, usize)>, sequence: u64, offset: usize) {
252    match indices.binary_search_by_key(&sequence, |(value, _)| *value) {
253        Ok(position) => indices[position] = (sequence, offset),
254        Err(position) => indices.insert(position, (sequence, offset)),
255    }
256}
257
258pub(super) fn validate_record_size(payload: &[u8]) -> SyncResult<()> {
259    if payload.len() > MAX_REPLICATION_RECORD_BYTES {
260        return Err(SyncError::ReplicationFailed(
261            "replication record exceeds size limit".to_string(),
262        ));
263    }
264    Ok(())
265}
266
267pub(super) fn validate_record_count(record_count: usize) -> SyncResult<()> {
268    if record_count > MAX_REPLICATION_RECORDS {
269        return Err(SyncError::ReplicationFailed(
270            "replication record limit exceeded".to_string(),
271        ));
272    }
273    Ok(())
274}
275
276pub(super) fn validate_page_limits(max_records: usize, max_bytes: usize) -> SyncResult<()> {
277    if max_records == 0
278        || max_records > MAX_REPLICATION_PAGE_RECORDS
279        || max_bytes == 0
280        || max_bytes > MAX_REPLICATION_PAGE_BYTES
281    {
282        return Err(SyncError::ReplicationFailed(
283            "invalid replication page limits".to_string(),
284        ));
285    }
286    Ok(())
287}
288
289pub(super) fn validate_log_index(index: usize, len: usize) -> SyncResult<()> {
290    if index > len {
291        return Err(SyncError::LogIndexOutOfBounds { index, len });
292    }
293    Ok(())
294}
295
296fn bounded_page<'a>(
297    payloads: impl Iterator<Item = Cow<'a, [u8]>>,
298    max_records: usize,
299    max_bytes: usize,
300) -> SyncResult<Vec<Vec<u8>>> {
301    let mut page = Vec::with_capacity(max_records);
302    let mut bytes = 0usize;
303    for payload in payloads.take(max_records) {
304        let next = bytes
305            .checked_add(payload.len())
306            .ok_or_else(|| SyncError::ReplicationFailed("replication page overflow".to_string()))?;
307        if next > max_bytes {
308            if page.is_empty() {
309                return Err(SyncError::ReplicationFailed(
310                    "replication page byte limit too small".to_string(),
311                ));
312            }
313            break;
314        }
315        bytes = next;
316        page.push(payload.into_owned());
317    }
318    Ok(page)
319}
320
321pub(super) fn replication_record_hash(
322    previous_hash: &str,
323    sequence: u64,
324    payload: &[u8],
325) -> String {
326    let mut hasher = Sha256::new();
327    hasher.update(REPLICATION_LOG_FORMAT_V1.as_bytes());
328    hasher.update((previous_hash.len() as u64).to_be_bytes());
329    hasher.update(previous_hash.as_bytes());
330    hasher.update(sequence.to_be_bytes());
331    hasher.update((payload.len() as u64).to_be_bytes());
332    hasher.update(payload);
333    bytes_to_hex(&hasher.finalize())
334}
appcore_sync/sync/log.rs

appcore_sync/sync/
log.rs
