linkrs_storage/edge/single_mutable_csr/
persistence.rs

1//! Persistence for `SingleMutableCsr`: dump and load with CRC32 trailer.
2
3use crate::persistence::read_u64_le;
4use linkrs_core::{StorageError, StorageResult};
5
6use super::super::mutable_csr::serialization::{
7    decode_topology_i64_column, decode_topology_u32_column, decode_topology_u64_column,
8    encode_topology_i64_column, encode_topology_u32_column, encode_topology_u64_column,
9};
10use super::super::{ColdStamps, EdgeId, HotNbr, INVALID_EDGE_ID};
11use super::{SingleMutableCsr, SingleSegment};
12
13impl SingleMutableCsr {
14    /// Dump with integer column encoding for neighbor and edge-id columns.
15    ///
16    /// Offsets are trivial for the single-edge layout (slot index equals row),
17    /// so only neighbor, rank, edge-id and stamp columns go through the
18    /// column path with a narrow plain fallback.
19    pub fn dump(&self) -> Vec<u8> {
20        let mut result = Vec::new();
21        self.dump_into(&mut result);
22        result
23    }
24
25    /// Borrow-based dump into `out`, byte-identical to `dump`. The payload
26    /// carries a trailing CRC32 trailer verified on load.
27    pub fn dump_into(&self, out: &mut Vec<u8>) {
28        let start = out.len();
29        let slot_count = self.vertex_capacity;
30        out.extend_from_slice(&self.edge_count.to_le_bytes());
31        out.extend_from_slice(&(slot_count as u64).to_le_bytes());
32
33        {
34            let endpoints: Vec<u32> = (0..slot_count)
35                .map(|i| self.hot_at(i).map(|h| h.endpoint).unwrap_or(0))
36                .collect();
37            let (_, endpoints_payload) = encode_topology_u32_column(&endpoints);
38            out.extend_from_slice(&endpoints_payload);
39        }
40        {
41            let ranks: Vec<i64> = (0..slot_count)
42                .map(|i| self.hot_at(i).map(|h| h.rank).unwrap_or(0))
43                .collect();
44            let (_, ranks_payload) = encode_topology_i64_column(&ranks);
45            out.extend_from_slice(&ranks_payload);
46        }
47        {
48            let edge_ids: Vec<u64> = (0..slot_count)
49                .map(|i| {
50                    self.hot_at(i)
51                        .map(|h| h.edge_id.0)
52                        .unwrap_or(INVALID_EDGE_ID.0)
53                })
54                .collect();
55            let (_, edge_ids_payload) = encode_topology_u64_column(&edge_ids);
56            out.extend_from_slice(&edge_ids_payload);
57        }
58        {
59            let delete_stamps: Vec<u64> = (0..slot_count)
60                .map(|i| {
61                    self.cold_at(i)
62                        .map(|c| c.delete_ts)
63                        .unwrap_or(ColdStamps::dead_gap().delete_ts)
64                })
65                .collect();
66            let (_, delete_payload) = encode_topology_u64_column(&delete_stamps);
67            out.extend_from_slice(&delete_payload);
68        }
69        let crc = crc32fast::hash(&out[start..]);
70        out.extend_from_slice(&crc.to_le_bytes());
71    }
72
73    /// Reserved slot memory plus struct overhead.
74    ///
75    /// Per-shape accounting by design: the single-slot layout owns no offset
76    /// arrays, live sets or overflow maps, so its shard-level total is
77    /// narrower than the multi-edge layout. Cross-shape comparison belongs at
78    /// the table layer after the shared authority estimate is added, never at
79    /// the shard level directly.
80    pub fn used_memory_size(&self) -> usize {
81        self.sparse_memory_bytes() + std::mem::size_of::<Self>()
82    }
83
84    /// Load the single persisted layout. The trailing
85    /// CRC32 is verified before any parsing.
86    pub fn load(&mut self, data: &[u8]) -> StorageResult<()> {
87        if data.len() < 16 {
88            return Err(StorageError::deserialize_error(
89                "Single CSR data too short for header",
90            ));
91        }
92        let (body, trailer) = data.split_at(data.len() - 4);
93        let mut stored_bytes = [0u8; 4];
94        stored_bytes.copy_from_slice(trailer);
95        let stored = u32::from_le_bytes(stored_bytes);
96        let computed = crc32fast::hash(body);
97        if stored != computed {
98            return Err(StorageError::deserialize_error(format!(
99                "Single CSR dump CRC mismatch: stored={:#x} computed={:#x}",
100                stored, computed
101            )));
102        }
103        let data = body;
104
105        let mut offset = 0usize;
106
107        let edge_count = read_u64_le(data, &mut offset)?;
108        let slot_count = read_u64_le(data, &mut offset)? as usize;
109
110        let endpoints = decode_topology_u32_column(data, &mut offset)?;
111        let ranks = decode_topology_i64_column(data, &mut offset)?;
112        let edge_ids = decode_topology_u64_column(data, &mut offset)?;
113        let delete_stamps = decode_topology_u64_column(data, &mut offset)?;
114        if endpoints.len() != slot_count
115            || ranks.len() != slot_count
116            || edge_ids.len() != slot_count
117            || delete_stamps.len() != slot_count
118        {
119            return Err(StorageError::deserialize_error(
120                "Single CSR column length mismatch",
121            ));
122        }
123        let mut new_csr = SingleMutableCsr::with_capacity(slot_count);
124        let mut recomputed = 0u64;
125        for index in 0..slot_count {
126            let hot = HotNbr {
127                endpoint: endpoints[index],
128                rank: ranks[index],
129                edge_id: EdgeId(edge_ids[index]),
130            };
131            let cold = ColdStamps {
132                delete_ts: delete_stamps[index],
133            };
134            let (seg, off) = Self::locate(index);
135            if hot.edge_id != INVALID_EDGE_ID {
136                let segment =
137                    new_csr.segments[seg].get_or_insert_with(|| Box::new(SingleSegment::fresh()));
138                segment.hot[off] = hot;
139                segment.cold[off] = cold;
140                new_csr.set_present(index, true);
141                if cold.is_live() {
142                    recomputed += 1;
143                }
144            }
145        }
146        new_csr.edge_count = edge_count;
147        if recomputed != edge_count {
148            return Err(StorageError::deserialize_error(format!(
149                "Single CSR edge count mismatch: stored={}, recomputed={}",
150                edge_count, recomputed
151            )));
152        }
153        if offset != data.len() {
154            return Err(StorageError::deserialize_error(
155                "Single CSR has trailing bytes: unsupported format",
156            ));
157        }
158
159        *self = new_csr;
160
161        Ok(())
162    }
163}
linkrs_storage/edge/single_mutable_csr/persistence.rs

linkrs_storage/edge/single_mutable_csr/
persistence.rs
