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
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191

use linkrs_core::{StorageError, StorageResult};
use linkrs_transaction::wal::Lsn;

use crate::engine::persistence_coordinator::{CheckpointStats, PersistenceState};

#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct SnapshotStats {
    pub snapshot_count: usize,
    pub total_size_bytes: u64,
    pub latest_snapshot_id: Option<u64>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PersistenceDiagnostics {
    pub state: PersistenceState,
    pub checkpoint_sequence: u64,
    pub safe_lsn: Lsn,
    pub last_checkpoint_error: Option<String>,
    pub last_snapshot_error: Option<String>,
    pub temporary_checkpoint_count: usize,
    pub last_checkpoint_stats: Option<CheckpointStats>,
    /// Number of catalog lock acquisitions observed by the storage engine.
    pub catalog_lock_acquisitions: u64,
    /// Total time spent waiting for catalog locks, in nanoseconds.
    pub catalog_lock_wait_nanos: u64,
    /// Total time catalog guards were held, in nanoseconds.
    pub catalog_lock_hold_nanos: u64,
    /// Number of catalog acquisitions that observed measurable contention.
    pub catalog_lock_contentions: u64,
    /// Lock metrics split by catalog operation type.
    pub catalog_lock_by_operation: Vec<CatalogLockDiagnostic>,
    /// Edge-table read lock acquisitions across all partitions.
    pub table_lock_read_acquisitions: u64,
    /// Total time spent waiting for edge-table read locks, in nanoseconds.
    pub table_lock_read_wait_nanos: u64,
    /// Edge-table read acquisitions that observed measurable contention.
    pub table_lock_read_contentions: u64,
    /// Edge-table write lock acquisitions across all partitions.
    pub table_lock_write_acquisitions: u64,
    /// Total time spent waiting for edge-table write locks, in nanoseconds.
    pub table_lock_write_wait_nanos: u64,
    /// Edge-table write acquisitions that observed measurable contention.
    pub table_lock_write_contentions: u64,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CheckpointDiagnostics {
    pub last_stats: Option<CheckpointStats>,
    pub pending_checkpoint: bool,
    pub last_trigger_age_secs: Option<u64>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CatalogLockDiagnostic {
    pub operation: String,
    pub acquisitions: u64,
    pub wait_nanos: u64,
    pub hold_nanos: u64,
    pub contentions: u64,
}

impl crate::engine::persistence_coordinator::PersistenceCoordinator {
    pub fn verify_snapshot(&self, snapshot_id: u64) -> StorageResult<bool> {
        let snapshot_manager = self
            .snapshot_manager
            .as_ref()
            .ok_or_else(|| StorageError::not_supported("Snapshots are not enabled"))?;

        snapshot_manager.verify_snapshot(snapshot_id)
    }

    pub fn cleanup_old_snapshots(&self) -> StorageResult<usize> {
        let snapshot_manager = self
            .snapshot_manager
            .as_ref()
            .ok_or_else(|| StorageError::not_supported("Snapshots are not enabled"))?;

        snapshot_manager.cleanup_old_snapshots()
    }

    /// Remove published checkpoints older than the retention limit while
    /// keeping the newest valid recovery points.
    pub fn cleanup_old_checkpoints(&self, max_checkpoints: usize) -> StorageResult<usize> {
        let keep = max_checkpoints.max(1);
        let current_sequence = self.checkpoint_manager.read().current_seq();
        let retained_by_snapshot = self
            .snapshot_manager
            .as_ref()
            .map(|manager| manager.retained_checkpoint_sequences())
            .unwrap_or_default();
        let mut checkpoints: Vec<(u64, std::path::PathBuf)> =
            std::fs::read_dir(&self.config.checkpoint_dir)?
                .filter_map(Result::ok)
                .filter_map(|entry| {
                    let path = entry.path();
                    let name = path.file_name()?.to_str()?;
                    let sequence = name.strip_prefix("checkpoint_")?.parse::<u64>().ok()?;
                    path.is_dir().then_some((sequence, path))
                })
                .collect();
        checkpoints.sort_by_key(|(sequence, _)| *sequence);
        let remove_count = checkpoints.len().saturating_sub(keep);
        let mut removed = 0;
        for (_sequence, path) in checkpoints.into_iter().filter(|(sequence, _)| {
            *sequence != current_sequence && !retained_by_snapshot.contains(sequence)
        }) {
            if removed >= remove_count {
                break;
            }
            std::fs::remove_dir_all(path)?;
            removed += 1;
        }
        if removed > 0 {
            super::fs_util::sync_directory(&self.config.checkpoint_dir)?;
        }
        Ok(removed)
    }

    pub fn snapshot_stats(&self) -> SnapshotStats {
        if let Some(snapshot_manager) = self.snapshot_manager.as_ref() {
            SnapshotStats {
                snapshot_count: snapshot_manager.snapshot_count(),
                total_size_bytes: snapshot_manager.total_snapshot_size(),
                latest_snapshot_id: snapshot_manager.get_latest_snapshot().map(|info| info.id),
            }
        } else {
            SnapshotStats::default()
        }
    }

    pub fn diagnostics(&self) -> PersistenceDiagnostics {
        let temporary_checkpoint_count = std::fs::read_dir(&self.config.checkpoint_dir)
            .map(|entries| {
                entries
                    .filter_map(Result::ok)
                    .filter(|entry| {
                        entry
                            .file_name()
                            .to_str()
                            .is_some_and(|name| name.ends_with(".tmp"))
                    })
                    .count()
            })
            .unwrap_or(0);
        let manifest_safe_lsn = self
            .manifest_manager
            .latest_safe_lsn()
            .map(|lsn| Lsn::new(lsn.get()))
            .unwrap_or_else(|_| *self.last_checkpoint_lsn.read());
        PersistenceDiagnostics {
            state: *self.state.read(),
            checkpoint_sequence: self.checkpoint_manager.read().current_seq(),
            safe_lsn: manifest_safe_lsn,
            last_checkpoint_error: self.last_checkpoint_error.read().clone(),
            last_snapshot_error: self.last_snapshot_error.read().clone(),
            temporary_checkpoint_count,
            last_checkpoint_stats: self.last_checkpoint_stats.read().clone(),
            catalog_lock_acquisitions: 0,
            catalog_lock_wait_nanos: 0,
            catalog_lock_hold_nanos: 0,
            catalog_lock_contentions: 0,
            catalog_lock_by_operation: Vec::new(),
            table_lock_read_acquisitions: 0,
            table_lock_read_wait_nanos: 0,
            table_lock_read_contentions: 0,
            table_lock_write_acquisitions: 0,
            table_lock_write_wait_nanos: 0,
            table_lock_write_contentions: 0,
        }
    }

    pub fn mark_flushed(&self, lsn: Lsn) {
        *self.last_flush_lsn.write() = lsn;
        *self.last_flush_time.write() = std::time::Instant::now();
    }

    pub fn mark_checkpointed(&self, lsn: Lsn) {
        *self.last_checkpoint_lsn.write() = lsn;
        *self.last_checkpoint_time.write() = std::time::Instant::now();
        *self.last_flush_lsn.write() = lsn;
        *self.last_flush_time.write() = std::time::Instant::now();
    }

    pub fn checkpoint_diagnostics(&self, pending_checkpoint: bool) -> CheckpointDiagnostics {
        CheckpointDiagnostics {
            last_stats: self.last_checkpoint_stats.read().clone(),
            pending_checkpoint,
            last_trigger_age_secs: Some(self.last_checkpoint_time.read().elapsed().as_secs()),
        }
    }
}