appcore_sync/sync/
log_file.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: log_file.rs
4//    ##   ## ##   ##    P: AppCore-Runtime
5//         ## ##
6//                       C: 2026/09/02 00:00:00 by dnettoRaw
7//    ##   ## ##   ##    U: 2026/09/02 00:00:00 by dnettoRaw
8//      ###########      S: 1.0.2-rc
9// =============================================================================
10
11//! File-backed replication log with incremental scans and offset-only indexes.
12
13use crate::sync::error::{SyncError, SyncResult};
14use crate::sync::log::{
15    replication_record_hash, validate_log_index, validate_page_limits, validate_record_count,
16    validate_record_size, ReplicationLog, MAX_REPLICATION_LOG_BYTES, MAX_REPLICATION_PAGE_BYTES,
17    MAX_REPLICATION_PAGE_RECORDS, REPLICATION_LOG_FORMAT_V1,
18};
19use crate::sync::log_file_format::{
20    anchor_matches, append_record, read_header, read_payload, read_payload_from, scan_tail,
21    write_record, RecordLocation, TailScan,
22};
23use crate::sync::persistence::{
24    acquire_persistence_lock, atomic_write_with, reject_symlink, truncate_synced,
25};
26use crate::sync::snapshot::{
27    snapshot_from_payloads, validate_snapshot_contract, ReplicationSnapshot,
28};
29use std::fs::{self, File};
30use std::io::Write;
31use std::path::{Path, PathBuf};
32
33/// File-backed append-only replication log for local Runtime sync.
34#[derive(Debug, Clone, PartialEq, Eq)]
35pub struct FileReplicationLog {
36    file_path: PathBuf,
37    state: LogState,
38}
39
40#[derive(Debug, Clone, PartialEq, Eq)]
41struct LogState {
42    records: Vec<RecordLocation>,
43    sequence_indices: Vec<(u64, usize)>,
44    scanned_bytes: u64,
45    record_count: usize,
46    line_count: usize,
47    chain_head: String,
48}
49
50impl FileReplicationLog {
51    /// Opens a relative append-only log below `storage_path`.
52    pub fn new(storage_path: impl AsRef<Path>, relative_path: &str) -> SyncResult<Self> {
53        let file_path = validated_path(storage_path.as_ref(), relative_path)?;
54        if let Some(parent) = file_path.parent() {
55            fs::create_dir_all(parent).map_err(replication_error)?;
56        }
57        let _process_lock = acquire_persistence_lock(&file_path)?;
58        if !file_path.exists() {
59            create_empty_log(&file_path)?;
60        }
61        let state = load_state(&file_path)?;
62        Ok(Self { file_path, state })
63    }
64
65    /// Returns the durable log file path.
66    pub fn file_path(&self) -> &Path {
67        &self.file_path
68    }
69
70    /// Re-reads and validates all durable records from disk.
71    pub fn reload(&mut self) -> SyncResult<()> {
72        let _process_lock = acquire_persistence_lock(&self.file_path)?;
73        self.state = load_state(&self.file_path)?;
74        Ok(())
75    }
76
77    /// Reads one page after `index`, bounded before payload allocation.
78    pub fn events_page(
79        &self,
80        index: usize,
81        max_records: usize,
82        max_bytes: usize,
83    ) -> SyncResult<Vec<Vec<u8>>> {
84        validate_log_index(index, self.state.records.len())?;
85        validate_page_limits(max_records, max_bytes)?;
86        read_page(
87            &self.file_path,
88            &self.state.records[index..],
89            max_records,
90            max_bytes,
91        )
92    }
93
94    fn refresh_unlocked(&mut self) -> SyncResult<()> {
95        let header = read_header(&self.file_path)?;
96        if header.incomplete || header.file_bytes < self.state.scanned_bytes {
97            self.state = load_state(&self.file_path)?;
98            return Ok(());
99        }
100        if !anchor_matches(
101            &self.file_path,
102            self.state.records.last(),
103            &self.state.chain_head,
104        )? {
105            self.state = load_state(&self.file_path)?;
106            return Ok(());
107        }
108        if header.file_bytes == self.state.scanned_bytes {
109            return Ok(());
110        }
111        let tail = scan_tail(
112            &self.file_path,
113            self.state.scanned_bytes,
114            self.state.record_count,
115            self.state.line_count,
116            &self.state.chain_head,
117            &self.state.records,
118            &self.state.sequence_indices,
119        )?;
120        apply_tail(&mut self.state, tail, &self.file_path)
121    }
122
123    fn append_record(&mut self, payload: Vec<u8>, sequence: u64) -> SyncResult<usize> {
124        let _process_lock = acquire_persistence_lock(&self.file_path)?;
125        self.refresh_unlocked()?;
126        validate_record_size(&payload)?;
127        validate_record_count(self.state.record_count.saturating_add(1))?;
128        if sequence > 0 {
129            if let Some(index) = find_sequence(&self.state.sequence_indices, sequence) {
130                return if read_payload(&self.file_path, &self.state.records[index])? == payload {
131                    Ok(index + 1)
132                } else {
133                    Err(SyncError::SequenceConflict(sequence))
134                };
135            }
136        }
137        let previous_hash = self.state.chain_head.clone();
138        let record_hash = replication_record_hash(&previous_hash, sequence, &payload);
139        let (location, next_offset) = append_record(
140            &self.file_path,
141            self.state.scanned_bytes,
142            sequence,
143            &payload,
144            &previous_hash,
145            &record_hash,
146        )?;
147        let index = self.state.records.len();
148        self.state.records.push(location);
149        if sequence > 0 {
150            insert_sequence(&mut self.state.sequence_indices, sequence, index);
151        }
152        self.state.scanned_bytes = next_offset;
153        self.state.record_count += 1;
154        self.state.line_count += 1;
155        self.state.chain_head = record_hash;
156        Ok(index + 1)
157    }
158}
159
160impl ReplicationLog for FileReplicationLog {
161    fn append(&mut self, record: Vec<u8>) -> SyncResult<usize> {
162        self.append_record(record, 0)
163    }
164
165    fn append_with_sequence(&mut self, record: Vec<u8>, sequence: u64) -> SyncResult<usize> {
166        self.append_record(record, sequence)
167    }
168
169    fn event_at_sequence(&self, sequence: u64) -> SyncResult<Option<Vec<u8>>> {
170        if sequence == 0 {
171            return Ok(None);
172        }
173        find_sequence(&self.state.sequence_indices, sequence)
174            .map(|index| read_payload(&self.file_path, &self.state.records[index]))
175            .transpose()
176    }
177
178    fn events_since(&self, index: usize) -> SyncResult<Vec<Vec<u8>>> {
179        validate_log_index(index, self.state.records.len())?;
180        let records = &self.state.records[index..];
181        if records.len() > MAX_REPLICATION_PAGE_RECORDS
182            || aggregate_payload_bytes(records)? > MAX_REPLICATION_PAGE_BYTES
183        {
184            return Err(replication_message(
185                "complete replication read exceeds page limits; use events_page",
186            ));
187        }
188        read_page(
189            &self.file_path,
190            records,
191            MAX_REPLICATION_PAGE_RECORDS,
192            MAX_REPLICATION_PAGE_BYTES,
193        )
194    }
195
196    fn events_page(
197        &self,
198        index: usize,
199        max_records: usize,
200        max_bytes: usize,
201    ) -> SyncResult<Vec<Vec<u8>>> {
202        Self::events_page(self, index, max_records, max_bytes)
203    }
204
205    fn last_index(&self) -> SyncResult<usize> {
206        Ok(self.state.records.len())
207    }
208
209    fn len(&self) -> SyncResult<usize> {
210        Ok(self.state.records.len())
211    }
212
213    fn is_empty(&self) -> SyncResult<bool> {
214        Ok(self.state.records.is_empty())
215    }
216
217    fn create_snapshot(&self) -> SyncResult<ReplicationSnapshot> {
218        let mut file = open_log(&self.file_path)?;
219        snapshot_from_payloads(self.state.records.iter().map(|location| {
220            read_payload_from(&mut file, location).map(|payload| (location.sequence, payload))
221        }))
222    }
223
224    fn restore_snapshot(&mut self, snapshot: &ReplicationSnapshot) -> SyncResult<()> {
225        let _process_lock = acquire_persistence_lock(&self.file_path)?;
226        validate_snapshot_contract(snapshot)?;
227        write_snapshot(&self.file_path, snapshot)?;
228        self.state = load_state(&self.file_path)?;
229        Ok(())
230    }
231}
232
233fn load_state(path: &Path) -> SyncResult<LogState> {
234    let header = read_header(path)?;
235    if header.incomplete {
236        create_empty_log(path)?;
237        return Ok(empty_state());
238    }
239    let tail = scan_tail(path, header.body_offset, 0, 0, "", &[], &[])?;
240    let mut state = empty_state();
241    apply_tail(&mut state, tail, path)?;
242    Ok(state)
243}
244
245fn apply_tail(state: &mut LogState, tail: TailScan, path: &Path) -> SyncResult<()> {
246    let start = state.records.len();
247    for (offset, location) in tail.locations.iter().enumerate() {
248        if location.sequence > 0 {
249            insert_sequence(
250                &mut state.sequence_indices,
251                location.sequence,
252                start + offset,
253            );
254        }
255    }
256    state.records.extend(tail.locations);
257    state.scanned_bytes = tail.scanned_bytes;
258    state.record_count = tail.record_count;
259    state.line_count = tail.line_count;
260    state.chain_head = tail.chain_head;
261    if tail.recovered_tail {
262        truncate_synced(path, state.scanned_bytes)?;
263    }
264    Ok(())
265}
266
267fn read_page(
268    path: &Path,
269    records: &[RecordLocation],
270    max_records: usize,
271    max_bytes: usize,
272) -> SyncResult<Vec<Vec<u8>>> {
273    let mut file = open_log(path)?;
274    let mut page = Vec::with_capacity(records.len().min(max_records));
275    let mut bytes = 0usize;
276    for location in records.iter().take(max_records) {
277        let next = bytes
278            .checked_add(location.payload_bytes as usize)
279            .ok_or_else(|| replication_message("replication page overflow"))?;
280        if next > max_bytes {
281            if page.is_empty() {
282                return Err(replication_message("replication page byte limit too small"));
283            }
284            break;
285        }
286        page.push(read_payload_from(&mut file, location)?);
287        bytes = next;
288    }
289    Ok(page)
290}
291
292fn aggregate_payload_bytes(records: &[RecordLocation]) -> SyncResult<usize> {
293    records.iter().try_fold(0usize, |total, location| {
294        total
295            .checked_add(location.payload_bytes as usize)
296            .ok_or_else(|| replication_message("replication payload byte count overflow"))
297    })
298}
299
300fn write_snapshot(path: &Path, snapshot: &ReplicationSnapshot) -> SyncResult<()> {
301    atomic_write_with(path, |file| {
302        write_header(file)?;
303        let mut offset = (REPLICATION_LOG_FORMAT_V1.len() + 1) as u64;
304        let mut previous_hash = String::new();
305        for record in &snapshot.records {
306            let hash = replication_record_hash(&previous_hash, record.sequence, &record.payload);
307            let (_, written_bytes) = write_record(
308                file,
309                offset,
310                record.sequence,
311                &record.payload,
312                &previous_hash,
313                &hash,
314            )?;
315            offset = offset
316                .checked_add(written_bytes)
317                .filter(|bytes| *bytes <= MAX_REPLICATION_LOG_BYTES)
318                .ok_or_else(|| replication_message("replication log exceeds configured limit"))?;
319            previous_hash = hash;
320        }
321        Ok(())
322    })
323}
324
325fn create_empty_log(path: &Path) -> SyncResult<()> {
326    atomic_write_with(path, write_header)
327}
328
329fn write_header(file: &mut File) -> SyncResult<()> {
330    file.write_all(REPLICATION_LOG_FORMAT_V1.as_bytes())
331        .and_then(|()| file.write_all(b"\n"))
332        .map_err(replication_error)
333}
334
335fn empty_state() -> LogState {
336    LogState {
337        records: Vec::new(),
338        sequence_indices: Vec::new(),
339        scanned_bytes: (REPLICATION_LOG_FORMAT_V1.len() + 1) as u64,
340        record_count: 0,
341        line_count: 0,
342        chain_head: String::new(),
343    }
344}
345
346fn find_sequence(index: &[(u64, usize)], sequence: u64) -> Option<usize> {
347    index
348        .binary_search_by_key(&sequence, |(key, _)| *key)
349        .ok()
350        .map(|position| index[position].1)
351}
352
353fn insert_sequence(index: &mut Vec<(u64, usize)>, sequence: u64, record_index: usize) {
354    match index.binary_search_by_key(&sequence, |(key, _)| *key) {
355        Ok(position) => index[position] = (sequence, record_index),
356        Err(position) => index.insert(position, (sequence, record_index)),
357    }
358}
359
360fn validated_path(storage_root: &Path, relative_path: &str) -> SyncResult<PathBuf> {
361    let relative = PathBuf::from(relative_path);
362    if relative.as_os_str().is_empty()
363        || relative.is_absolute()
364        || relative
365            .components()
366            .any(|component| matches!(component, std::path::Component::ParentDir))
367    {
368        return Err(replication_message("invalid replication log path"));
369    }
370    Ok(storage_root.join(relative))
371}
372
373fn open_log(path: &Path) -> SyncResult<File> {
374    reject_symlink(path)?;
375    File::open(path).map_err(replication_error)
376}
377
378fn replication_message(message: &str) -> SyncError {
379    SyncError::ReplicationFailed(message.to_string())
380}
381
382fn replication_error(error: std::io::Error) -> SyncError {
383    SyncError::ReplicationFailed(error.to_string())
384}
appcore_sync/sync/log_file.rs

appcore_sync/sync/
log_file.rs
