appcore_sync/sync/
receiver.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: receiver.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//! Follower receiver state and acknowledgements.
12
13use crate::sync::checkpoint::SyncCheckpointStore;
14use crate::sync::error::{SyncError, SyncResult};
15use crate::sync::log::{validate_record_size, ReplicationLog};
16use crate::sync::types::SyncMessage;
17use crate::sync::wire::SyncEnvelopeV1;
18use appcore_core::CoreIdentity;
19use parking_lot::Mutex;
20use std::sync::Arc;
21
22const DEFAULT_MAX_EVENTS: usize = 10_000;
23
24#[derive(Debug, Clone, Default)]
25struct ProcessedBatches {
26    set: std::collections::HashSet<String>,
27    queue: std::collections::VecDeque<String>,
28}
29
30/// In-memory receiver state for follower sync endpoint.
31#[derive(Clone)]
32pub struct SyncReceiverState {
33    replication_log: Arc<Mutex<Box<dyn ReplicationLog + Send>>>,
34    checkpoint_store: Arc<dyn SyncCheckpointStore>,
35    processed_batches: Arc<Mutex<ProcessedBatches>>,
36    local_identity: Option<CoreIdentity>,
37}
38
39/// Ack returned by `/v1/sync/events`.
40#[derive(Debug, Clone, PartialEq, Eq)]
41pub struct SyncReceiveAck {
42    /// Whether the batch passed validation and checkpoint rules.
43    pub accepted: bool,
44    /// Number of newly appended events.
45    pub received: usize,
46    /// Number of idempotently skipped events.
47    pub skipped: usize,
48    /// Final source sequence represented by the accepted batch.
49    pub last_sequence: u64,
50}
51
52impl SyncReceiverState {
53    /// Creates receiver state backed by a replication log and checkpoint store.
54    pub fn new(
55        replication_log: Arc<Mutex<Box<dyn ReplicationLog + Send>>>,
56        checkpoint_store: Arc<dyn SyncCheckpointStore>,
57    ) -> Self {
58        Self {
59            replication_log,
60            checkpoint_store,
61            processed_batches: Arc::new(Mutex::new(ProcessedBatches::default())),
62            local_identity: None,
63        }
64    }
65
66    /// Configures the distributed identity used to validate v1 sync envelopes.
67    pub fn with_local_identity(mut self, local_identity: CoreIdentity) -> Self {
68        self.local_identity = Some(local_identity);
69        self
70    }
71
72    /// Returns a shared handle to the receiver's replication log.
73    pub fn replication_log(&self) -> Arc<Mutex<Box<dyn ReplicationLog + Send>>> {
74        Arc::clone(&self.replication_log)
75    }
76
77    /// Validates and idempotently applies one already-authenticated batch.
78    pub fn apply_sync_message(&self, message: &SyncMessage) -> SyncResult<SyncReceiveAck> {
79        self.validate_message(message)?;
80
81        let peer_id = message.source_node_id.as_str();
82        let last_sequence = self.checkpoint_store.get_last_sequence(peer_id)?;
83
84        // Skip completely duplicate old sequence ranges
85        if message.sequence_end <= last_sequence {
86            return Ok(SyncReceiveAck {
87                accepted: true,
88                received: 0,
89                skipped: message.events.len(),
90                last_sequence,
91            });
92        }
93
94        // New data must cover the next sequence; verified overlap is allowed for
95        // recovery when the sender loses its outbound cursor after a successful send.
96        let next_sequence = last_sequence
97            .checked_add(1)
98            .ok_or(SyncError::InvalidSyncMessage(
99                "checkpoint sequence is exhausted",
100            ))?;
101        if message.sequence_start > next_sequence {
102            return Err(SyncError::InvalidSequence(message.sequence_start));
103        }
104
105        // Verify previous batch hash chain
106        if last_sequence > 0 && message.sequence_start == next_sequence {
107            if let Some((_, last_hash)) = self.checkpoint_store.get_checkpoint(peer_id)? {
108                if !last_hash.is_empty() {
109                    let prev_hash = message.previous_batch_hash.as_deref().unwrap_or("");
110                    if prev_hash != last_hash {
111                        return Err(SyncError::InvalidSyncMessage(
112                            "previous batch hash mismatch",
113                        ));
114                    }
115                }
116            }
117        }
118
119        let (received, skipped) = self.append_events_to_log(message, peer_id, last_sequence)?;
120        self.record_processed_batch(message.batch_id.clone());
121
122        Ok(SyncReceiveAck {
123            accepted: true,
124            received,
125            skipped,
126            last_sequence: message.sequence_end,
127        })
128    }
129
130    /// Validates a decoded wire envelope before applying its replication batch.
131    pub fn apply_sync_envelope(&self, envelope: &SyncEnvelopeV1) -> SyncResult<SyncReceiveAck> {
132        let local_identity = self
133            .local_identity
134            .as_ref()
135            .ok_or(SyncError::InvalidSyncMessage(
136                "local sync identity is not configured",
137            ))?;
138        envelope.validate_for(local_identity)?;
139        self.apply_sync_message(&envelope.message)
140    }
141
142    fn validate_message(&self, message: &SyncMessage) -> SyncResult<()> {
143        if message.sequence_start == 0 {
144            return Err(SyncError::InvalidSequence(message.sequence_start));
145        }
146        if message.events.is_empty() || message.event_count == 0 {
147            return Err(SyncError::InvalidSyncMessage("empty batch events"));
148        }
149        if message.event_count != message.events.len() {
150            return Err(SyncError::InvalidSyncMessage("event count mismatch"));
151        }
152        if message.events.len() > DEFAULT_MAX_EVENTS {
153            return Err(SyncError::TooManyEvents {
154                count: message.events.len(),
155                max: DEFAULT_MAX_EVENTS,
156            });
157        }
158        for event in &message.events {
159            validate_record_size(event)?;
160        }
161        if message.sequence_start > message.sequence_end {
162            return Err(SyncError::InvalidSyncMessage("invalid sequence range"));
163        }
164        let event_count = u64::try_from(message.event_count)
165            .map_err(|_| SyncError::InvalidSyncMessage("event count exceeds sequence range"))?;
166        let expected_end = message
167            .sequence_start
168            .checked_add(event_count.saturating_sub(1))
169            .ok_or(SyncError::InvalidSyncMessage("sequence range overflow"))?;
170        if message.sequence_end != expected_end {
171            return Err(SyncError::InvalidSyncMessage("inconsistent sequence range"));
172        }
173        let computed_hash = crate::sync::types::compute_events_hash(
174            &message.batch_id,
175            &message.source_node_id,
176            message.sequence_start,
177            message.sequence_end,
178            message.created_at_ms,
179            message.previous_batch_hash.as_deref(),
180            &message.events,
181        );
182        if computed_hash != message.events_hash {
183            return Err(SyncError::InvalidSyncMessage("invalid events hash"));
184        }
185        {
186            let processed = self.processed_batches.lock();
187            if processed.set.contains(&message.batch_id) {
188                return Err(SyncError::InvalidSyncMessage("duplicate batch_id"));
189            }
190        }
191        Ok(())
192    }
193
194    fn append_events_to_log(
195        &self,
196        message: &SyncMessage,
197        peer_id: &str,
198        last_sequence: u64,
199    ) -> SyncResult<(usize, usize)> {
200        let mut guard = self.replication_log.lock();
201        let mut received = 0usize;
202        let mut skipped = 0usize;
203        for (index, event) in message.events.iter().enumerate() {
204            let offset = u64::try_from(index)
205                .map_err(|_| SyncError::InvalidSyncMessage("event index exceeds sequence range"))?;
206            let seq = message
207                .sequence_start
208                .checked_add(offset)
209                .ok_or(SyncError::InvalidSyncMessage("sequence range overflow"))?;
210            match guard.event_at_sequence(seq)? {
211                Some(existing) if existing == *event => skipped += 1,
212                Some(_) => return Err(SyncError::SequenceConflict(seq)),
213                None if seq <= last_sequence => return Err(SyncError::InvalidSequence(seq)),
214                None => {
215                    received += 1;
216                    let _ = guard.append_with_sequence(event.clone(), seq)?;
217                }
218            }
219        }
220        self.checkpoint_store.set_checkpoint(
221            peer_id,
222            message.sequence_end,
223            &message.events_hash,
224        )?;
225        Ok((received, skipped))
226    }
227
228    fn record_processed_batch(&self, batch_id: String) {
229        let mut processed = self.processed_batches.lock();
230        processed.set.insert(batch_id.clone());
231        processed.queue.push_back(batch_id);
232        if processed.set.len() > 10_000 {
233            if let Some(oldest) = processed.queue.pop_front() {
234                processed.set.remove(&oldest);
235            }
236        }
237    }
238}
appcore_sync/sync/receiver.rs

appcore_sync/sync/
receiver.rs
