nodedb_lite/sync/array/inbound/
resume.rs

1//! Session catch-up — the requests a new sync session sends to resume every
2//! array this replica holds, and the tracker access the apply paths use to
3//! record what they applied.
4
5use std::sync::Arc;
6
7use nodedb_types::sync::wire::array::ArrayCatchupRequestMsg;
8
9use crate::error::LiteError;
10use crate::storage::engine::StorageEngine;
11use crate::sync::array::catchup::CatchupTracker;
12
13use super::dispatcher::ArrayInbound;
14
15impl<S: StorageEngine> ArrayInbound<S> {
16    /// The per-array high-water mark of ops applied from Origin.
17    pub fn catchup_tracker(&self) -> &Arc<CatchupTracker<S>> {
18        &self.catchup
19    }
20
21    /// Start the catch-up round for a new session.
22    ///
23    /// Drops partial snapshot assemblies left by the previous session, then
24    /// returns one [`ArrayCatchupRequestMsg`] per array in the local schema
25    /// registry, sorted by name. Each request starts at the tracker's
26    /// last-seen HLC, or at [`nodedb_array::sync::hlc::Hlc::ZERO`] for an
27    /// array a `RetentionFloor` reject flagged. Returns an empty list when
28    /// the replica holds no arrays.
29    pub fn begin_catchup_round(&self) -> Result<Vec<ArrayCatchupRequestMsg>, LiteError> {
30        self.discard_partial_snapshots()?;
31        let mut arrays = self.schemas.list_arrays();
32        arrays.sort_unstable();
33        Ok(arrays
34            .iter()
35            .map(|array| self.catchup.build_request(array))
36            .collect())
37    }
38}
39
40#[cfg(test)]
41mod tests {
42    use nodedb_array::sync::hlc::Hlc;
43
44    use super::super::fixtures::{hlc, make_inbound, simple_schema};
45
46    #[tokio::test(flavor = "multi_thread")]
47    async fn catchup_round_is_empty_without_arrays() {
48        let (inbound, _schemas, _pending, _storage) = make_inbound().await;
49        assert!(inbound.begin_catchup_round().unwrap().is_empty());
50    }
51
52    #[tokio::test(flavor = "multi_thread")]
53    async fn catchup_round_requests_each_array_from_its_last_seen_hlc() {
54        let (inbound, schemas, _pending, _storage) = make_inbound().await;
55        schemas.put_schema("b", &simple_schema("b")).await.unwrap();
56        schemas.put_schema("a", &simple_schema("a")).await.unwrap();
57        inbound
58            .catchup_tracker()
59            .record("b", hlc(700))
60            .await
61            .unwrap();
62
63        let requests = inbound.begin_catchup_round().unwrap();
64        assert_eq!(requests.len(), 2);
65        assert_eq!(requests[0].array, "a");
66        assert_eq!(requests[0].from_hlc_bytes, Hlc::ZERO.to_bytes());
67        assert_eq!(requests[1].array, "b");
68        assert_eq!(requests[1].from_hlc_bytes, hlc(700).to_bytes());
69    }
70
71    #[tokio::test(flavor = "multi_thread")]
72    async fn catchup_round_requests_flagged_array_from_zero() {
73        let (inbound, schemas, _pending, _storage) = make_inbound().await;
74        schemas.put_schema("a", &simple_schema("a")).await.unwrap();
75        let tracker = inbound.catchup_tracker();
76        tracker.record("a", hlc(900)).await.unwrap();
77        tracker.record_reject_retention_floor("a").await.unwrap();
78
79        let requests = inbound.begin_catchup_round().unwrap();
80        assert_eq!(requests.len(), 1);
81        assert_eq!(requests[0].from_hlc_bytes, Hlc::ZERO.to_bytes());
82    }
83}
nodedb_lite/sync/array/inbound/resume.rs

nodedb_lite/sync/array/inbound/
resume.rs
