softchat/
sync.rs

1//! Bounded host-driven Negentropy synchronization coordinator.
2
3use std::collections::BTreeSet;
4
5use serde::{Deserialize, Serialize};
6
7use crate::{MAX_NEGENTROPY_ITEMS, NegentropyClient, NegentropyItem, NostrEventId, SoftchatError};
8
9/// Default relay/local snapshot page size.
10pub const SYNC_BASE_PAGE_LIMIT: u32 = 900;
11/// Maximum page expansion used to capture a complete timestamp cohort.
12pub const SYNC_MAX_PAGE_LIMIT: u32 = 100_000;
13/// Maximum IDs in one ordinary NIP-01 fetch or resend request.
14pub const SYNC_EVENT_REQUEST_CHUNK: usize = 500;
15
16/// Durable synchronization phase.
17#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
18#[serde(rename_all = "camelCase")]
19#[cfg_attr(feature = "native-bindings", derive(uniffi::Enum))]
20pub enum SyncPhase {
21    /// Created but not started.
22    Idle,
23    /// Waiting for a bounded platform database snapshot.
24    LoadingSnapshot,
25    /// NEG-OPEN/NEG-MSG reconciliation is active.
26    Reconciling,
27    /// Reconciliation closed and actions are being scheduled.
28    ReconciliationClosed,
29    /// Waiting for requested events to be durably ingested.
30    FetchingEvents,
31    /// Waiting for the platform to persist the final checkpoint.
32    SavingCheckpoint,
33    /// Checkpoint committed and synchronization completed.
34    Complete,
35    /// Host permanently cancelled this coordinator.
36    Cancelled,
37}
38
39/// Action emitted by the host-driven synchronization coordinator.
40#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
41#[serde(rename_all = "camelCase")]
42#[cfg_attr(feature = "native-bindings", derive(uniffi::Enum))]
43pub enum SyncActionKind {
44    /// Read one bounded local event snapshot.
45    LoadSnapshot,
46    /// Send NEG-OPEN with the filter and binary initial message.
47    SendNegOpen,
48    /// Send one subsequent NEG-MSG binary message.
49    SendNegMessage,
50    /// Close the current Negentropy subscription.
51    SendNegClose,
52    /// Resend locally held events missing from the relay.
53    ResendEvents,
54    /// Fetch relay-held events absent locally.
55    RequestEvents,
56    /// Commit the current sync checkpoint.
57    SaveCheckpoint,
58    /// Synchronization is complete.
59    Complete,
60}
61
62/// One transport/database action with fields selected by [`SyncActionKind`].
63#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
64#[serde(rename_all = "camelCase")]
65#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
66pub struct SyncAction {
67    /// Action discriminator.
68    pub kind: SyncActionKind,
69    /// Correlated Negentropy or ordinary event subscription.
70    pub subscription_id: String,
71    /// Inclusive lower Nostr timestamp, or `-1` when absent.
72    pub since_timestamp: i64,
73    /// Inclusive upper Nostr timestamp, or `-1` when unused.
74    pub until_timestamp: i64,
75    /// Snapshot/filter limit.
76    pub limit: u32,
77    /// Binary Negentropy message.
78    pub message: Vec<u8>,
79    /// Complete text frame for Negentropy transport actions.
80    pub frame_json: String,
81    /// Canonical filter JSON for ordinary event-request subscriptions.
82    pub filters_json: Vec<String>,
83    /// Canonical event IDs for fetch/resend.
84    pub event_ids: Vec<String>,
85    /// Final checkpoint value for `SaveCheckpoint`, otherwise `-1`.
86    pub checkpoint_timestamp: i64,
87}
88
89/// Redacted serializable synchronization state.
90#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
91#[serde(rename_all = "camelCase")]
92#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
93pub struct SyncEngineSnapshot {
94    /// Stable synchronization ID.
95    pub sync_id: String,
96    /// Current phase.
97    pub phase: SyncPhase,
98    /// Current inclusive upper timestamp.
99    pub until_timestamp: i64,
100    /// Current page limit.
101    pub page_limit: u32,
102    /// Accumulated locally-held ID count.
103    pub have_count: u32,
104    /// Accumulated remotely-held ID count.
105    pub need_count: u32,
106    /// Requested event count.
107    pub requested_count: u32,
108    /// Durably committed requested event count.
109    pub committed_count: u32,
110}
111
112/// Serialized synchronization state machine.
113pub struct SyncEngine {
114    sync_id: String,
115    neg_subscription_id: String,
116    event_subscription_id: String,
117    since_timestamp: Option<i64>,
118    checkpoint_timestamp: i64,
119    phase: SyncPhase,
120    pagination: SyncPagination,
121    page_event_count: usize,
122    page_oldest_timestamp: Option<i64>,
123    negentropy: Option<NegentropyClient>,
124    have_event_ids: BTreeSet<String>,
125    need_event_ids: BTreeSet<String>,
126    requested_event_ids: BTreeSet<String>,
127    fetched_event_ids: BTreeSet<String>,
128    committed_event_ids: BTreeSet<String>,
129}
130
131impl std::fmt::Debug for SyncEngine {
132    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
133        formatter
134            .debug_struct("SyncEngine")
135            .field("sync_id", &self.sync_id)
136            .field("phase", &self.phase)
137            .field("pagination", &self.pagination)
138            .field("have_count", &self.have_event_ids.len())
139            .field("need_count", &self.need_event_ids.len())
140            .finish()
141    }
142}
143
144#[derive(Clone, Copy, Debug, Eq, PartialEq)]
145struct SyncPagination {
146    until_timestamp: i64,
147    limit: u32,
148    expansion_boundary: Option<i64>,
149}
150
151impl SyncEngine {
152    /// Construct one synchronization coordinator.
153    ///
154    /// # Errors
155    ///
156    /// Returns a stable sync-state error for invalid identifiers, timestamps,
157    /// or an inverted checkpoint range.
158    pub fn new(
159        sync_id: String,
160        neg_subscription_id: String,
161        event_subscription_id: String,
162        since_timestamp: Option<i64>,
163        checkpoint_timestamp: i64,
164    ) -> Result<Self, SoftchatError> {
165        validate_sync_identifier(&sync_id)?;
166        validate_sync_identifier(&neg_subscription_id)?;
167        validate_sync_identifier(&event_subscription_id)?;
168        if neg_subscription_id == event_subscription_id
169            || checkpoint_timestamp < 0
170            || since_timestamp.is_some_and(|since| since < 0 || since > checkpoint_timestamp)
171        {
172            return Err(SoftchatError::InvalidSyncState);
173        }
174        Ok(Self {
175            sync_id,
176            neg_subscription_id,
177            event_subscription_id,
178            since_timestamp,
179            checkpoint_timestamp,
180            phase: SyncPhase::Idle,
181            pagination: SyncPagination {
182                until_timestamp: checkpoint_timestamp,
183                limit: SYNC_BASE_PAGE_LIMIT,
184                expansion_boundary: None,
185            },
186            page_event_count: 0,
187            page_oldest_timestamp: None,
188            negentropy: None,
189            have_event_ids: BTreeSet::new(),
190            need_event_ids: BTreeSet::new(),
191            requested_event_ids: BTreeSet::new(),
192            fetched_event_ids: BTreeSet::new(),
193            committed_event_ids: BTreeSet::new(),
194        })
195    }
196
197    /// Start synchronization by requesting the first platform snapshot.
198    ///
199    /// # Errors
200    ///
201    /// Returns a stable sync-state error when called outside `Idle`.
202    pub fn begin(&mut self) -> Result<SyncAction, SoftchatError> {
203        if self.phase != SyncPhase::Idle {
204            return Err(SoftchatError::InvalidSyncState);
205        }
206        self.phase = SyncPhase::LoadingSnapshot;
207        Ok(self.load_snapshot_action())
208    }
209
210    /// Consume one bounded local snapshot and produce NEG-OPEN.
211    ///
212    /// Items may arrive in any order; the coordinator computes the oldest
213    /// timestamp and the reviewed Negentropy engine canonicalizes the set.
214    ///
215    /// # Errors
216    ///
217    /// Returns a stable synchronization/Negentropy error for a stale phase,
218    /// oversized page, duplicate ID, invalid timestamp, or invalid frame size.
219    pub fn accept_snapshot(
220        &mut self,
221        items: Vec<NegentropyItem>,
222        frame_size_limit: u64,
223    ) -> Result<SyncAction, SoftchatError> {
224        if self.phase != SyncPhase::LoadingSnapshot
225            || items.len() > self.pagination.limit as usize
226            || items.len() > MAX_NEGENTROPY_ITEMS
227        {
228            return Err(SoftchatError::InvalidSyncState);
229        }
230        let mut unique = BTreeSet::new();
231        let mut oldest = None;
232        for item in &items {
233            if item.created_at < 0
234                || item.created_at > self.pagination.until_timestamp
235                || self
236                    .since_timestamp
237                    .is_some_and(|since| item.created_at < since)
238                || !unique.insert(item.event_id.clone())
239            {
240                return Err(SoftchatError::InvalidSyncState);
241            }
242            NostrEventId::from_hex(&item.event_id).map_err(|_| SoftchatError::InvalidSyncState)?;
243            oldest = Some(oldest.map_or(item.created_at, |value: i64| value.min(item.created_at)));
244        }
245        let client = NegentropyClient::new(items.clone(), frame_size_limit)?;
246        let message = client.initiate()?;
247        self.page_event_count = items.len();
248        self.page_oldest_timestamp = oldest;
249        self.negentropy = Some(client);
250        self.have_event_ids.clear();
251        self.need_event_ids.clear();
252        self.requested_event_ids.clear();
253        self.fetched_event_ids.clear();
254        self.committed_event_ids.clear();
255        self.phase = SyncPhase::Reconciling;
256        Ok(SyncAction {
257            kind: SyncActionKind::SendNegOpen,
258            subscription_id: self.neg_subscription_id.clone(),
259            since_timestamp: self.since_timestamp.unwrap_or(-1),
260            until_timestamp: self.pagination.until_timestamp,
261            limit: self.pagination.limit,
262            message,
263            frame_json: String::new(),
264            filters_json: Vec::new(),
265            event_ids: Vec::new(),
266            checkpoint_timestamp: -1,
267        })
268    }
269
270    /// Consume one correlated NEG-MSG and emit all resulting actions.
271    ///
272    /// Have/need IDs accumulate across every frame. Stale subscription IDs and
273    /// messages after reconciliation closes are rejected.
274    ///
275    /// # Errors
276    ///
277    /// Returns a stable sync/Negentropy error for malformed or stale input.
278    pub fn reconcile(
279        &mut self,
280        subscription_id: String,
281        message: Vec<u8>,
282    ) -> Result<Vec<SyncAction>, SoftchatError> {
283        if self.phase != SyncPhase::Reconciling || subscription_id != self.neg_subscription_id {
284            return Err(SoftchatError::InvalidSyncState);
285        }
286        let step = self
287            .negentropy
288            .as_ref()
289            .ok_or(SoftchatError::InvalidSyncState)?
290            .reconcile(message)?;
291        self.have_event_ids.extend(step.have_event_ids);
292        self.need_event_ids.extend(step.need_event_ids);
293        if self.have_event_ids.len() > MAX_NEGENTROPY_ITEMS
294            || self.need_event_ids.len() > MAX_NEGENTROPY_ITEMS
295        {
296            return Err(SoftchatError::InvalidSyncState);
297        }
298        if let Some(outgoing) = step.outgoing {
299            return Ok(vec![SyncAction {
300                kind: SyncActionKind::SendNegMessage,
301                subscription_id: self.neg_subscription_id.clone(),
302                since_timestamp: -1,
303                until_timestamp: -1,
304                limit: 0,
305                message: outgoing,
306                frame_json: String::new(),
307                filters_json: Vec::new(),
308                event_ids: Vec::new(),
309                checkpoint_timestamp: -1,
310            }]);
311        }
312        if !step.complete {
313            return Err(SoftchatError::InvalidSyncState);
314        }
315
316        self.negentropy = None;
317        self.phase = SyncPhase::ReconciliationClosed;
318        let mut actions = vec![simple_action(
319            SyncActionKind::SendNegClose,
320            self.neg_subscription_id.clone(),
321        )];
322        actions.extend(chunked_actions(
323            SyncActionKind::ResendEvents,
324            String::new(),
325            &self.have_event_ids,
326        ));
327        if self.need_event_ids.is_empty() {
328            actions.extend(self.advance_page()?);
329        } else {
330            self.requested_event_ids.clone_from(&self.need_event_ids);
331            self.phase = SyncPhase::FetchingEvents;
332            actions.extend(chunked_actions(
333                SyncActionKind::RequestEvents,
334                self.event_subscription_id.clone(),
335                &self.requested_event_ids,
336            ));
337        }
338        Ok(actions)
339    }
340
341    /// Record one requested event received on the current subscription.
342    ///
343    /// Authentication and persistence are separate; this call alone does not
344    /// permit checkpoint advancement.
345    ///
346    /// # Errors
347    ///
348    /// Returns a stable sync-state error for a stale subscription, duplicate,
349    /// malformed, or unrequested event ID.
350    pub fn record_fetched_event(
351        &mut self,
352        subscription_id: String,
353        event_id: String,
354    ) -> Result<(), SoftchatError> {
355        if self.phase != SyncPhase::FetchingEvents
356            || subscription_id != self.event_subscription_id
357            || NostrEventId::from_hex(&event_id).is_err()
358            || !self.requested_event_ids.contains(&event_id)
359            || !self.fetched_event_ids.insert(event_id)
360        {
361            return Err(SoftchatError::InvalidSyncState);
362        }
363        Ok(())
364    }
365
366    /// Validate a complete candidate batch before its durable transaction.
367    ///
368    /// This permits the Rust-owned account runtime to hold the serialized sync
369    /// coordinator, authenticate and commit the batch, and then advance the
370    /// in-memory state without a cross-language persistence callback.
371    #[cfg(feature = "sqlite-storage")]
372    pub(crate) fn validate_fetched_events(
373        &self,
374        subscription_id: &str,
375        event_ids: &[String],
376    ) -> Result<(), SoftchatError> {
377        if self.phase != SyncPhase::FetchingEvents
378            || subscription_id != self.event_subscription_id
379            || event_ids.is_empty()
380        {
381            return Err(SoftchatError::InvalidSyncState);
382        }
383        let mut unique = BTreeSet::new();
384        for event_id in event_ids {
385            if NostrEventId::from_hex(event_id).is_err()
386                || !self.requested_event_ids.contains(event_id)
387                || self.fetched_event_ids.contains(event_id)
388                || !unique.insert(event_id)
389            {
390                return Err(SoftchatError::InvalidSyncState);
391            }
392        }
393        Ok(())
394    }
395
396    /// Confirm an authoritative platform transaction committed requested IDs.
397    ///
398    /// # Errors
399    ///
400    /// Returns a stable sync-state error for a stale phase, duplicate,
401    /// unrequested, or not-yet-received ID.
402    pub fn confirm_committed(&mut self, event_ids: Vec<String>) -> Result<(), SoftchatError> {
403        if self.phase != SyncPhase::FetchingEvents || event_ids.is_empty() {
404            return Err(SoftchatError::InvalidSyncState);
405        }
406        for event_id in event_ids {
407            if !self.fetched_event_ids.contains(&event_id)
408                || !self.committed_event_ids.insert(event_id)
409            {
410                return Err(SoftchatError::InvalidSyncState);
411            }
412        }
413        Ok(())
414    }
415
416    /// Close the ordinary event fetch only after every requested ID committed.
417    ///
418    /// # Errors
419    ///
420    /// Returns a stable sync-state error for a stale subscription or incomplete
421    /// receipt set.
422    pub fn finish_fetch(
423        &mut self,
424        subscription_id: String,
425    ) -> Result<Vec<SyncAction>, SoftchatError> {
426        if self.phase != SyncPhase::FetchingEvents
427            || subscription_id != self.event_subscription_id
428            || self.fetched_event_ids != self.requested_event_ids
429            || self.committed_event_ids != self.requested_event_ids
430        {
431            return Err(SoftchatError::InvalidSyncState);
432        }
433        self.phase = SyncPhase::ReconciliationClosed;
434        self.advance_page()
435    }
436
437    /// Confirm the platform durably saved the final checkpoint.
438    ///
439    /// # Errors
440    ///
441    /// Returns a stable sync-state error for a stale phase or mismatched value.
442    pub fn checkpoint_saved(
443        &mut self,
444        checkpoint_timestamp: i64,
445    ) -> Result<SyncAction, SoftchatError> {
446        if self.phase != SyncPhase::SavingCheckpoint
447            || checkpoint_timestamp != self.checkpoint_timestamp
448        {
449            return Err(SoftchatError::InvalidSyncState);
450        }
451        self.phase = SyncPhase::Complete;
452        Ok(simple_action(SyncActionKind::Complete, String::new()))
453    }
454
455    /// Cancel and clear process-local reconciliation state.
456    #[must_use]
457    pub fn cancel(&mut self) -> SyncEngineSnapshot {
458        self.phase = SyncPhase::Cancelled;
459        self.negentropy = None;
460        self.have_event_ids.clear();
461        self.need_event_ids.clear();
462        self.requested_event_ids.clear();
463        self.fetched_event_ids.clear();
464        self.committed_event_ids.clear();
465        self.snapshot()
466    }
467
468    /// Return one redacted serializable snapshot.
469    #[must_use]
470    pub fn snapshot(&self) -> SyncEngineSnapshot {
471        SyncEngineSnapshot {
472            sync_id: self.sync_id.clone(),
473            phase: self.phase,
474            until_timestamp: self.pagination.until_timestamp,
475            page_limit: self.pagination.limit,
476            have_count: bounded_count(self.have_event_ids.len()),
477            need_count: bounded_count(self.need_event_ids.len()),
478            requested_count: bounded_count(self.requested_event_ids.len()),
479            committed_count: bounded_count(self.committed_event_ids.len()),
480        }
481    }
482
483    fn advance_page(&mut self) -> Result<Vec<SyncAction>, SoftchatError> {
484        let next = self.pagination.next(
485            self.page_event_count,
486            self.page_oldest_timestamp,
487            self.since_timestamp,
488        )?;
489        self.have_event_ids.clear();
490        self.need_event_ids.clear();
491        self.requested_event_ids.clear();
492        self.fetched_event_ids.clear();
493        self.committed_event_ids.clear();
494        if let Some(pagination) = next {
495            self.pagination = pagination;
496            self.page_event_count = 0;
497            self.page_oldest_timestamp = None;
498            self.phase = SyncPhase::LoadingSnapshot;
499            Ok(vec![self.load_snapshot_action()])
500        } else {
501            self.phase = SyncPhase::SavingCheckpoint;
502            Ok(vec![SyncAction {
503                kind: SyncActionKind::SaveCheckpoint,
504                subscription_id: String::new(),
505                since_timestamp: -1,
506                until_timestamp: -1,
507                limit: 0,
508                message: Vec::new(),
509                frame_json: String::new(),
510                filters_json: Vec::new(),
511                event_ids: Vec::new(),
512                checkpoint_timestamp: self.checkpoint_timestamp,
513            }])
514        }
515    }
516
517    fn load_snapshot_action(&self) -> SyncAction {
518        SyncAction {
519            kind: SyncActionKind::LoadSnapshot,
520            subscription_id: String::new(),
521            since_timestamp: self.since_timestamp.unwrap_or(-1),
522            until_timestamp: self.pagination.until_timestamp,
523            limit: self.pagination.limit,
524            message: Vec::new(),
525            frame_json: String::new(),
526            filters_json: Vec::new(),
527            event_ids: Vec::new(),
528            checkpoint_timestamp: -1,
529        }
530    }
531}
532
533impl SyncPagination {
534    fn next(
535        self,
536        event_count: usize,
537        oldest_timestamp: Option<i64>,
538        since_timestamp: Option<i64>,
539    ) -> Result<Option<Self>, SoftchatError> {
540        if event_count < self.limit as usize {
541            return Ok(None);
542        }
543        let oldest_timestamp = oldest_timestamp.ok_or(SoftchatError::InvalidSyncState)?;
544        if let Some(boundary) = self.expansion_boundary
545            && oldest_timestamp < boundary
546        {
547            if since_timestamp.is_some_and(|since| boundary <= since) {
548                return Ok(None);
549            }
550            let until_timestamp = boundary
551                .checked_sub(1)
552                .ok_or(SoftchatError::InvalidSyncState)?;
553            return Ok(Some(Self {
554                until_timestamp,
555                limit: SYNC_BASE_PAGE_LIMIT,
556                expansion_boundary: None,
557            }));
558        }
559        let limit = self
560            .limit
561            .checked_mul(2)
562            .filter(|limit| *limit <= SYNC_MAX_PAGE_LIMIT)
563            .ok_or(SoftchatError::InvalidSyncState)?;
564        Ok(Some(Self {
565            until_timestamp: self.until_timestamp,
566            limit,
567            expansion_boundary: Some(self.expansion_boundary.unwrap_or(oldest_timestamp)),
568        }))
569    }
570}
571
572fn validate_sync_identifier(value: &str) -> Result<(), SoftchatError> {
573    if value.is_empty()
574        || value.len() > 128
575        || !value
576            .bytes()
577            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b':'))
578    {
579        Err(SoftchatError::InvalidSyncState)
580    } else {
581        Ok(())
582    }
583}
584
585fn chunked_actions(
586    kind: SyncActionKind,
587    subscription_id: String,
588    event_ids: &BTreeSet<String>,
589) -> Vec<SyncAction> {
590    event_ids
591        .iter()
592        .cloned()
593        .collect::<Vec<_>>()
594        .chunks(SYNC_EVENT_REQUEST_CHUNK)
595        .map(|chunk| SyncAction {
596            kind,
597            subscription_id: subscription_id.clone(),
598            since_timestamp: -1,
599            until_timestamp: -1,
600            limit: 0,
601            message: Vec::new(),
602            frame_json: String::new(),
603            filters_json: Vec::new(),
604            event_ids: chunk.to_vec(),
605            checkpoint_timestamp: -1,
606        })
607        .collect()
608}
609
610fn simple_action(kind: SyncActionKind, subscription_id: String) -> SyncAction {
611    SyncAction {
612        kind,
613        subscription_id,
614        since_timestamp: -1,
615        until_timestamp: -1,
616        limit: 0,
617        message: Vec::new(),
618        frame_json: String::new(),
619        filters_json: Vec::new(),
620        event_ids: Vec::new(),
621        checkpoint_timestamp: -1,
622    }
623}
624
625fn bounded_count(value: usize) -> u32 {
626    u32::try_from(value).unwrap_or(u32::MAX)
627}
628
629#[cfg(test)]
630#[allow(clippy::unwrap_used)]
631mod tests {
632    use softrelay_negentropy::{Id, Negentropy, NegentropyStorageVector, Storage};
633
634    use super::*;
635
636    fn item(created_at: i64, byte: u8) -> NegentropyItem {
637        NegentropyItem {
638            created_at,
639            event_id: hex::encode([byte; 32]),
640        }
641    }
642
643    fn relay_response(initial: &[u8], items: &[(u64, u8)]) -> Vec<u8> {
644        let mut storage = NegentropyStorageVector::new();
645        for (timestamp, byte) in items {
646            storage
647                .insert(*timestamp, Id::from_byte_array([*byte; 32]))
648                .unwrap();
649        }
650        storage.seal().unwrap();
651        let mut relay = Negentropy::new(Storage::Borrowed(&storage), 4_096).unwrap();
652        relay.reconcile(initial).unwrap()
653    }
654
655    fn engine() -> SyncEngine {
656        SyncEngine::new(
657            "sync-1".to_owned(),
658            "neg-1".to_owned(),
659            "events-1".to_owned(),
660            Some(1),
661            10,
662        )
663        .unwrap()
664    }
665
666    #[test]
667    fn requires_requested_events_to_commit_before_checkpoint_progress() {
668        let mut engine = engine();
669        assert_eq!(engine.begin().unwrap().kind, SyncActionKind::LoadSnapshot);
670        let open = engine
671            .accept_snapshot(vec![item(1, 0x11), item(3, 0x33)], 4_096)
672            .unwrap();
673        let response = relay_response(&open.message, &[(2, 0x22), (3, 0x33)]);
674        let actions = engine.reconcile("neg-1".to_owned(), response).unwrap();
675        assert!(
676            actions
677                .iter()
678                .any(|action| action.kind == SyncActionKind::ResendEvents)
679        );
680        let request = actions
681            .iter()
682            .find(|action| action.kind == SyncActionKind::RequestEvents)
683            .unwrap();
684        let needed = request.event_ids[0].clone();
685        engine
686            .record_fetched_event("events-1".to_owned(), needed.clone())
687            .unwrap();
688        assert!(matches!(
689            engine.finish_fetch("events-1".to_owned()),
690            Err(SoftchatError::InvalidSyncState)
691        ));
692        engine.confirm_committed(vec![needed]).unwrap();
693        let checkpoint = engine.finish_fetch("events-1".to_owned()).unwrap();
694        assert_eq!(checkpoint[0].kind, SyncActionKind::SaveCheckpoint);
695        assert_eq!(
696            engine.checkpoint_saved(10).unwrap().kind,
697            SyncActionKind::Complete
698        );
699    }
700
701    #[test]
702    fn expands_full_timestamp_cohorts_then_advances_older_window() {
703        let pagination = SyncPagination {
704            until_timestamp: 100,
705            limit: 900,
706            expansion_boundary: None,
707        };
708        let expanded = pagination.next(900, Some(50), Some(1)).unwrap().unwrap();
709        assert_eq!(expanded.limit, 1_800);
710        assert_eq!(expanded.until_timestamp, 100);
711        assert_eq!(expanded.expansion_boundary, Some(50));
712
713        let older = expanded.next(1_800, Some(49), Some(1)).unwrap().unwrap();
714        assert_eq!(older.limit, 900);
715        assert_eq!(older.until_timestamp, 49);
716        assert_eq!(older.expansion_boundary, None);
717    }
718
719    #[test]
720    fn rejects_stale_subscriptions_and_missing_events() {
721        let mut engine = engine();
722        engine.begin().unwrap();
723        let open = engine.accept_snapshot(Vec::new(), 4_096).unwrap();
724        assert!(matches!(
725            engine.reconcile("stale".to_owned(), open.message),
726            Err(SoftchatError::InvalidSyncState)
727        ));
728    }
729}
softchat/sync.rs

softchat/
sync.rs
