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

appcore_sync/sync/
outbox_journal.rs
