linkrs-storage 0.1.0

Storage engine with CSR and memory-mapped containers
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

use std::sync::Arc;

use super::GraphStorageContext;

impl GraphStorageContext {
    pub(crate) fn migration_history(
        &self,
    ) -> &Arc<parking_lot::RwLock<crate::migration_history::MigrationHistoryManager>> {
        &self.persistent.migration_history
    }

    pub fn record_migration_history(
        &self,
        record: crate::migration_history::MigrationHistoryRecord,
    ) -> linkrs_core::StorageResult<()> {
        let mut mgr = self.persistent.migration_history.write();
        mgr.record(record.clone())?;
        if let Some(paths) = self.storage_paths() {
            let path = paths.migration_history_file();
            if let Err(e) = mgr.save_to_file(&path) {
                log::warn!("Failed to persist migration_history: {}", e);
            }
        }
        Ok(())
    }

    pub fn list_migration_history(
        &self,
        space: &str,
        label: &str,
        is_edge: bool,
    ) -> Vec<crate::migration_history::MigrationHistoryRecord> {
        self.persistent
            .migration_history
            .read()
            .list(space, label, is_edge)
    }

    pub fn list_all_migration_history(
        &self,
    ) -> Vec<crate::migration_history::MigrationHistoryRecord> {
        self.persistent.migration_history.read().list_all()
    }

    pub fn get_applied_versions(&self, space: &str, label: &str, is_edge: bool) -> Vec<u64> {
        self.persistent
            .migration_history
            .read()
            .get_applied_versions_sorted(space, label, is_edge)
    }

    pub(crate) fn load_migration_history(&self) -> linkrs_core::StorageResult<()> {
        if let Some(work_dir) = self.work_dir().as_ref().cloned() {
            use crate::engine::paths::StoragePaths;
            use linkrs_sync::checkpoint_manifest::CheckpointManifestManager;
            let checkpoint_root = work_dir.join("checkpoint");
            let mut candidate_paths = Vec::new();
            if checkpoint_root.exists() {
                if let Ok(Some(manifest)) =
                    CheckpointManifestManager::new(checkpoint_root.join("manifests")).load_latest()
                {
                    candidate_paths.push(
                        StoragePaths::new(manifest.storage_snapshot.path).migration_history_file(),
                    );
                }
            }
            if let Some(paths) = self.storage_paths() {
                candidate_paths.push(paths.migration_history_file());
            }
            for path in candidate_paths {
                if path.exists() {
                    let mgr =
                        crate::migration_history::MigrationHistoryManager::load_from_file(&path)?;
                    *self.persistent.migration_history.write() = mgr;
                    break;
                }
            }
        } else if let Some(paths) = self.storage_paths() {
            let path = paths.migration_history_file();
            if path.exists() {
                let mgr = crate::migration_history::MigrationHistoryManager::load_from_file(&path)?;
                *self.persistent.migration_history.write() = mgr;
            }
        }
        Ok(())
    }

    pub(crate) fn save_migration_history(&self) -> linkrs_core::StorageResult<()> {
        if let Some(paths) = self.storage_paths() {
            let path = paths.migration_history_file();
            self.persistent
                .migration_history
                .read()
                .save_to_file(&path)?;
        }
        Ok(())
    }
}