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

//! Segment-statistics and form-profile snapshot persistence.

use super::super::core::EdgeStore;
use super::layout::{file_bytes, form_profile_path, segment_stats_path};
use linkrs_core::{StorageError, StorageResult};
use std::path::Path;

impl EdgeStore {
    /// Persist the checkpoint-collected segment statistics snapshot.
    ///
    /// Written on every checkpoint after the collection step so scans prune
    /// from the same epoch as the topology and property shards. The snapshot
    /// shares the manifest commit point: it lands in a shadow file before
    /// the metadata commit, and a crash before the manifest publish leaves
    /// only the previous consistent snapshot.
    pub(crate) fn flush_segment_stats(
        &self,
        dir: &Path,
        page_size: usize,
        level: i32,
    ) -> StorageResult<u64> {
        use super::super::stats::encode_segment_snapshot;
        let snapshot = encode_segment_snapshot(&self.segment_stats);
        let mut payload = Vec::new();
        crate::persistence::write_header_to(
            &mut payload,
            crate::persistence::section::EDGE_SEGMENT_STATS,
        )
        .map_err(|e| {
            StorageError::io_error(format!("Failed to write segment stats header: {}", e))
        })?;
        payload.extend_from_slice(&(snapshot.len() as u64).to_le_bytes());
        payload.extend_from_slice(&snapshot);
        let path = segment_stats_path(dir);
        super::super::persistence::write_pages_to_file_without_dir_sync(
            &path,
            &payload,
            page_size,
            level,
            self.segment_stats.len() as u32,
        )?;
        Ok(file_bytes(&path))
    }

    /// Load the segment-statistics snapshot, failing closed on section
    /// or trailing mismatches. A missing snapshot fails the load,
    /// never rebuilt silently.
    pub(crate) fn load_segment_stats(&mut self, dir: &Path) -> StorageResult<()> {
        use super::super::stats::decode_segment_snapshot;
        use std::io::Read as _;
        let path = segment_stats_path(dir);
        let (raw, _) = super::super::persistence::read_pages_from_file(&path).map_err(|e| {
            StorageError::deserialize_error(format!(
                "missing segment statistics snapshot at {}: {}",
                path.display(),
                e
            ))
        })?;
        let mut cursor = &raw[..];
        let mut header_buf = [0u8; crate::persistence::HEADER_SIZE];
        cursor.read_exact(&mut header_buf)?;
        {
            let mut slice = &header_buf[..];
            let sid = crate::persistence::read_header(&mut slice)?;
            if sid != crate::persistence::section::EDGE_SEGMENT_STATS {
                return Err(StorageError::deserialize_error(format!(
                    "unexpected section id in segment stats: expected {:#06x}, got {:#06x}",
                    crate::persistence::section::EDGE_SEGMENT_STATS,
                    sid
                )));
            }
        }
        let mut len_bytes = [0u8; 8];
        cursor.read_exact(&mut len_bytes)?;
        let len = u64::from_le_bytes(len_bytes) as usize;
        let mut data = vec![0u8; len];
        cursor.read_exact(&mut data)?;
        if !cursor.is_empty() {
            return Err(StorageError::deserialize_error(
                "unexpected trailing data in segment stats".to_string(),
            ));
        }
        let stats = decode_segment_snapshot(&data)?;
        self.restore_segment_stats(stats);
        Ok(())
    }

    /// Persist the width and access profile snapshot.
    ///
    /// Same commit point as the segment statistics: shadow file before the
    /// manifest publish. Totals only, so the file stays 32 bytes plus the
    /// page header regardless of table size.
    pub(crate) fn flush_form_profile(
        &self,
        dir: &Path,
        page_size: usize,
        level: i32,
    ) -> StorageResult<u64> {
        let snapshot = self.form_profile_snapshot();
        let encoded = snapshot.encode();
        let mut payload = Vec::new();
        crate::persistence::write_header_to(
            &mut payload,
            crate::persistence::section::EDGE_FORM_PROFILE,
        )
        .map_err(|e| {
            StorageError::io_error(format!("Failed to write form profile header: {}", e))
        })?;
        payload.extend_from_slice(&(encoded.len() as u64).to_le_bytes());
        payload.extend_from_slice(&encoded);
        let path = form_profile_path(dir);
        super::super::persistence::write_pages_to_file_without_dir_sync(
            &path, &payload, page_size, level, 1,
        )?;
        Ok(file_bytes(&path))
    }

    /// Load the width and access profile snapshot.
    ///
    /// The snapshot shares the manifest commit point and must exist in every
    /// checkpoint epoch. A missing file means a torn epoch or an old layout
    /// without the profile: old products are rebuilt through a fresh
    /// checkpoint instead of loading through a compat default.
    pub(crate) fn load_form_profile(&self, dir: &Path) -> StorageResult<()> {
        use std::io::Read as _;
        let path = form_profile_path(dir);
        if !path.exists() {
            return Err(StorageError::deserialize_error(format!(
                "missing form profile snapshot at {}: torn write or old layout, rebuild through a fresh checkpoint",
                path.display()
            )));
        }
        let (raw, _) = super::super::persistence::read_pages_from_file(&path).map_err(|e| {
            StorageError::deserialize_error(format!(
                "missing form profile snapshot at {}: {}",
                path.display(),
                e
            ))
        })?;
        let mut cursor = &raw[..];
        let mut header_buf = [0u8; crate::persistence::HEADER_SIZE];
        cursor.read_exact(&mut header_buf)?;
        {
            let mut slice = &header_buf[..];
            let sid = crate::persistence::read_header(&mut slice)?;
            if sid != crate::persistence::section::EDGE_FORM_PROFILE {
                return Err(StorageError::deserialize_error(format!(
                    "unexpected section id in form profile: expected {:#06x}, got {:#06x}",
                    crate::persistence::section::EDGE_FORM_PROFILE,
                    sid
                )));
            }
        }
        let mut len_bytes = [0u8; 8];
        cursor.read_exact(&mut len_bytes)?;
        let len = u64::from_le_bytes(len_bytes) as usize;
        let mut data = vec![0u8; len];
        cursor.read_exact(&mut data)?;
        if !cursor.is_empty() {
            return Err(StorageError::deserialize_error(
                "unexpected trailing data in form profile".to_string(),
            ));
        }
        let snapshot = super::super::stats::FormProfileSnapshot::decode(&data)?;
        self.restore_form_profile(snapshot);
        Ok(())
    }
}