nodedb_graph/csr/rebuild/
install.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Finishing a rebuild on the owning thread: restore the compacted copy
4//! and replay the journal onto it.
5
6use nodedb_mem::ScopedMemory;
7
8use super::seed::{CsrRebuilt, NodeLabelState, restore_checkpoint};
9use crate::GraphError;
10use crate::csr::index::CsrIndex;
11
12/// Most node labels one index interns. The label bitset is a `u64`.
13const MAX_NODE_LABELS: usize = 64;
14
15impl CsrIndex {
16    /// Close the journal of `rebuilt` and return the copy that replaces
17    /// this index.
18    ///
19    /// The copy holds the snapshot, compacted, plus every mutation this
20    /// index took since `begin_rebuild`, in order. It keeps this index's
21    /// partition tag, so node ids handed out before the swap stay valid.
22    /// The caller installs it in one step. On error the journal is closed,
23    /// this index stays as it is, and the copy is dropped:
24    ///
25    /// - [`GraphError::RebuildSuperseded`]: no journal of this rebuild is open.
26    /// - [`GraphError::RebuildJournalOverflow`]: the journal hit its bound.
27    /// - [`GraphError::RebuildReplayDiverged`]: a replayed mutation returned
28    ///   another outcome than it did here.
29    /// - [`GraphError::RebuildSnapshotInvalid`]: the copy does not decode.
30    pub fn finish_rebuild(
31        &mut self,
32        rebuilt: CsrRebuilt,
33        memory: ScopedMemory,
34    ) -> Result<CsrIndex, GraphError> {
35        let journal = match self.rebuild_journal.take() {
36            Some(journal) if journal.token == rebuilt.token => journal,
37            other => {
38                self.rebuild_journal = other;
39                return Err(GraphError::RebuildSuperseded);
40            }
41        };
42        let ops = journal.into_ops()?;
43        let mut copy = restore_checkpoint(&rebuilt.checkpoint, memory)?;
44        copy.install_node_labels(rebuilt.node_labels)?;
45        for (op, live_outcome) in &ops {
46            if copy.replay_op(op) != *live_outcome {
47                return Err(GraphError::RebuildReplayDiverged { op: op.kind() });
48            }
49        }
50        copy.partition_tag = self.partition_tag;
51        Ok(copy)
52    }
53
54    /// Put the node labels a rebuild carried onto a freshly restored copy.
55    fn install_node_labels(&mut self, labels: NodeLabelState) -> Result<(), GraphError> {
56        if labels.bits.len() != self.id_to_node.len() {
57            return Err(GraphError::RebuildSnapshotInvalid {
58                detail: format!(
59                    "node label bitsets cover {} nodes, the snapshot holds {}",
60                    labels.bits.len(),
61                    self.id_to_node.len()
62                ),
63            });
64        }
65        if labels.names.len() > MAX_NODE_LABELS {
66            return Err(GraphError::RebuildSnapshotInvalid {
67                detail: format!(
68                    "{} node labels exceed the {MAX_NODE_LABELS}-label bitset",
69                    labels.names.len()
70                ),
71            });
72        }
73        self.node_label_to_id = labels
74            .names
75            .iter()
76            .enumerate()
77            .map(|(id, name)| (name.clone(), id as u8))
78            .collect();
79        self.node_label_names = labels.names;
80        self.node_label_bits = labels.bits;
81        Ok(())
82    }
83}
84
85#[cfg(test)]
86mod tests {
87    use nodedb_types::Surrogate;
88
89    use crate::GraphError;
90    use crate::csr::index::{CsrIndex, Direction};
91    use crate::test_support::test_memory;
92
93    const JOURNAL_BYTES: usize = 1 << 20;
94
95    fn seeded() -> CsrIndex {
96        let mut csr = CsrIndex::new(test_memory());
97        csr.add_edge_in_collection("a", "L", "b", "c").unwrap();
98        csr.add_edge_in_collection("b", "L", "c", "c").unwrap();
99        csr.add_node_label("a", "Person").unwrap();
100        csr.set_node_surrogate("a", Surrogate::new(7));
101        csr
102    }
103
104    fn out_of(csr: &CsrIndex, node: &str) -> Vec<String> {
105        let mut dsts: Vec<String> = csr
106            .neighbors(node, &[], Direction::Out)
107            .into_iter()
108            .map(|(_, d)| d)
109            .collect();
110        dsts.sort();
111        dsts
112    }
113
114    #[test]
115    fn writes_during_the_build_reach_the_installed_copy() {
116        let mut live = seeded();
117        let seed = live.begin_rebuild(JOURNAL_BYTES).unwrap();
118
119        // Writes that land after the snapshot.
120        live.add_edge_in_collection("a", "L", "d", "c").unwrap();
121        live.remove_edge_in_collection("b", "L", "c", "c");
122        live.put_edge_in_collection("x", "L", "y", "c", 2.5)
123            .unwrap();
124        live.add_node_label("x", "Person").unwrap();
125        live.set_node_surrogate("x", Surrogate::new(9));
126
127        let rebuilt = seed.build(test_memory()).unwrap();
128        let tag_before = live.partition_tag;
129        let copy = live.finish_rebuild(rebuilt, test_memory()).unwrap();
130
131        assert_eq!(out_of(&copy, "a"), vec!["b".to_string(), "d".to_string()]);
132        assert!(
133            out_of(&copy, "b").is_empty(),
134            "a delete during the build carries over"
135        );
136        assert_eq!(
137            copy.edge_weight_in_collection("x", "L", "y", "c"),
138            Some(2.5)
139        );
140        let x = copy.node_id_raw("x").unwrap();
141        assert!(copy.node_has_label(x, "Person"));
142        let a = copy.node_id_raw("a").unwrap();
143        assert!(copy.node_has_label(a, "Person"), "snapshot labels survive");
144        assert_eq!(copy.node_id_for_surrogate(Surrogate::new(9)), Some("x"));
145        assert_eq!(copy.partition_tag, tag_before);
146        assert!(!live.rebuild_in_progress(), "finishing closes the journal");
147    }
148
149    #[test]
150    fn a_result_from_another_rebuild_is_refused() {
151        let mut live = seeded();
152        let seed = live.begin_rebuild(JOURNAL_BYTES).unwrap();
153        live.abort_rebuild(seed.token());
154        let _second = live.begin_rebuild(JOURNAL_BYTES).unwrap();
155        let rebuilt = seed.build(test_memory()).unwrap();
156        assert!(matches!(
157            live.finish_rebuild(rebuilt, test_memory()),
158            Err(GraphError::RebuildSuperseded)
159        ));
160        assert!(live.rebuild_in_progress(), "the newer journal stays open");
161    }
162
163    #[test]
164    fn a_second_rebuild_is_refused_while_one_runs() {
165        let mut live = seeded();
166        let _seed = live.begin_rebuild(JOURNAL_BYTES).unwrap();
167        assert!(matches!(
168            live.begin_rebuild(JOURNAL_BYTES),
169            Err(GraphError::RebuildInProgress)
170        ));
171    }
172
173    #[test]
174    fn journal_overflow_refuses_the_copy_and_keeps_the_live_writes() {
175        let mut live = seeded();
176        let seed = live.begin_rebuild(1).unwrap();
177        live.add_edge_in_collection("a", "L", "z", "c").unwrap();
178        let rebuilt = seed.build(test_memory()).unwrap();
179        assert!(matches!(
180            live.finish_rebuild(rebuilt, test_memory()),
181            Err(GraphError::RebuildJournalOverflow { cap_bytes: 1 })
182        ));
183        assert!(out_of(&live, "a").contains(&"z".to_string()));
184    }
185}
nodedb_graph/csr/rebuild/install.rs

nodedb_graph/csr/rebuild/
install.rs
