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::HashSet<Arc<str>>,
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        let peer_id = message.source_node_id.as_str();
83        let last_sequence = self.checkpoint_store.get_last_sequence(peer_id)?;
84
85        // Skip completely duplicate old sequence ranges
86        if message.sequence_end <= last_sequence {
87            return Ok(SyncReceiveAck {
88                accepted: true,
89                received: 0,
90                skipped: message.events.len(),
91                last_sequence,
92            });
93        }
94
95        // New data must cover the next sequence; verified overlap is allowed for
96        // recovery when the sender loses its outbound cursor after a successful send.
97        let next_sequence = last_sequence
98            .checked_add(1)
99            .ok_or(SyncError::InvalidSyncMessage(
100                "checkpoint sequence is exhausted",
101            ))?;
102        if message.sequence_start > next_sequence {
103            return Err(SyncError::InvalidSequence(message.sequence_start));
104        }
105
106        // Verify previous batch hash chain
107        if last_sequence > 0 && message.sequence_start == next_sequence {
108            if let Some((_, last_hash)) = self.checkpoint_store.get_checkpoint(peer_id)? {
109                if !last_hash.is_empty() {
110                    let prev_hash = message.previous_batch_hash.as_deref().unwrap_or("");
111                    if prev_hash != last_hash {
112                        return Err(SyncError::InvalidSyncMessage(
113                            "previous batch hash mismatch",
114                        ));
115                    }
116                }
117            }
118        }
119
120        let (received, skipped) = self.append_events_to_log(message, peer_id, last_sequence)?;
121        self.record_processed_batch(&message.batch_id);
122
123        Ok(SyncReceiveAck {
124            accepted: true,
125            received,
126            skipped,
127            last_sequence: message.sequence_end,
128        })
129    }
130
131    /// Validates a decoded wire envelope before applying its replication batch.
132    pub fn apply_sync_envelope(&self, envelope: &SyncEnvelopeV1) -> SyncResult<SyncReceiveAck> {
133        let local_identity = self
134            .local_identity
135            .as_ref()
136            .ok_or(SyncError::InvalidSyncMessage(
137                "local sync identity is not configured",
138            ))?;
139        envelope.validate_for(local_identity)?;
140        self.apply_sync_message(&envelope.message)
141    }
142
143    fn validate_message(&self, message: &SyncMessage) -> SyncResult<()> {
144        if !is_valid_sync_batch_id(&message.batch_id) {
145            return Err(SyncError::InvalidSyncMessage("invalid batch_id"));
146        }
147        if message.sequence_start == 0 {
148            return Err(SyncError::InvalidSequence(message.sequence_start));
149        }
150        if message.events.is_empty() || message.event_count == 0 {
151            return Err(SyncError::InvalidSyncMessage("empty batch events"));
152        }
153        if message.event_count != message.events.len() {
154            return Err(SyncError::InvalidSyncMessage("event count mismatch"));
155        }
156        if message.events.len() > DEFAULT_MAX_EVENTS {
157            return Err(SyncError::TooManyEvents {
158                count: message.events.len(),
159                max: DEFAULT_MAX_EVENTS,
160            });
161        }
162        for event in &message.events {
163            validate_record_size(event)?;
164        }
165        if message.sequence_start > message.sequence_end {
166            return Err(SyncError::InvalidSyncMessage("invalid sequence range"));
167        }
168        let event_count = u64::try_from(message.event_count)
169            .map_err(|_| SyncError::InvalidSyncMessage("event count exceeds sequence range"))?;
170        let expected_end = message
171            .sequence_start
172            .checked_add(event_count.saturating_sub(1))
173            .ok_or(SyncError::InvalidSyncMessage("sequence range overflow"))?;
174        if message.sequence_end != expected_end {
175            return Err(SyncError::InvalidSyncMessage("inconsistent sequence range"));
176        }
177        let computed_hash = crate::sync::types::compute_events_hash(
178            &message.batch_id,
179            &message.source_node_id,
180            message.sequence_start,
181            message.sequence_end,
182            message.created_at_ms,
183            message.previous_batch_hash.as_deref(),
184            &message.events,
185        );
186        if computed_hash != message.events_hash {
187            return Err(SyncError::InvalidSyncMessage("invalid events hash"));
188        }
189        {
190            let processed = self.processed_batches.lock();
191            if processed.set.contains(message.batch_id.as_str()) {
192                return Err(SyncError::InvalidSyncMessage("duplicate batch_id"));
193            }
194        }
195        Ok(())
196    }
197
198    fn append_events_to_log(
199        &self,
200        message: &SyncMessage,
201        peer_id: &str,
202        last_sequence: u64,
203    ) -> SyncResult<(usize, usize)> {
204        let mut guard = self.replication_log.lock();
205        let mut received = 0usize;
206        let mut skipped = 0usize;
207        for (index, event) in message.events.iter().enumerate() {
208            let offset = u64::try_from(index)
209                .map_err(|_| SyncError::InvalidSyncMessage("event index exceeds sequence range"))?;
210            let seq = message
211                .sequence_start
212                .checked_add(offset)
213                .ok_or(SyncError::InvalidSyncMessage("sequence range overflow"))?;
214            match guard.event_at_sequence(seq)? {
215                Some(existing) if existing == *event => skipped += 1,
216                Some(_) => return Err(SyncError::SequenceConflict(seq)),
217                None if seq <= last_sequence => return Err(SyncError::InvalidSequence(seq)),
218                None => {
219                    received += 1;
220                    let _ = guard.append_with_sequence(event.clone(), seq)?;
221                }
222            }
223        }
224        self.checkpoint_store.set_checkpoint(
225            peer_id,
226            message.sequence_end,
227            &message.events_hash,
228        )?;
229        Ok((received, skipped))
230    }
231
232    fn record_processed_batch(&self, batch_id: &str) {
233        let mut processed = self.processed_batches.lock();
234        let shared: Arc<str> = Arc::from(batch_id);
235        processed.set.insert(Arc::clone(&shared));
236        processed.queue.push_back(shared);
237        if processed.set.len() > PROCESSED_BATCH_CAPACITY {
238            if let Some(oldest) = processed.queue.pop_front() {
239                processed.set.remove(oldest.as_ref());
240            }
241        }
242    }
243}
appcore_sync/sync/receiver.rs

appcore_sync/sync/
receiver.rs
