powdb_sync/
replica.rs

1use std::fs;
2use std::io;
3use std::path::Path;
4
5use crate::metadata::{
6    now_unix_secs, read_replica_cursors_unlocked, replace_replica_cursors_unlocked,
7    with_cursor_metadata_lock,
8};
9use crate::segment::list_segment_files;
10use crate::{read_identity, retained_segments_dir, retained_tail_progress, ReplicaCursor};
11
12#[derive(Debug, Clone, Copy, PartialEq, Eq)]
13pub enum SyncRepairAction {
14    None,
15    Pull,
16    AwaitArchive,
17    Rebootstrap,
18}
19
20#[derive(Debug, Clone, PartialEq, Eq)]
21pub struct ReplicaSyncStatus {
22    pub replica_id: String,
23    pub active: bool,
24    pub last_applied_lsn: Option<u64>,
25    pub remote_lsn: u64,
26    /// Largest contiguous retained-unit LSN currently servable after the
27    /// replica cursor. This can lag behind `remote_lsn` while the primary has
28    /// progressed but its latest WAL records have not yet been published into
29    /// retained segments.
30    pub servable_lsn: Option<u64>,
31    /// Primary progress that is not yet represented in contiguous retained
32    /// history for this replica.
33    pub unarchived_lsn: Option<u64>,
34    pub lag_lsn: Option<u64>,
35    /// Estimated retained segment bytes that overlap the unapplied LSN range.
36    pub lag_bytes: Option<u64>,
37    /// Milliseconds since the primary-side cursor was last acknowledged while stale.
38    pub lag_ms: Option<u64>,
39    pub stale: bool,
40    pub repair_action: SyncRepairAction,
41    pub last_sync_error: Option<String>,
42}
43
44#[derive(Debug, Clone, PartialEq, Eq)]
45pub struct ReplicaApplyAckSummary {
46    pub replica_id: String,
47    pub previous_applied_lsn: u64,
48    pub applied_lsn: u64,
49    pub remote_lsn: u64,
50    pub advanced: bool,
51    pub status: ReplicaSyncStatus,
52}
53
54/// Acknowledge that a replica successfully applied retained history through an
55/// LSN observed on the primary.
56///
57/// This is the primary-side half of the future pull protocol: after a replica
58/// applies a tail locally, the server should call this to advance the retained
59/// history floor for that replica. The update is monotonic and fail-closed:
60/// stale acknowledgements, inactive cursors, and acknowledgements beyond the
61/// advertised primary LSN are rejected.
62pub fn acknowledge_replica_apply(
63    data_dir: &Path,
64    replica_id: &str,
65    applied_lsn: u64,
66    remote_lsn: u64,
67) -> io::Result<ReplicaApplyAckSummary> {
68    acknowledge_replica_apply_at(
69        data_dir,
70        replica_id,
71        applied_lsn,
72        remote_lsn,
73        now_unix_secs(),
74    )
75}
76
77/// Return primary-side sync status for one replica cursor.
78pub fn replica_sync_status(
79    data_dir: &Path,
80    replica_id: &str,
81    remote_lsn: u64,
82) -> io::Result<ReplicaSyncStatus> {
83    replica_sync_status_at(data_dir, replica_id, remote_lsn, now_unix_secs())
84}
85
86fn acknowledge_replica_apply_at(
87    data_dir: &Path,
88    replica_id: &str,
89    applied_lsn: u64,
90    remote_lsn: u64,
91    now: u64,
92) -> io::Result<ReplicaApplyAckSummary> {
93    if applied_lsn > remote_lsn {
94        return Err(invalid_input(format!(
95            "replica acknowledgement LSN {applied_lsn} is ahead of remote LSN {remote_lsn}"
96        )));
97    }
98
99    with_cursor_metadata_lock(data_dir, || {
100        let mut cursors = read_replica_cursors_unlocked(data_dir)?;
101        let Some(cursor) = cursors
102            .iter_mut()
103            .find(|cursor| cursor.replica_id == replica_id)
104        else {
105            return Err(io::Error::new(
106                io::ErrorKind::NotFound,
107                "replica cursor not found; rebootstrap required",
108            ));
109        };
110        if !cursor.active {
111            return Err(invalid_input(
112                "replica cursor is inactive; rebootstrap required",
113            ));
114        }
115        if applied_lsn < cursor.applied_lsn {
116            return Err(invalid_input(format!(
117                "replica acknowledgement LSN {applied_lsn} is behind primary cursor LSN {}",
118                cursor.applied_lsn
119            )));
120        }
121
122        let previous_applied_lsn = cursor.applied_lsn;
123        let mut acknowledged = cursor.clone();
124        acknowledged.applied_lsn = applied_lsn;
125        acknowledged.updated_unix_secs = now;
126        let status = status_for_cursor(data_dir, replica_id, Some(&acknowledged), remote_lsn, now)?;
127        *cursor = acknowledged;
128        replace_replica_cursors_unlocked(data_dir, cursors)?;
129
130        Ok(ReplicaApplyAckSummary {
131            replica_id: replica_id.to_string(),
132            previous_applied_lsn,
133            applied_lsn,
134            remote_lsn,
135            advanced: applied_lsn > previous_applied_lsn,
136            status,
137        })
138    })
139}
140
141fn replica_sync_status_at(
142    data_dir: &Path,
143    replica_id: &str,
144    remote_lsn: u64,
145    now: u64,
146) -> io::Result<ReplicaSyncStatus> {
147    let cursors = read_replica_cursors_unlocked(data_dir)?;
148    let cursor = cursors
149        .iter()
150        .find(|cursor| cursor.replica_id == replica_id);
151    status_for_cursor(data_dir, replica_id, cursor, remote_lsn, now)
152}
153
154fn status_for_cursor(
155    data_dir: &Path,
156    replica_id: &str,
157    cursor: Option<&ReplicaCursor>,
158    remote_lsn: u64,
159    now: u64,
160) -> io::Result<ReplicaSyncStatus> {
161    let Some(cursor) = cursor else {
162        return Ok(ReplicaSyncStatus {
163            replica_id: replica_id.to_string(),
164            active: false,
165            last_applied_lsn: None,
166            remote_lsn,
167            servable_lsn: None,
168            unarchived_lsn: None,
169            lag_lsn: None,
170            lag_bytes: None,
171            lag_ms: None,
172            stale: true,
173            repair_action: SyncRepairAction::Rebootstrap,
174            last_sync_error: Some("replica cursor not found; rebootstrap required".to_string()),
175        });
176    };
177
178    let lag_lsn = remote_lsn.checked_sub(cursor.applied_lsn);
179    if !cursor.active {
180        return Ok(ReplicaSyncStatus {
181            replica_id: replica_id.to_string(),
182            active: false,
183            last_applied_lsn: Some(cursor.applied_lsn),
184            remote_lsn,
185            servable_lsn: None,
186            unarchived_lsn: None,
187            lag_lsn,
188            lag_bytes: None,
189            lag_ms: None,
190            stale: true,
191            repair_action: SyncRepairAction::Rebootstrap,
192            last_sync_error: Some("replica cursor is inactive; rebootstrap required".to_string()),
193        });
194    }
195    if cursor.applied_lsn > remote_lsn {
196        return Ok(ReplicaSyncStatus {
197            replica_id: replica_id.to_string(),
198            active: true,
199            last_applied_lsn: Some(cursor.applied_lsn),
200            remote_lsn,
201            servable_lsn: None,
202            unarchived_lsn: None,
203            lag_lsn: None,
204            lag_bytes: None,
205            lag_ms: None,
206            stale: true,
207            repair_action: SyncRepairAction::Rebootstrap,
208            last_sync_error: Some(format!(
209                "replica cursor LSN {} is ahead of remote LSN {remote_lsn}",
210                cursor.applied_lsn
211            )),
212        });
213    }
214    if cursor.applied_lsn == remote_lsn {
215        return Ok(ReplicaSyncStatus {
216            replica_id: replica_id.to_string(),
217            active: true,
218            last_applied_lsn: Some(cursor.applied_lsn),
219            remote_lsn,
220            servable_lsn: Some(remote_lsn),
221            unarchived_lsn: Some(0),
222            lag_lsn: Some(0),
223            lag_bytes: Some(0),
224            lag_ms: Some(0),
225            stale: false,
226            repair_action: SyncRepairAction::None,
227            last_sync_error: None,
228        });
229    }
230
231    let lag_ms = now
232        .saturating_sub(cursor.updated_unix_secs)
233        .saturating_mul(1000);
234    let identity = read_identity(data_dir)?;
235    let segment_dir = retained_segments_dir(data_dir);
236    let progress = retained_tail_progress(
237        &segment_dir,
238        identity.segment_identity(),
239        cursor.applied_lsn,
240        remote_lsn,
241    );
242
243    match progress {
244        Ok(progress) => {
245            let servable_lsn = progress.contiguous_through_lsn;
246            let unarchived_lsn = remote_lsn.saturating_sub(servable_lsn);
247            let (repair_action, last_sync_error) = if servable_lsn > cursor.applied_lsn {
248                (SyncRepairAction::Pull, None)
249            } else {
250                (
251                    SyncRepairAction::AwaitArchive,
252                    Some("primary has advanced, but retained history is not yet archived; retry after archive".to_string()),
253                )
254            };
255            Ok(ReplicaSyncStatus {
256                replica_id: replica_id.to_string(),
257                active: true,
258                last_applied_lsn: Some(cursor.applied_lsn),
259                remote_lsn,
260                servable_lsn: Some(servable_lsn),
261                unarchived_lsn: Some(unarchived_lsn),
262                lag_lsn,
263                lag_bytes: Some(retained_lag_bytes(
264                    &segment_dir,
265                    cursor.applied_lsn,
266                    servable_lsn,
267                )?),
268                lag_ms: Some(lag_ms),
269                stale: true,
270                repair_action,
271                last_sync_error,
272            })
273        }
274        Err(err)
275            if err.kind() == io::ErrorKind::InvalidData
276                && (err.to_string().contains("gap") || err.to_string().contains("missing")) =>
277        {
278            Ok(ReplicaSyncStatus {
279                replica_id: replica_id.to_string(),
280                active: true,
281                last_applied_lsn: Some(cursor.applied_lsn),
282                remote_lsn,
283                servable_lsn: None,
284                unarchived_lsn: None,
285                lag_lsn,
286                lag_bytes: None,
287                lag_ms: Some(lag_ms),
288                stale: true,
289                repair_action: SyncRepairAction::Rebootstrap,
290                last_sync_error: Some(format!(
291                    "retained history is unavailable; rebootstrap required: {err}"
292                )),
293            })
294        }
295        Err(err) => Err(err),
296    }
297}
298
299fn retained_lag_bytes(dir: &Path, applied_lsn: u64, remote_lsn: u64) -> io::Result<u64> {
300    let mut total = 0u64;
301    for file in list_segment_files(dir)? {
302        if file.end_lsn <= applied_lsn || file.start_lsn > remote_lsn {
303            continue;
304        }
305        let len = fs::metadata(file.path)?.len();
306        total = total
307            .checked_add(len)
308            .ok_or_else(|| io::Error::other("retained lag byte count overflow"))?;
309    }
310    Ok(total)
311}
312
313fn invalid_input(message: impl Into<String>) -> io::Error {
314    io::Error::new(io::ErrorKind::InvalidInput, message.into())
315}
316
317#[cfg(test)]
318mod tests {
319    use super::*;
320    use crate::{
321        minimum_retained_lsn, read_replica_cursors, write_identity_snapshot, write_segment_atomic,
322        DatabaseIdentity, IdentitySnapshot, RetainedSegment, RetainedUnit,
323    };
324
325    fn identity() -> DatabaseIdentity {
326        DatabaseIdentity {
327            database_id: *b"replica-status!!",
328            primary_generation: 1,
329        }
330    }
331
332    fn unit(lsn: u64) -> RetainedUnit {
333        RetainedUnit {
334            tx_id: 1,
335            record_type: 4,
336            lsn,
337            data: lsn.to_le_bytes().to_vec(),
338        }
339    }
340
341    fn write_identity(data_dir: &Path) {
342        let snapshot = IdentitySnapshot::from_identity(identity(), 1);
343        write_identity_snapshot(data_dir, &snapshot).unwrap();
344    }
345
346    fn write_segment(data_dir: &Path, start_lsn: u64, end_lsn: u64) {
347        let units = (start_lsn..=end_lsn).map(unit).collect();
348        let segment = RetainedSegment::new(identity().segment_identity(), units).unwrap();
349        write_segment_atomic(&retained_segments_dir(data_dir), &segment).unwrap();
350    }
351
352    fn active_cursor(replica_id: &str, applied_lsn: u64, updated_unix_secs: u64) -> ReplicaCursor {
353        ReplicaCursor {
354            replica_id: replica_id.to_string(),
355            applied_lsn,
356            updated_unix_secs,
357            active: true,
358        }
359    }
360
361    #[test]
362    fn acknowledgement_advances_cursor_and_clears_lag() {
363        let dir = tempfile::tempdir().unwrap();
364        write_identity(dir.path());
365        write_segment(dir.path(), 1, 5);
366        write_segment(dir.path(), 6, 10);
367        crate::upsert_replica_cursor(dir.path(), ReplicaCursor::active("replica-a", 0)).unwrap();
368
369        let ack = acknowledge_replica_apply_at(dir.path(), "replica-a", 10, 10, 200).unwrap();
370
371        assert_eq!(ack.previous_applied_lsn, 0);
372        assert_eq!(ack.applied_lsn, 10);
373        assert!(ack.advanced);
374        assert_eq!(ack.status.repair_action, SyncRepairAction::None);
375        assert!(!ack.status.stale);
376        assert_eq!(ack.status.lag_lsn, Some(0));
377        assert_eq!(ack.status.lag_bytes, Some(0));
378        assert_eq!(minimum_retained_lsn(dir.path()).unwrap(), Some(11));
379
380        let cursor = read_replica_cursors(dir.path())
381            .unwrap()
382            .into_iter()
383            .find(|cursor| cursor.replica_id == "replica-a")
384            .unwrap();
385        assert_eq!(cursor.applied_lsn, 10);
386        assert_eq!(cursor.updated_unix_secs, 200);
387    }
388
389    #[test]
390    fn stale_status_reports_pull_with_lag_bytes() {
391        let dir = tempfile::tempdir().unwrap();
392        write_identity(dir.path());
393        write_segment(dir.path(), 1, 5);
394        write_segment(dir.path(), 6, 10);
395        crate::upsert_replica_cursor(dir.path(), active_cursor("replica-a", 5, 100)).unwrap();
396
397        let status = replica_sync_status_at(dir.path(), "replica-a", 10, 500).unwrap();
398
399        assert!(status.stale);
400        assert_eq!(status.repair_action, SyncRepairAction::Pull);
401        assert_eq!(status.servable_lsn, Some(10));
402        assert_eq!(status.unarchived_lsn, Some(0));
403        assert_eq!(status.lag_lsn, Some(5));
404        assert!(status.lag_bytes.unwrap() > 0);
405        assert!(status.lag_ms.unwrap() > 0);
406        assert_eq!(status.last_sync_error, None);
407    }
408
409    #[test]
410    fn stale_status_reports_pull_for_contiguous_archived_prefix() {
411        let dir = tempfile::tempdir().unwrap();
412        write_identity(dir.path());
413        write_segment(dir.path(), 1, 8);
414        crate::upsert_replica_cursor(dir.path(), active_cursor("replica-a", 5, 100)).unwrap();
415
416        let status = replica_sync_status_at(dir.path(), "replica-a", 10, 500).unwrap();
417
418        assert!(status.stale);
419        assert_eq!(status.repair_action, SyncRepairAction::Pull);
420        assert_eq!(status.servable_lsn, Some(8));
421        assert_eq!(status.unarchived_lsn, Some(2));
422        assert_eq!(status.lag_lsn, Some(5));
423        assert!(status.lag_bytes.unwrap() > 0);
424        assert_eq!(status.last_sync_error, None);
425    }
426
427    #[test]
428    fn stale_status_reports_await_archive_when_primary_is_ahead_of_retained_tail() {
429        let dir = tempfile::tempdir().unwrap();
430        write_identity(dir.path());
431        crate::upsert_replica_cursor(dir.path(), active_cursor("replica-a", 5, 100)).unwrap();
432
433        let status = replica_sync_status_at(dir.path(), "replica-a", 10, 500).unwrap();
434
435        assert!(status.stale);
436        assert_eq!(status.repair_action, SyncRepairAction::AwaitArchive);
437        assert_eq!(status.servable_lsn, Some(5));
438        assert_eq!(status.unarchived_lsn, Some(5));
439        assert_eq!(status.lag_lsn, Some(5));
440        assert_eq!(status.lag_bytes, Some(0));
441        assert!(status.last_sync_error.unwrap().contains("not yet archived"));
442    }
443
444    #[test]
445    fn missing_retained_tail_reports_rebootstrap() {
446        let dir = tempfile::tempdir().unwrap();
447        write_identity(dir.path());
448        crate::upsert_replica_cursor(dir.path(), active_cursor("replica-a", 5, 100)).unwrap();
449        write_segment(dir.path(), 8, 10);
450
451        let status = replica_sync_status_at(dir.path(), "replica-a", 10, 500).unwrap();
452
453        assert!(status.stale);
454        assert_eq!(status.repair_action, SyncRepairAction::Rebootstrap);
455        assert_eq!(status.servable_lsn, None);
456        assert_eq!(status.unarchived_lsn, None);
457        assert_eq!(status.lag_bytes, None);
458        assert!(status
459            .last_sync_error
460            .unwrap()
461            .contains("rebootstrap required"));
462    }
463
464    #[test]
465    fn acknowledgement_rejects_stale_and_inactive_cursors() {
466        let dir = tempfile::tempdir().unwrap();
467        write_identity(dir.path());
468        write_segment(dir.path(), 1, 10);
469        crate::upsert_replica_cursor(dir.path(), ReplicaCursor::active("replica-a", 5)).unwrap();
470
471        let err = acknowledge_replica_apply_at(dir.path(), "replica-a", 4, 10, 100).unwrap_err();
472        assert_eq!(err.kind(), io::ErrorKind::InvalidInput);
473        assert!(err.to_string().contains("behind primary cursor"));
474
475        crate::retire_replica_cursor(dir.path(), "replica-a", 101).unwrap();
476        let err = acknowledge_replica_apply_at(dir.path(), "replica-a", 6, 10, 102).unwrap_err();
477        assert_eq!(err.kind(), io::ErrorKind::InvalidInput);
478        assert!(err.to_string().contains("rebootstrap required"));
479    }
480
481    #[test]
482    fn acknowledgement_status_error_leaves_cursor_unchanged() {
483        let dir = tempfile::tempdir().unwrap();
484        write_identity(dir.path());
485        write_segment(dir.path(), 1, 10);
486        crate::upsert_replica_cursor(dir.path(), active_cursor("replica-a", 5, 100)).unwrap();
487        fs::remove_file(crate::sync_state_dir(dir.path()).join(crate::IDENTITY_FILE)).unwrap();
488
489        let err = acknowledge_replica_apply_at(dir.path(), "replica-a", 6, 10, 200).unwrap_err();
490
491        assert_eq!(err.kind(), io::ErrorKind::NotFound);
492        let cursor = read_replica_cursors(dir.path())
493            .unwrap()
494            .into_iter()
495            .find(|cursor| cursor.replica_id == "replica-a")
496            .unwrap();
497        assert_eq!(cursor.applied_lsn, 5);
498        assert_eq!(cursor.updated_unix_secs, 100);
499    }
500}
powdb_sync/replica.rs

powdb_sync/
replica.rs
