diff --git a/.github/actions/setup-rust/action.yml b/.github/actions/setup-rust/action.yml index f560363..5c36ca3 100644 --- a/.github/actions/setup-rust/action.yml +++ b/.github/actions/setup-rust/action.yml @@ -2,20 +2,11 @@ name: Setup Rust (cache & toolchain) runs: using: composite steps: - - name: Cargo cache - uses: actions/cache@v3 - with: - path: | - ~/.cargo/bin/ - ~/.cargo/registry/index/ - ~/.cargo/registry/cache/ - ~/.cargo/git/db/ - target/ - key: ${{ runner.os }}-cargo-${{ hashFiles('**/Cargo.lock') }} - - name: Install toolchain 1.75.0 uses: actions-rs/toolchain@v1 with: profile: minimal toolchain: 1.75.0 - components: rustfmt, clippy \ No newline at end of file + components: rustfmt, clippy + + - uses: Swatinem/rust-cache@v2 \ No newline at end of file diff --git a/.github/workflows/cc.yml b/.github/workflows/cc.yml index 5368b1c..c657505 100644 --- a/.github/workflows/cc.yml +++ b/.github/workflows/cc.yml @@ -15,6 +15,19 @@ jobs: runs-on: ubuntu-latest steps: + - name: Free Disk Space (Ubuntu) + uses: jlumbroso/free-disk-space@main + with: + # this might remove tools that are actually needed, + # if set to "true" but frees about 6 GB + tool-cache: false + android: true + dotnet: true + haskell: true + large-packages: false + docker-images: true + swap-storage: true + - name: Checkout sources uses: actions/checkout@v3 with: diff --git a/common/append_merkle/src/lib.rs b/common/append_merkle/src/lib.rs index f9d7fb9..d206239 100644 --- a/common/append_merkle/src/lib.rs +++ b/common/append_merkle/src/lib.rs @@ -88,6 +88,10 @@ impl> AppendMerkleTree { for (index, h) in initial_data.known_leaves { merkle.fill_leaf(index, h); } + for (layer_index, position, h) in initial_data.extra_mpt_nodes { + // TODO: Delete duplicate nodes from DB. + merkle.layers[layer_index][position] = h; + } Ok(merkle) } @@ -192,7 +196,10 @@ impl> AppendMerkleTree { /// This requires that the proof is built against this tree. /// This should only be called after validating the proof (including checking root existence). /// Returns `Error` if the data is conflict with existing ones. - pub fn fill_with_range_proof(&mut self, proof: RangeProof) -> Result<()> { + pub fn fill_with_range_proof( + &mut self, + proof: RangeProof, + ) -> Result> { self.fill_with_proof( proof .left_proof @@ -212,7 +219,7 @@ impl> AppendMerkleTree { proof: Proof, mut tx_merkle_nodes: Vec<(usize, E)>, start_index: u64, - ) -> Result<()> { + ) -> Result> { let tx_merkle_nodes_size = tx_merkle_nodes.len(); if self.leaf_height != 0 { tx_merkle_nodes = tx_merkle_nodes @@ -227,7 +234,7 @@ impl> AppendMerkleTree { .collect(); } if tx_merkle_nodes.is_empty() { - return Ok(()); + return Ok(Vec::new()); } let mut position_and_data = proof.file_proof_nodes_in_tree(tx_merkle_nodes, tx_merkle_nodes_size); @@ -239,7 +246,12 @@ impl> AppendMerkleTree { } /// This assumes that the proof leaf is no lower than the tree leaf. It holds for both SegmentProof and ChunkProof. - fn fill_with_proof(&mut self, position_and_data: Vec<(usize, E)>) -> Result<()> { + /// Return the inserted nodes and position. + fn fill_with_proof( + &mut self, + position_and_data: Vec<(usize, E)>, + ) -> Result> { + let mut updated_nodes = Vec::new(); // A valid proof should not fail the following checks. for (i, (position, data)) in position_and_data.into_iter().enumerate() { let layer = &mut self.layers[i]; @@ -254,7 +266,10 @@ impl> AppendMerkleTree { // skip padding node. continue; } - if layer[position] != E::null() && layer[position] != data { + if layer[position] == E::null() { + layer[position] = data.clone(); + updated_nodes.push((i, position, data)) + } else if layer[position] != data { bail!( "conflict data layer={} position={} tree_data={:?} proof_data={:?}", i, @@ -263,9 +278,8 @@ impl> AppendMerkleTree { data ); } - layer[position] = data; } - Ok(()) + Ok(updated_nodes) } pub fn gen_range_proof(&self, start_index: usize, end_index: usize) -> Result> { diff --git a/common/append_merkle/src/merkle_tree.rs b/common/append_merkle/src/merkle_tree.rs index 11a97af..0dd928e 100644 --- a/common/append_merkle/src/merkle_tree.rs +++ b/common/append_merkle/src/merkle_tree.rs @@ -127,6 +127,11 @@ pub struct MerkleTreeInitialData { /// These leaves are in some large subtrees of `subtree_list`. 1-node subtrees are also leaves, /// but they will not be duplicated in `known_leaves`. pub known_leaves: Vec<(usize, E)>, + + /// A list of `(layer_index, position, hash)`. + /// These are the nodes known from proofs. + /// They should only be inserted after constructing the tree. + pub extra_mpt_nodes: Vec<(usize, usize, E)>, } impl MerkleTreeInitialData { diff --git a/node/storage/src/log_store/flow_store.rs b/node/storage/src/log_store/flow_store.rs index 412286d..3933067 100644 --- a/node/storage/src/log_store/flow_store.rs +++ b/node/storage/src/log_store/flow_store.rs @@ -1,7 +1,9 @@ use super::load_chunk::EntryBatch; use super::{MineLoadChunk, SealAnswer, SealTask}; use crate::error::Error; -use crate::log_store::log_manager::{bytes_to_entries, COL_ENTRY_BATCH, COL_ENTRY_BATCH_ROOT}; +use crate::log_store::log_manager::{ + bytes_to_entries, COL_ENTRY_BATCH, COL_ENTRY_BATCH_ROOT, COL_FLOW_MPT_NODES, +}; use crate::log_store::{FlowRead, FlowSeal, FlowWrite}; use crate::{try_option, ZgsKeyValueDB}; use anyhow::{anyhow, bail, Result}; @@ -70,6 +72,10 @@ impl FlowStore { })?; merkle.gen_proof(sector_index) } + + pub fn put_mpt_node_list(&self, node_list: Vec<(usize, usize, DataRoot)>) -> Result<()> { + self.db.put_mpt_node_list(node_list) + } } #[derive(Clone, Debug)] @@ -441,9 +447,11 @@ impl FlowDBStore { ); } } + let extra_node_list = self.get_mpt_node_list()?; Ok(MerkleTreeInitialData { subtree_list: root_list, known_leaves: leaf_list, + extra_mpt_nodes: extra_node_list, }) } @@ -492,6 +500,32 @@ impl FlowDBStore { self.kvdb.write(tx)?; Ok(index_to_reseal) } + + fn put_mpt_node_list(&self, mpt_node_list: Vec<(usize, usize, DataRoot)>) -> Result<()> { + let mut tx = self.kvdb.transaction(); + for (layer_index, position, data) in mpt_node_list { + tx.put( + COL_FLOW_MPT_NODES, + &encode_mpt_node_key(layer_index, position), + data.as_bytes(), + ); + } + Ok(self.kvdb.write(tx)?) + } + + fn get_mpt_node_list(&self) -> Result> { + let mut node_list = Vec::new(); + for r in self.kvdb.iter(COL_FLOW_MPT_NODES) { + let (index_bytes, node_bytes) = r?; + let (layer_index, position) = decode_mpt_node_key(index_bytes.as_ref())?; + node_list.push(( + layer_index, + position, + DataRoot::from_slice(node_bytes.as_ref()), + )); + } + Ok(node_list) + } } #[derive(DeriveEncode, DeriveDecode, Clone, Debug)] @@ -537,3 +571,18 @@ fn decode_batch_root_key(data: &[u8]) -> Result<(usize, usize)> { let subtree_depth = usize::MAX - try_decode_usize(&data[mem::size_of::()..])?; Ok((batch_index, subtree_depth)) } + +fn encode_mpt_node_key(layer_index: usize, position: usize) -> Vec { + let mut key = layer_index.to_be_bytes().to_vec(); + key.extend_from_slice(&position.to_be_bytes()); + key +} + +fn decode_mpt_node_key(data: &[u8]) -> Result<(usize, usize)> { + if data.len() != mem::size_of::() * 2 { + bail!("invalid data length"); + } + let layer_index = try_decode_usize(&data[..mem::size_of::()])?; + let position = try_decode_usize(&data[mem::size_of::()..])?; + Ok((layer_index, position)) +} diff --git a/node/storage/src/log_store/log_manager.rs b/node/storage/src/log_store/log_manager.rs index 115d6f3..eec1558 100644 --- a/node/storage/src/log_store/log_manager.rs +++ b/node/storage/src/log_store/log_manager.rs @@ -36,7 +36,8 @@ pub const COL_ENTRY_BATCH_ROOT: u32 = 3; pub const COL_TX_COMPLETED: u32 = 4; pub const COL_MISC: u32 = 5; pub const COL_SEAL_CONTEXT: u32 = 6; -pub const COL_NUM: u32 = 7; +pub const COL_FLOW_MPT_NODES: u32 = 7; +pub const COL_NUM: u32 = 8; pub struct LogManager { pub(crate) db: Arc, @@ -119,11 +120,12 @@ impl LogStoreChunkWrite for LogManager { self.append_entries(flow_entry_array)?; if let Some(file_proof) = maybe_file_proof { - self.pora_chunks_merkle.fill_with_file_proof( + let updated_node_list = self.pora_chunks_merkle.fill_with_file_proof( file_proof, tx.merkle_nodes, tx.start_entry_index, )?; + self.flow_store.put_mpt_node_list(updated_node_list)?; } Ok(true) } @@ -273,8 +275,10 @@ impl LogStoreWrite for LogManager { ) -> Result { let valid = self.validate_range_proof(tx_seq, data)?; if valid { - self.pora_chunks_merkle + let updated_nodes = self + .pora_chunks_merkle .fill_with_range_proof(data.proof.clone())?; + self.flow_store.put_mpt_node_list(updated_nodes)?; } Ok(valid) } diff --git a/tests/sync_test.py b/tests/sync_test.py index 157e9e2..96e9017 100755 --- a/tests/sync_test.py +++ b/tests/sync_test.py @@ -89,6 +89,11 @@ class SyncTest(TestFramework): assert_equal(client2.zgs_get_file_info(data_root)["finalized"], False) assert(client2.zgs_download_segment_decoded(data_root, 1024, 2048) is None) + # Restart node 1 to check if the proof nodes are persisted. + self.stop_storage_node(0) + self.start_storage_node(0) + self.nodes[0].wait_for_rpc_connection() + # Trigger chunks sync by rpc assert(client2.admin_start_sync_chunks(1, 1024, 2048) is None) wait_until(lambda: client2.sycn_status_is_completed_or_unknown(1))