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

appcore_sync/sync/
receiver.rs
