nodedb_lite/sync/array/
catchup.rs

1//! Catchup-request helper: tracks the last-seen inbound HLC per array and
2//! builds [`ArrayCatchupRequestMsg`] messages for the transport layer.
3//!
4//! # Persistence
5//!
6//! Each per-array `last_seen_hlc` is persisted under `Namespace::Meta` with
7//! key `b"array.last_seen_hlc:{name}"`. [`CatchupTracker::load`] restores all
8//! entries on startup so the first catchup request after a restart asks only
9//! for ops the device has never seen.
10
11use std::collections::HashMap;
12use std::sync::{Arc, Mutex};
13
14use nodedb_array::sync::hlc::Hlc;
15use nodedb_types::Namespace;
16use nodedb_types::sync::wire::array::ArrayCatchupRequestMsg;
17
18use crate::error::LiteError;
19use crate::storage::engine::StorageEngine;
20
21/// Storage key prefix for per-array last-seen HLC entries.
22const LAST_SEEN_PREFIX: &str = "array.last_seen_hlc:";
23
24/// Storage key prefix for per-array catchup-needed flags.
25const CATCHUP_NEEDED_PREFIX: &str = "array.catchup_needed:";
26
27/// Storage key for a named array's last-seen HLC.
28fn last_seen_key(array: &str) -> Vec<u8> {
29    format!("{LAST_SEEN_PREFIX}{array}").into_bytes()
30}
31
32/// Storage key for a named array's catchup-needed flag.
33fn catchup_needed_key(array: &str) -> Vec<u8> {
34    format!("{CATCHUP_NEEDED_PREFIX}{array}").into_bytes()
35}
36
37/// Tracks the last successfully applied inbound HLC per array.
38///
39/// Used by the transport layer to populate catchup-request messages after
40/// reconnect or when Origin's log GC horizon advances past the local log.
41///
42/// Thread-safe via an internal [`Mutex`].
43pub struct CatchupTracker<S: StorageEngine> {
44    storage: Arc<S>,
45    state: Mutex<HashMap<String, Hlc>>,
46    /// Arrays that need a full catchup on next connect.
47    /// Set when Origin sends `ArrayRejectMsg::RetentionFloor`.
48    catchup_needed: Mutex<std::collections::HashSet<String>>,
49    /// Serializes every mutation with its storage write.
50    ///
51    /// A mutation updates memory under a `std::sync` guard, releases it, then
52    /// awaits the storage write. Without this lock two writers can reorder
53    /// their writes, and storage keeps an older HLC or flag than memory holds.
54    write_lock: tokio::sync::Mutex<()>,
55}
56
57impl<S: StorageEngine> CatchupTracker<S> {
58    /// Load all persisted `last_seen_hlc` entries from `Namespace::Meta`.
59    ///
60    /// Scans keys starting with `b"array.last_seen_hlc:"` and parses each
61    /// 18-byte value as an [`Hlc`]. Returns an empty tracker if no entries
62    /// are stored yet (first run or after a full wipe).
63    pub async fn load(storage: Arc<S>) -> Result<Self, LiteError> {
64        let prefix = LAST_SEEN_PREFIX.as_bytes();
65        let pairs = storage.scan_prefix(Namespace::Meta, prefix).await?;
66
67        let mut state = HashMap::new();
68        for (key, value) in pairs {
69            let name_bytes = &key[prefix.len()..];
70            let name = std::str::from_utf8(name_bytes).map_err(|e| LiteError::Storage {
71                detail: format!("catchup_tracker: non-UTF8 array name in storage: {e}"),
72            })?;
73
74            let hlc_arr: [u8; 18] = value.try_into().map_err(|v: Vec<u8>| LiteError::Storage {
75                detail: format!(
76                    "catchup_tracker: last_seen_hlc wrong length ({}) for '{name}'",
77                    v.len()
78                ),
79            })?;
80            state.insert(name.to_owned(), Hlc::from_bytes(&hlc_arr));
81        }
82
83        // Load catchup-needed flags.
84        let needed_prefix = CATCHUP_NEEDED_PREFIX.as_bytes();
85        let needed_pairs = storage.scan_prefix(Namespace::Meta, needed_prefix).await?;
86        let mut catchup_needed = std::collections::HashSet::new();
87        for (key, _) in needed_pairs {
88            let name_bytes = &key[needed_prefix.len()..];
89            if let Ok(name) = std::str::from_utf8(name_bytes) {
90                catchup_needed.insert(name.to_owned());
91            }
92        }
93
94        Ok(Self {
95            storage,
96            state: Mutex::new(state),
97            catchup_needed: Mutex::new(catchup_needed),
98            write_lock: tokio::sync::Mutex::new(()),
99        })
100    }
101
102    /// Record that `hlc` has been successfully applied for `array`.
103    ///
104    /// Updates both the in-memory map and the durable storage entry.
105    /// Only persists if `hlc` is strictly greater than the current last-seen
106    /// value (monotonic advancement).
107    pub async fn record(&self, array: &str, hlc: Hlc) -> Result<(), LiteError> {
108        let _write = self.write_lock.lock().await;
109        let advanced = {
110            let mut state = self.state.lock().map_err(|_| LiteError::LockPoisoned)?;
111            let current = state.get(array).copied().unwrap_or(Hlc::ZERO);
112            if hlc > current {
113                state.insert(array.to_owned(), hlc);
114            }
115            hlc > current
116        };
117        if !advanced {
118            return Ok(());
119        }
120        self.storage
121            .put(Namespace::Meta, &last_seen_key(array), &hlc.to_bytes())
122            .await
123    }
124
125    /// Return the last-seen HLC for `array`, or [`Hlc::ZERO`] if unknown.
126    ///
127    /// [`Hlc::ZERO`] in a catchup request tells Origin to replay all history
128    /// for the array from the beginning.
129    pub fn last_seen(&self, array: &str) -> Hlc {
130        self.state
131            .lock()
132            .ok()
133            .and_then(|s| s.get(array).copied())
134            .unwrap_or(Hlc::ZERO)
135    }
136
137    /// Build an [`ArrayCatchupRequestMsg`] for `array`.
138    ///
139    /// The request starts at the stored last-seen HLC. An array flagged by a
140    /// `RetentionFloor` reject starts at [`Hlc::ZERO`] instead: that HLC is
141    /// below Origin's GC boundary, so Origin answers with a snapshot followed
142    /// by the op stream above it.
143    pub fn build_request(&self, array: &str) -> ArrayCatchupRequestMsg {
144        let from = if self.is_catchup_needed(array) {
145            Hlc::ZERO
146        } else {
147            self.last_seen(array)
148        };
149        ArrayCatchupRequestMsg {
150            array: array.to_owned(),
151            from_hlc_bytes: from.to_bytes(),
152        }
153    }
154
155    /// Returns `true` if a `RetentionFloor` reject flagged `array` and no
156    /// snapshot has completed since.
157    pub fn is_catchup_needed(&self, array: &str) -> bool {
158        self.catchup_needed
159            .lock()
160            .ok()
161            .is_some_and(|s| s.contains(array))
162    }
163
164    /// Returns `true` if `array` should request a full catch-up on connect.
165    ///
166    /// True when:
167    /// - No `last_seen_hlc` is recorded (first connect), OR
168    /// - `record_reject_retention_floor` was called (Origin GC advanced past our log).
169    pub fn should_request_catchup(&self, array: &str, _current_local_hlc: Hlc) -> bool {
170        if self.is_catchup_needed(array) {
171            return true;
172        }
173        // First connect: no last-seen HLC recorded.
174        let state = self.state.lock().ok();
175        state.map(|s| !s.contains_key(array)).unwrap_or(false)
176    }
177
178    /// Mark `array` as needing a full catch-up on the next connect.
179    ///
180    /// Called when Origin sends `ArrayRejectMsg::RetentionFloor`.
181    /// Persisted under `Namespace::Meta` `"array.catchup_needed:{array}"`.
182    pub async fn record_reject_retention_floor(&self, array: &str) -> Result<(), LiteError> {
183        let _write = self.write_lock.lock().await;
184        self.catchup_needed
185            .lock()
186            .map_err(|_| LiteError::LockPoisoned)?
187            .insert(array.to_owned());
188        // Persist flag (value = 1 byte sentinel).
189        self.storage
190            .put(Namespace::Meta, &catchup_needed_key(array), &[1u8])
191            .await
192    }
193
194    /// Clear the catchup-needed flag for `array` after a successful catch-up.
195    ///
196    /// Called after the snapshot stream has been fully applied.
197    pub async fn clear_catchup_needed(&self, array: &str) -> Result<(), LiteError> {
198        let _write = self.write_lock.lock().await;
199        self.catchup_needed
200            .lock()
201            .map_err(|_| LiteError::LockPoisoned)?
202            .remove(array);
203        // Remove the persisted flag.
204        self.storage
205            .delete(Namespace::Meta, &catchup_needed_key(array))
206            .await
207    }
208
209    /// Return all arrays that are marked as needing catch-up.
210    pub fn arrays_needing_catchup(&self) -> Vec<String> {
211        self.catchup_needed
212            .lock()
213            .ok()
214            .map(|s| s.iter().cloned().collect())
215            .unwrap_or_default()
216    }
217}
218
219// ─── Tests ────────────────────────────────────────────────────────────────────
220
221#[cfg(test)]
222mod tests {
223    use super::*;
224    use crate::storage::pagedb_storage::{PagedbStorageDefault, PagedbStorageMem};
225    use nodedb_array::sync::replica_id::ReplicaId;
226
227    fn rep() -> ReplicaId {
228        ReplicaId::new(1)
229    }
230
231    fn hlc(ms: u64) -> Hlc {
232        Hlc::new(ms, 0, rep()).unwrap()
233    }
234
235    async fn make_tracker() -> CatchupTracker<PagedbStorageMem> {
236        let storage = Arc::new(PagedbStorageMem::open_in_memory().await.unwrap());
237        CatchupTracker::load(storage).await.unwrap()
238    }
239
240    #[tokio::test]
241    async fn load_returns_zero_when_empty() {
242        let tracker = make_tracker().await;
243        assert_eq!(tracker.last_seen("any_array"), Hlc::ZERO);
244    }
245
246    #[tokio::test]
247    async fn record_persists_across_load() {
248        let dir = tempfile::tempdir().unwrap();
249        let path = dir.path().join("catchup_test.pagedb");
250
251        let target_hlc = hlc(42_000);
252
253        {
254            let storage = Arc::new(
255                PagedbStorageDefault::open(
256                    &path,
257                    crate::storage::encryption::Encryption::Plaintext,
258                )
259                .await
260                .unwrap(),
261            );
262            let tracker = CatchupTracker::load(Arc::clone(&storage)).await.unwrap();
263            tracker.record("arr", target_hlc).await.unwrap();
264            assert_eq!(tracker.last_seen("arr"), target_hlc);
265        }
266
267        {
268            let storage = Arc::new(
269                PagedbStorageDefault::open(
270                    &path,
271                    crate::storage::encryption::Encryption::Plaintext,
272                )
273                .await
274                .unwrap(),
275            );
276            let tracker = CatchupTracker::load(storage).await.unwrap();
277            assert_eq!(
278                tracker.last_seen("arr"),
279                target_hlc,
280                "last_seen must survive storage restart"
281            );
282        }
283    }
284
285    #[tokio::test]
286    async fn record_is_monotonic_only() {
287        let tracker = make_tracker().await;
288        let h1 = hlc(100);
289        let h2 = hlc(200);
290
291        tracker.record("x", h2).await.unwrap();
292        // Recording a smaller HLC must not regress the stored value.
293        tracker.record("x", h1).await.unwrap();
294        assert_eq!(tracker.last_seen("x"), h2);
295    }
296
297    #[tokio::test]
298    async fn build_request_carries_last_seen() {
299        let tracker = make_tracker().await;
300        let h = hlc(9_999);
301        tracker.record("feed", h).await.unwrap();
302
303        let req = tracker.build_request("feed");
304        assert_eq!(req.array, "feed");
305        assert_eq!(req.from_hlc_bytes, h.to_bytes());
306    }
307
308    #[tokio::test]
309    async fn build_request_starts_at_zero_while_retention_floor_flag_is_set() {
310        let tracker = make_tracker().await;
311        let h = hlc(5_000);
312        tracker.record("feed", h).await.unwrap();
313        tracker.record_reject_retention_floor("feed").await.unwrap();
314
315        assert!(tracker.is_catchup_needed("feed"));
316        assert_eq!(
317            tracker.build_request("feed").from_hlc_bytes,
318            Hlc::ZERO.to_bytes(),
319            "a flagged array must ask for history below the GC boundary"
320        );
321
322        tracker.clear_catchup_needed("feed").await.unwrap();
323        assert!(!tracker.is_catchup_needed("feed"));
324        assert_eq!(tracker.build_request("feed").from_hlc_bytes, h.to_bytes());
325    }
326
327    #[tokio::test]
328    async fn build_request_zero_when_unknown() {
329        let tracker = make_tracker().await;
330        let req = tracker.build_request("unknown");
331        assert_eq!(req.from_hlc_bytes, Hlc::ZERO.to_bytes());
332    }
333}
nodedb_lite/sync/array/catchup.rs

nodedb_lite/sync/array/
catchup.rs
