nodedb 0.2.1

Local-first, real-time, edge-to-cloud hybrid database for multi-modal workloads
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109

// SPDX-License-Identifier: BUSL-1.1

//! Calvin scheduler WAL recovery.
//!
//! Provides [`read_last_applied_epoch`] which scans the WAL for
//! `RecordType::CalvinApplied` records and returns the highest epoch that was
//! successfully committed on a given vshard.  The scheduler uses this on
//! startup to determine the rebuild range (`last_applied + 1 ..= committed`).

use nodedb_wal::record::RecordType;
use nodedb_wal::{CalvinAppliedPayload, WalRecord};
use tracing::warn;

use crate::wal::manager::WalManager;

/// Sentinel used when no Calvin epoch has ever been applied on this vshard.
pub const NOT_YET_APPLIED_EPOCH: u64 = u64::MAX;

/// Scan the WAL and return the highest `epoch` from `CalvinApplied` records
/// for the given vshard.
///
/// Returns [`NOT_YET_APPLIED_EPOCH`] for a greenfield node (no
/// `CalvinApplied` records exist).
///
/// Records that fail to decode are logged and skipped — a corrupt record does
/// not abort the scan.
pub fn read_last_applied_epoch(wal: &WalManager, vshard_id: u32) -> crate::Result<u64> {
    let records = wal.replay()?;
    let mut last_epoch = NOT_YET_APPLIED_EPOCH;

    for record in &records {
        if !is_calvin_applied_record(record) {
            continue;
        }
        match CalvinAppliedPayload::from_bytes(&record.payload) {
            Ok(p) if p.vshard_id == vshard_id => {
                if last_epoch == NOT_YET_APPLIED_EPOCH || p.epoch > last_epoch {
                    last_epoch = p.epoch;
                }
            }
            Ok(_) => {
                // Different vshard — skip.
            }
            Err(e) => {
                warn!(
                    lsn = record.header.lsn,
                    error = %e,
                    "calvin recovery: failed to decode CalvinApplied payload; skipping"
                );
            }
        }
    }

    Ok(last_epoch)
}

fn is_calvin_applied_record(record: &WalRecord) -> bool {
    // Strip the encryption flag (bit 31) before comparing record type.
    let raw_type = record.header.record_type & !nodedb_wal::record::ENCRYPTED_FLAG;
    matches!(
        RecordType::from_raw(raw_type),
        Some(RecordType::CalvinApplied)
    )
}

#[cfg(test)]
mod tests {
    use super::*;
    use tempfile::TempDir;

    use crate::wal::manager::WalManager;

    fn open_wal(dir: &TempDir) -> WalManager {
        WalManager::open(dir.path(), false).expect("open wal")
    }

    #[test]
    fn read_last_applied_epoch_zero_for_greenfield() {
        let dir = TempDir::new().unwrap();
        let wal = open_wal(&dir);
        let epoch = read_last_applied_epoch(&wal, 1).unwrap();
        assert_eq!(
            epoch, NOT_YET_APPLIED_EPOCH,
            "greenfield WAL should return the not-yet-applied sentinel"
        );
    }

    #[test]
    fn read_last_applied_epoch_returns_max_calvin_applied_for_vshard() {
        let dir = TempDir::new().unwrap();
        let wal = open_wal(&dir);

        use crate::types::VShardId;
        // Write three CalvinApplied records for vshard 1 at epochs 2, 5, 3.
        wal.append_calvin_applied(VShardId::new(1), 2, 0).unwrap();
        wal.append_calvin_applied(VShardId::new(1), 5, 0).unwrap();
        wal.append_calvin_applied(VShardId::new(1), 3, 0).unwrap();
        // Write one record for a different vshard (should be ignored).
        wal.append_calvin_applied(VShardId::new(2), 99, 0).unwrap();
        // Flush the WAL write buffer to disk so replay() can read the records.
        wal.sync().unwrap();

        let epoch = read_last_applied_epoch(&wal, 1).unwrap();
        assert_eq!(epoch, 5, "should return the highest epoch for vshard 1");

        let epoch_other = read_last_applied_epoch(&wal, 2).unwrap();
        assert_eq!(epoch_other, 99, "vshard 2 should see its own epoch");
    }
}