appcore_sync/sync/
outbox_journal.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: outbox_journal.rs
4//    ##   ## ##   ##    P: AppCore-Runtime
5//         ## ##
6//                       C: 2026/08/26 00:00:00 by dnettoRaw
7//    ##   ## ##   ##    U: 2026/08/26 00:00:00 by dnettoRaw
8//      ###########      S: 2.0.0
9// =============================================================================
10
11//! Incremental binary journal for the durable synchronization outbox.
12
13use crate::sync::error::{SyncError, SyncResult};
14use crate::sync::outbox::{validate_page_limits, SyncOutbox, SyncOutboxReceipt, SyncOutboxStats};
15use crate::sync::outbox_format::{
16    append_record, corrupt_record, create_empty_journal, encode_attempt, encoded_frame_bytes,
17    ensure_append_capacity, new_generation, outbox_full, read_header, record_hash, scan_records,
18    validate_batch_id, validate_record_data, write_frame, write_header, JournalOperation,
19    ScanPending, ScanResult, ACK_SPACE_RESERVE_BYTES, ATTEMPT_KIND, COMPACTION_ACK_RECORDS,
20    COMPACTION_RECLAIM_BYTES, ENQUEUE_KIND, GENERATION_BYTES, HASH_BYTES, HEADER_BYTES,
21    MAX_OUTBOX_FILE_BYTES, RECEIPT_KIND,
22};
23use crate::sync::outbox_journal_view::{page, stats, validate_receipt_prefix};
24use crate::sync::persistence::{acquire_persistence_lock, atomic_write_with, truncate_synced};
25use crate::sync::types::SyncMessage;
26use parking_lot::Mutex;
27use std::collections::VecDeque;
28use std::path::{Path, PathBuf};
29
30/// Stable on-disk format marker for incremental durable sync outboxes.
31pub use crate::sync::outbox_format::SYNC_OUTBOX_FORMAT_V2;
32
33/// Crash-consistent incremental file-backed synchronization outbox.
34pub struct FileSyncOutbox {
35    file_path: PathBuf,
36    state: Mutex<JournalState>,
37}
38
39struct JournalState {
40    generation: [u8; GENERATION_BYTES],
41    messages: VecDeque<PendingMessage>,
42    scanned_bytes: u64,
43    record_count: u64,
44    acknowledged_records: u64,
45    live_frame_bytes: u64,
46    chain_head: [u8; HASH_BYTES],
47}
48
49pub(super) struct PendingMessage {
50    pub(super) message: SyncMessage,
51    pub(super) encoded_bytes: usize,
52    pub(super) frame_bytes: u64,
53    pub(super) attempt_frame_bytes: u64,
54    pub(super) attempts: u32,
55    pub(super) next_ready_at_ms: u64,
56}
57
58impl std::fmt::Debug for FileSyncOutbox {
59    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
60        formatter
61            .debug_struct("FileSyncOutbox")
62            .field("file_path", &self.file_path)
63            .field("pending_messages", &self.state.lock().messages.len())
64            .finish()
65    }
66}
67
68impl FileSyncOutbox {
69    /// Opens or creates a V2 outbox and validates every complete journal frame.
70    pub fn new(file_path: impl Into<PathBuf>) -> SyncResult<Self> {
71        let file_path = file_path.into();
72        if let Some(parent) = file_path.parent() {
73            std::fs::create_dir_all(parent)
74                .map_err(|error| SyncError::ReplicationFailed(error.to_string()))?;
75        }
76        let _process_lock = acquire_persistence_lock(&file_path)?;
77        if !file_path.exists() {
78            create_empty_journal(&file_path)?;
79        }
80        let state = load_state(&file_path)?;
81        Ok(Self {
82            file_path,
83            state: Mutex::new(state),
84        })
85    }
86
87    /// Returns the durable outbox journal path.
88    pub fn file_path(&self) -> &Path {
89        &self.file_path
90    }
91
92    fn refresh(&self, state: &mut JournalState) -> SyncResult<()> {
93        let header = read_header(&self.file_path)?;
94        if header.generation != state.generation || header.file_bytes < state.scanned_bytes {
95            *state = load_state(&self.file_path)?;
96            return Ok(());
97        }
98        if header.file_bytes == state.scanned_bytes {
99            return Ok(());
100        }
101        let pending = state
102            .messages
103            .iter()
104            .map(|pending| ScanPending::new(pending.message.batch_id.clone(), pending.attempts))
105            .collect();
106        let scan = scan_records(
107            &self.file_path,
108            state.scanned_bytes,
109            state.record_count,
110            state.chain_head,
111            pending,
112        )?;
113        apply_scan(state, scan, &self.file_path)
114    }
115
116    fn compact_if_needed(&self, state: &mut JournalState) -> SyncResult<()> {
117        let reclaimable = state
118            .scanned_bytes
119            .saturating_sub(HEADER_BYTES as u64)
120            .saturating_sub(state.live_frame_bytes);
121        if reclaimable >= COMPACTION_RECLAIM_BYTES
122            || state.acknowledged_records >= COMPACTION_ACK_RECORDS
123        {
124            compact(&self.file_path, state)?;
125        }
126        Ok(())
127    }
128}
129
130impl SyncOutbox for FileSyncOutbox {
131    fn try_enqueue(&self, message: SyncMessage, max_len: usize) -> SyncResult<bool> {
132        validate_batch_id(&message.batch_id)?;
133        let mut state = self.state.lock();
134        let _process_lock = acquire_persistence_lock(&self.file_path)?;
135        self.refresh(&mut state)?;
136        self.compact_if_needed(&mut state)?;
137        if state.messages.len() >= max_len {
138            return Ok(false);
139        }
140        let data = serde_json::to_vec(&message)
141            .map_err(|error| SyncError::ReplicationFailed(error.to_string()))?;
142        validate_record_data(&data)?;
143        let frame_bytes = encoded_frame_bytes(data.len())?;
144        if state
145            .scanned_bytes
146            .checked_add(frame_bytes)
147            .is_none_or(|size| size > MAX_OUTBOX_FILE_BYTES.saturating_sub(ACK_SPACE_RESERVE_BYTES))
148        {
149            compact(&self.file_path, &mut state)?;
150        }
151        ensure_append_capacity(state.scanned_bytes, frame_bytes, ACK_SPACE_RESERVE_BYTES)?;
152        let hash = append_record(
153            &self.file_path,
154            state.record_count,
155            state.chain_head,
156            ENQUEUE_KIND,
157            &data,
158        )?;
159        state.scanned_bytes += frame_bytes;
160        state.record_count += 1;
161        state.live_frame_bytes += frame_bytes;
162        state.chain_head = hash;
163        state.messages.push_back(PendingMessage {
164            message,
165            encoded_bytes: data.len(),
166            frame_bytes,
167            attempt_frame_bytes: 0,
168            attempts: 0,
169            next_ready_at_ms: 0,
170        });
171        Ok(true)
172    }
173
174    fn front(&self) -> SyncResult<Option<SyncMessage>> {
175        let mut state = self.state.lock();
176        let _process_lock = acquire_persistence_lock(&self.file_path)?;
177        self.refresh(&mut state)?;
178        Ok(state
179            .messages
180            .front()
181            .map(|pending| pending.message.clone()))
182    }
183
184    fn acknowledge_front(&self, batch_id: &str) -> SyncResult<()> {
185        let receipt = SyncOutboxReceipt::new(vec![batch_id.to_string()])?;
186        self.acknowledge_receipt(&receipt).map(|_| ())
187    }
188
189    fn messages(&self) -> SyncResult<Vec<SyncMessage>> {
190        let mut state = self.state.lock();
191        let _process_lock = acquire_persistence_lock(&self.file_path)?;
192        self.refresh(&mut state)?;
193        Ok(state
194            .messages
195            .iter()
196            .map(|pending| pending.message.clone())
197            .collect())
198    }
199
200    fn len(&self) -> SyncResult<usize> {
201        let mut state = self.state.lock();
202        let _process_lock = acquire_persistence_lock(&self.file_path)?;
203        self.refresh(&mut state)?;
204        Ok(state.messages.len())
205    }
206
207    fn peek(&self, limit: usize, max_bytes: usize) -> SyncResult<Vec<SyncMessage>> {
208        validate_page_limits(limit, max_bytes)?;
209        let mut state = self.state.lock();
210        let _process_lock = acquire_persistence_lock(&self.file_path)?;
211        self.refresh(&mut state)?;
212        Ok(page(&state.messages, limit, max_bytes, None))
213    }
214
215    fn stats(&self) -> SyncResult<SyncOutboxStats> {
216        let mut state = self.state.lock();
217        let _process_lock = acquire_persistence_lock(&self.file_path)?;
218        self.refresh(&mut state)?;
219        stats(&state.messages)
220    }
221
222    fn mark_attempt(&self, batch_id: &str, next_ready_at_ms: u64) -> SyncResult<u32> {
223        validate_batch_id(batch_id)?;
224        let mut state = self.state.lock();
225        let _process_lock = acquire_persistence_lock(&self.file_path)?;
226        self.refresh(&mut state)?;
227        self.compact_if_needed(&mut state)?;
228        let pending = state
229            .messages
230            .front()
231            .filter(|pending| pending.message.batch_id == batch_id)
232            .ok_or(SyncError::InvalidSyncMessage("outbox attempt mismatch"))?;
233        let attempts = pending
234            .attempts
235            .checked_add(1)
236            .ok_or(SyncError::InvalidSyncMessage("outbox attempt overflow"))?;
237        let previous_attempt_bytes = pending.attempt_frame_bytes;
238        let data = encode_attempt(batch_id, attempts, next_ready_at_ms)?;
239        let frame_bytes = encoded_frame_bytes(data.len())?;
240        ensure_append_capacity(state.scanned_bytes, frame_bytes, ACK_SPACE_RESERVE_BYTES)?;
241        let hash = append_record(
242            &self.file_path,
243            state.record_count,
244            state.chain_head,
245            ATTEMPT_KIND,
246            &data,
247        )?;
248        let pending = state
249            .messages
250            .front_mut()
251            .ok_or(SyncError::InvalidSyncMessage("outbox attempt mismatch"))?;
252        pending.attempts = attempts;
253        pending.next_ready_at_ms = next_ready_at_ms;
254        pending.attempt_frame_bytes = frame_bytes;
255        state.scanned_bytes += frame_bytes;
256        state.record_count += 1;
257        state.live_frame_bytes = state
258            .live_frame_bytes
259            .saturating_sub(previous_attempt_bytes)
260            .saturating_add(frame_bytes);
261        state.chain_head = hash;
262        Ok(attempts)
263    }
264
265    fn next_ready(
266        &self,
267        now_ms: u64,
268        limit: usize,
269        max_bytes: usize,
270    ) -> SyncResult<Vec<SyncMessage>> {
271        validate_page_limits(limit, max_bytes)?;
272        let mut state = self.state.lock();
273        let _process_lock = acquire_persistence_lock(&self.file_path)?;
274        self.refresh(&mut state)?;
275        Ok(page(&state.messages, limit, max_bytes, Some(now_ms)))
276    }
277
278    fn acknowledge_receipt(&self, receipt: &SyncOutboxReceipt) -> SyncResult<usize> {
279        let mut state = self.state.lock();
280        let _process_lock = acquire_persistence_lock(&self.file_path)?;
281        self.refresh(&mut state)?;
282        self.compact_if_needed(&mut state)?;
283        validate_receipt_prefix(&state.messages, receipt)?;
284        let data = serde_json::to_vec(receipt.batch_ids())
285            .map_err(|_| SyncError::InvalidSyncMessage("outbox receipt serialization"))?;
286        validate_record_data(&data)?;
287        let frame_bytes = encoded_frame_bytes(data.len())?;
288        ensure_append_capacity(state.scanned_bytes, frame_bytes, 0)?;
289        let hash = append_record(
290            &self.file_path,
291            state.record_count,
292            state.chain_head,
293            RECEIPT_KIND,
294            &data,
295        )?;
296        let mut removed_live_bytes = 0u64;
297        for _ in receipt.batch_ids() {
298            let pending = state
299                .messages
300                .pop_front()
301                .ok_or(SyncError::InvalidSyncMessage(
302                    "outbox acknowledgement mismatch",
303                ))?;
304            removed_live_bytes = removed_live_bytes
305                .saturating_add(pending.frame_bytes)
306                .saturating_add(pending.attempt_frame_bytes);
307        }
308        state.scanned_bytes += frame_bytes;
309        state.record_count += 1;
310        state.acknowledged_records = state
311            .acknowledged_records
312            .saturating_add(receipt.batch_ids().len() as u64);
313        state.live_frame_bytes = state.live_frame_bytes.saturating_sub(removed_live_bytes);
314        state.chain_head = hash;
315        Ok(receipt.batch_ids().len())
316    }
317}
318
319fn load_state(path: &Path) -> SyncResult<JournalState> {
320    let header = read_header(path)?;
321    let scan = scan_records(
322        path,
323        HEADER_BYTES as u64,
324        0,
325        [0; HASH_BYTES],
326        VecDeque::new(),
327    )?;
328    let mut state = JournalState {
329        generation: header.generation,
330        messages: VecDeque::new(),
331        scanned_bytes: HEADER_BYTES as u64,
332        record_count: 0,
333        acknowledged_records: 0,
334        live_frame_bytes: 0,
335        chain_head: [0; HASH_BYTES],
336    };
337    apply_scan(&mut state, scan, path)?;
338    Ok(state)
339}
340
341fn apply_scan(state: &mut JournalState, scan: ScanResult, path: &Path) -> SyncResult<()> {
342    if scan.recovered_tail {
343        truncate_synced(path, scan.scanned_bytes)?;
344    }
345    for operation in scan.operations {
346        match operation {
347            JournalOperation::Enqueue {
348                message,
349                encoded_bytes,
350                frame_bytes,
351            } => {
352                state.live_frame_bytes += frame_bytes;
353                state.messages.push_back(PendingMessage {
354                    message,
355                    encoded_bytes,
356                    frame_bytes,
357                    attempt_frame_bytes: 0,
358                    attempts: 0,
359                    next_ready_at_ms: 0,
360                });
361            }
362            JournalOperation::Acknowledge { count } => {
363                for _ in 0..count {
364                    let acknowledged = state
365                        .messages
366                        .pop_front()
367                        .ok_or_else(|| corrupt_record(state.record_count))?;
368                    state.live_frame_bytes = state
369                        .live_frame_bytes
370                        .saturating_sub(acknowledged.frame_bytes)
371                        .saturating_sub(acknowledged.attempt_frame_bytes);
372                }
373                state.acknowledged_records =
374                    state.acknowledged_records.saturating_add(count as u64);
375            }
376            JournalOperation::Attempt {
377                attempts,
378                next_ready_at_ms,
379                frame_bytes,
380            } => {
381                let pending = state
382                    .messages
383                    .front_mut()
384                    .ok_or_else(|| corrupt_record(state.record_count))?;
385                state.live_frame_bytes = state
386                    .live_frame_bytes
387                    .saturating_sub(pending.attempt_frame_bytes)
388                    .saturating_add(frame_bytes);
389                pending.attempt_frame_bytes = frame_bytes;
390                pending.attempts = attempts;
391                pending.next_ready_at_ms = next_ready_at_ms;
392            }
393        }
394    }
395    state.scanned_bytes = scan.scanned_bytes;
396    state.record_count = scan.record_count;
397    state.chain_head = scan.chain_head;
398    Ok(())
399}
400
401fn compact(path: &Path, state: &mut JournalState) -> SyncResult<()> {
402    let generation = new_generation();
403    let mut frame_sizes = Vec::with_capacity(state.messages.len());
404    let mut chain_head = [0; HASH_BYTES];
405    let mut record_count = 0u64;
406    let mut total_bytes = HEADER_BYTES as u64;
407    atomic_write_with(path, |file| {
408        write_header(file, generation)?;
409        for pending in &state.messages {
410            let data = serde_json::to_vec(&pending.message)
411                .map_err(|error| SyncError::ReplicationFailed(error.to_string()))?;
412            validate_record_data(&data)?;
413            record_count += 1;
414            let hash = record_hash(record_count, ENQUEUE_KIND, &data, chain_head);
415            write_frame(file, record_count, ENQUEUE_KIND, &data, chain_head, hash)?;
416            chain_head = hash;
417            let size = encoded_frame_bytes(data.len())?;
418            total_bytes = total_bytes.checked_add(size).ok_or_else(outbox_full)?;
419            if total_bytes > MAX_OUTBOX_FILE_BYTES {
420                return Err(outbox_full());
421            }
422            let mut attempt_size = 0;
423            if pending.attempts > 0 {
424                let attempt = encode_attempt(
425                    &pending.message.batch_id,
426                    pending.attempts,
427                    pending.next_ready_at_ms,
428                )?;
429                record_count += 1;
430                let hash = record_hash(record_count, ATTEMPT_KIND, &attempt, chain_head);
431                write_frame(file, record_count, ATTEMPT_KIND, &attempt, chain_head, hash)?;
432                chain_head = hash;
433                attempt_size = encoded_frame_bytes(attempt.len())?;
434                total_bytes = total_bytes
435                    .checked_add(attempt_size)
436                    .ok_or_else(outbox_full)?;
437                if total_bytes > MAX_OUTBOX_FILE_BYTES {
438                    return Err(outbox_full());
439                }
440            }
441            frame_sizes.push((data.len(), size, attempt_size));
442        }
443        Ok(())
444    })?;
445    for (pending, (encoded_bytes, frame_bytes, attempt_frame_bytes)) in
446        state.messages.iter_mut().zip(frame_sizes)
447    {
448        pending.encoded_bytes = encoded_bytes;
449        pending.frame_bytes = frame_bytes;
450        pending.attempt_frame_bytes = attempt_frame_bytes;
451    }
452    state.generation = generation;
453    state.scanned_bytes = total_bytes;
454    state.record_count = record_count;
455    state.acknowledged_records = 0;
456    state.live_frame_bytes = total_bytes.saturating_sub(HEADER_BYTES as u64);
457    state.chain_head = chain_head;
458    Ok(())
459}
appcore_sync/sync/outbox_journal.rs

appcore_sync/sync/
outbox_journal.rs
