cfx_storage/impls/snapshot_sync/restoration/
full_sync_verifier.rs

1// Copyright 2019 Conflux Foundation. All rights reserved.
2// Conflux is free software and distributed under GNU General Public License.
3// See http://www.gnu.org/licenses/
4
5pub struct FullSyncVerifier<SnapshotDbManager: SnapshotDbManagerTrait> {
6    number_chunks: usize,
7    merkle_root: MerkleHash,
8    chunk_boundaries: Vec<Vec<u8>>,
9    chunk_boundary_proofs: Vec<TrieProof>,
10    chunk_verified: Vec<bool>,
11    number_incomplete_chunk: usize,
12
13    pending_boundary_nodes: HashMap<CompressedPathRaw, SnapshotMptNode>,
14    boundary_subtree_total_size: HashMap<BoundarySubtreeIndex, u64>,
15    chunk_index_by_upper_key: HashMap<Vec<u8>, usize>,
16
17    temp_snapshot_db: SnapshotDbManager::SnapshotDbWrite,
18}
19
20impl<SnapshotDbManager: SnapshotDbManagerTrait>
21    FullSyncVerifier<SnapshotDbManager>
22{
23    pub fn new(
24        number_chunks: usize, chunk_boundaries: Vec<Vec<u8>>,
25        chunk_boundary_proofs: Vec<TrieProof>, merkle_root: MerkleHash,
26        snapshot_db_manager: &SnapshotDbManager, epoch_id: &EpochId,
27        epoch_height: u64,
28    ) -> Result<Self> {
29        if number_chunks != chunk_boundaries.len() + 1 {
30            bail!(Error::InvalidSnapshotSyncProof)
31        }
32        if number_chunks != chunk_boundary_proofs.len() + 1 {
33            bail!(Error::InvalidSnapshotSyncProof)
34        }
35        let mut chunk_index_by_upper_key = HashMap::new();
36        let mut prev_boundary: Option<&[u8]> = None;
37        for (chunk_index, (chunk_boundary, proof)) in chunk_boundaries
38            .iter()
39            .zip(chunk_boundary_proofs.iter())
40            .enumerate()
41        {
42            // Strictly increasing: `chunk_index_by_upper_key` is keyed by
43            // boundary and `restore_chunk` derives ranges from adjacent
44            // boundaries, so a duplicate would silently collapse chunk indexes.
45            if let Some(prev) = prev_boundary {
46                if chunk_boundary.as_slice() <= prev {
47                    bail!(Error::InvalidSnapshotSyncProof)
48                }
49            }
50            prev_boundary = Some(chunk_boundary.as_slice());
51            // Reject over-long boundary keys before they rebuild proof paths
52            // during restore (see CompressedPathRaw::MAX_PATH_BYTES).
53            if chunk_boundary.len() > CompressedPathRaw::MAX_PATH_BYTES {
54                bail!(Error::InvalidSnapshotSyncProof)
55            }
56            if merkle_root.ne(proof.get_merkle_root()) {
57                bail!(Error::InvalidSnapshotSyncProof)
58            }
59            // We don't want the proof to carry extra nodes.
60            if proof.number_leaf_nodes() != 1 {
61                bail!(Error::InvalidSnapshotSyncProof)
62            }
63            if proof.if_proves_key(&*chunk_boundary)
64                != (true, proof.get_proof_nodes().last())
65            {
66                bail!(Error::InvalidSnapshotSyncProof)
67            }
68            chunk_index_by_upper_key
69                .insert(chunk_boundary.clone(), chunk_index);
70        }
71
72        Ok(Self {
73            number_chunks,
74            merkle_root,
75            chunk_boundaries,
76            chunk_boundary_proofs,
77            chunk_verified: vec![false; number_chunks],
78            number_incomplete_chunk: number_chunks,
79            pending_boundary_nodes: Default::default(),
80            boundary_subtree_total_size: Default::default(),
81            chunk_index_by_upper_key,
82            temp_snapshot_db: snapshot_db_manager
83                .new_temp_snapshot_for_full_sync(
84                    epoch_id,
85                    &merkle_root,
86                    epoch_height,
87                )?,
88        })
89    }
90
91    pub fn is_completed(&self) -> bool { self.number_incomplete_chunk == 0 }
92
93    // FIXME: multi-threading, where &mut can be dropped.
94    pub fn restore_chunk<Key: Borrow<[u8]> + Debug>(
95        &mut self, chunk_upper_key: &Option<Vec<u8>>, keys: &Vec<Key>,
96        values: Vec<Vec<u8>>,
97    ) -> Result<bool> {
98        let chunk_index = match chunk_upper_key {
99            None => self.number_chunks - 1,
100            Some(upper_key) => {
101                match self.chunk_index_by_upper_key.get(upper_key) {
102                    Some(index) => *index,
103                    None => {
104                        warn!("chunk key {:?} does not match boundaries in manifest", upper_key);
105                        return Ok(false);
106                    }
107                }
108            }
109        };
110        // Reject over-long keys before they become compressed paths during
111        // restore (see CompressedPathRaw::MAX_PATH_BYTES).
112        for key in keys {
113            if key.borrow().len() > CompressedPathRaw::MAX_PATH_BYTES {
114                warn!("chunk contains an over-long key, rejecting chunk");
115                return Ok(false);
116            }
117        }
118
119        // Check key monotone.
120        if !keys.is_empty() {
121            let mut previous = keys.first().unwrap();
122            for key in &keys[1..] {
123                if key.borrow().le(previous.borrow()) {
124                    warn!("chunk key not in order");
125                    return Ok(false);
126                }
127                previous = key;
128            }
129        }
130
131        let key_range_left;
132        let maybe_key_range_right_excl;
133        let maybe_left_proof;
134        let maybe_right_proof;
135        if chunk_index == 0 {
136            key_range_left = vec![];
137            maybe_left_proof = None;
138        } else {
139            key_range_left = self.chunk_boundaries[chunk_index - 1].clone();
140            maybe_left_proof = self.chunk_boundary_proofs.get(chunk_index - 1);
141
142            // Check key boundary.
143            if let Some(first_key) = keys.first() {
144                if first_key.borrow().lt(&*key_range_left) {
145                    warn!(
146                        "first chunk key {:?} less than left range {:?}",
147                        first_key, key_range_left
148                    );
149                    return Ok(false);
150                }
151            }
152        };
153        if chunk_index == self.number_chunks - 1 {
154            maybe_key_range_right_excl = None;
155            maybe_right_proof = None;
156        } else {
157            let key_range_right_excl =
158                self.chunk_boundaries[chunk_index].clone();
159            maybe_right_proof = self.chunk_boundary_proofs.get(chunk_index);
160
161            // Check key boundary.
162            if let Some(last_key) = keys.last() {
163                if last_key.borrow().ge(&*key_range_right_excl) {
164                    warn!(
165                        "last chunk key {:?} larger than left range {:?}",
166                        last_key, key_range_right_excl,
167                    );
168                    return Ok(false);
169                }
170            }
171
172            maybe_key_range_right_excl = Some(key_range_right_excl);
173        }
174
175        // FIXME: multi-threading.
176        // Restore.
177        let chunk_verifier = MptSliceVerifier::new(
178            maybe_left_proof,
179            &*key_range_left,
180            maybe_right_proof,
181            maybe_key_range_right_excl.as_ref().map(|v| &**v),
182            self.merkle_root.clone(),
183        );
184
185        let chunk_rebuilder = chunk_verifier.restore(keys, &values)?;
186        if chunk_rebuilder.is_valid {
187            self.chunk_verified[chunk_index] = true;
188            self.number_incomplete_chunk -= 1;
189
190            self.temp_snapshot_db.start_transaction()?;
191            // Commit key-values.
192            for (key, value) in keys.into_iter().zip(values.into_iter()) {
193                self.temp_snapshot_db.put_kv(key.borrow(), &*value)?;
194            }
195
196            // Commit inner nodes.
197            let mut snapshot_mpt =
198                self.temp_snapshot_db.open_snapshot_mpt_owned()?;
199            for (path, node) in chunk_rebuilder.inner_nodes_to_write {
200                snapshot_mpt.write_node(&path, &node)?;
201            }
202            drop(snapshot_mpt);
203            self.temp_snapshot_db.commit_transaction()?;
204
205            // Combine changes around boundary nodes.
206            for (path, node) in chunk_rebuilder.boundary_nodes {
207                let mut children_table = VanillaChildrenTable::default();
208                unsafe {
209                    for (child_index, merkle_ref) in
210                        node.get_children_table_ref().iter()
211                    {
212                        *children_table.get_child_mut_unchecked(child_index) =
213                            SubtreeMerkleWithSize {
214                                merkle: *merkle_ref,
215                                subtree_size: 0,
216                                delta_subtree_size: 0,
217                            }
218                    }
219                    *children_table.get_children_count_mut() =
220                        node.get_children_count();
221                }
222                self.pending_boundary_nodes.insert(
223                    path,
224                    SnapshotMptNode(VanillaTrieNode::new(
225                        node.get_merkle().clone(),
226                        children_table,
227                        node.value_as_slice()
228                            .into_option()
229                            .map(|ref_v| ref_v.into()),
230                        node.compressed_path_ref().into(),
231                    )),
232                );
233            }
234            for (subtree_index, subtree_size) in
235                chunk_rebuilder.boundary_subtree_total_size
236            {
237                *self
238                    .boundary_subtree_total_size
239                    .entry(subtree_index)
240                    .or_default() += subtree_size;
241            }
242        }
243
244        if self.is_completed() {
245            self.finalize()?
246        }
247
248        Ok(chunk_rebuilder.is_valid)
249    }
250
251    // FIXME: multi-threading
252    /// Combine and write boundary subtree nodes after all chunks have been
253    /// completed.
254    pub fn finalize(&mut self) -> Result<()> {
255        self.temp_snapshot_db.start_transaction()?;
256        let mut snapshot_mpt =
257            self.temp_snapshot_db.open_snapshot_mpt_owned()?;
258
259        for (path, mut node) in self.pending_boundary_nodes.drain() {
260            let mut subtree_index = BoundarySubtreeIndex {
261                parent_node: node.get_merkle().clone(),
262                child_index: 0,
263            };
264            for child_index in 0..CHILDREN_COUNT as u8 {
265                subtree_index.child_index = child_index;
266                if let Some(subtree_size) =
267                    self.boundary_subtree_total_size.get(&subtree_index)
268                {
269                    // Actually safe.
270                    unsafe {
271                        node.get_child_mut_unchecked(child_index)
272                            .subtree_size = *subtree_size;
273                    }
274                }
275            }
276
277            snapshot_mpt.write_node(&path, &node)?;
278        }
279
280        drop(snapshot_mpt);
281        self.temp_snapshot_db.commit_transaction()?;
282        Ok(())
283    }
284}
285
286use crate::{
287    impls::{
288        errors::*,
289        merkle_patricia_trie::{
290            trie_node::TrieNodeTrait, CompressedPathRaw, VanillaChildrenTable,
291            VanillaTrieNode, CHILDREN_COUNT,
292        },
293        snapshot_sync::restoration::mpt_slice_verifier::{
294            BoundarySubtreeIndex, MptSliceVerifier,
295        },
296    },
297    storage_db::{
298        SnapshotDbManagerTrait, SnapshotDbWriteableTrait, SnapshotMptNode,
299        SnapshotMptTraitRw, SubtreeMerkleWithSize,
300    },
301    TrieProof,
302};
303use primitives::{EpochId, MerkleHash};
304use std::{borrow::Borrow, collections::HashMap, fmt::Debug};
cfx_storage/impls/snapshot_sync/restoration/full_sync_verifier.rs

cfx_storage/impls/snapshot_sync/restoration/
full_sync_verifier.rs
