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