diff --git a/common/append_merkle/src/lib.rs b/common/append_merkle/src/lib.rs index ddf4c25..cddafaa 100644 --- a/common/append_merkle/src/lib.rs +++ b/common/append_merkle/src/lib.rs @@ -310,7 +310,7 @@ impl> AppendMerkleTree { /// TODO: Batch computing intermediate nodes. pub fn fill_leaf(&mut self, index: usize, leaf: E) { // print node leaf at 12288 index - + if leaf.is_null() { // fill leaf with null is not allowed. } else if self.node(0, index).is_null() { @@ -427,10 +427,6 @@ impl> AppendMerkleTree { self.root_to_tx_seq_map.contains_key(root) } - pub fn get_node_hash_by_index(&self, index: usize) -> Result> { - Ok(Some(self.node(0, index))) - } - pub fn leaf_at(&self, position: usize) -> Result> { if position >= self.leaves() { bail!("Out of bound: position={} end={}", position, self.leaves()); diff --git a/node/rpc/src/zgs/api.rs b/node/rpc/src/zgs/api.rs index 8862777..abc33ea 100644 --- a/node/rpc/src/zgs/api.rs +++ b/node/rpc/src/zgs/api.rs @@ -83,10 +83,7 @@ pub trait Rpc { ) -> RpcResult; #[method(name = "getHashAtNodeIndex")] - async fn get_hash_at_node_index( - &self, - node_index: u64, - ) -> RpcResult>; + async fn get_hash_at_node_index(&self, node_index: u64) -> RpcResult>; #[method(name = "getFlowContext")] async fn get_flow_context(&self) -> RpcResult<(H256, u64)>; diff --git a/node/rpc/src/zgs/impl.rs b/node/rpc/src/zgs/impl.rs index bc44b9a..cffc7fa 100644 --- a/node/rpc/src/zgs/impl.rs +++ b/node/rpc/src/zgs/impl.rs @@ -225,10 +225,7 @@ impl RpcServer for RpcServerImpl { Ok(proof.right_proof) } - async fn get_hash_at_node_index( - &self, - node_index: u64, - ) -> RpcResult> { + async fn get_hash_at_node_index(&self, node_index: u64) -> RpcResult> { debug!(%node_index, "zgs_getHashAtNodeIndex"); let hash = self .ctx diff --git a/node/storage-async/src/lib.rs b/node/storage-async/src/lib.rs index ca9b528..989b748 100644 --- a/node/storage-async/src/lib.rs +++ b/node/storage-async/src/lib.rs @@ -2,6 +2,7 @@ extern crate tracing; use anyhow::bail; +use append_merkle::OptionalHash; use shared_types::{ Chunk, ChunkArray, ChunkArrayWithProof, DataRoot, FlowProof, FlowRangeProof, Transaction, }; @@ -10,7 +11,6 @@ use std::sync::Arc; use storage::{error, error::Result, log_store::Store as LogStore, H256}; use task_executor::TaskExecutor; use tokio::sync::oneshot; -use append_merkle::OptionalHash; pub use storage::config::ShardConfig; use storage::log_store::config::ConfigurableExt; diff --git a/node/storage/src/log_store/flow_store.rs b/node/storage/src/log_store/flow_store.rs index 3d862fa..764a172 100644 --- a/node/storage/src/log_store/flow_store.rs +++ b/node/storage/src/log_store/flow_store.rs @@ -228,6 +228,17 @@ impl FlowWrite for FlowStore { if data.data.len() % BYTES_PER_SECTOR != 0 { bail!("append_entries: invalid data size, len={}", data.data.len()); } + + let batch_data_12288 = self + .data_db + .get_entry_batch(12288)? + .unwrap_or_else(|| EntryBatch::new(12288)); + + debug!( + "Entry batch data at 12288 before insert: {:?}", + batch_data_12288.as_ssz_bytes() + ); + let mut batch_list = Vec::new(); for (start_entry_index, end_entry_index) in batch_iter( data.start_index, @@ -250,6 +261,7 @@ impl FlowWrite for FlowStore { .data_db .get_entry_batch(chunk_index)? .unwrap_or_else(|| EntryBatch::new(chunk_index)); + let completed_seals = batch.insert_data( (chunk.start_index % self.config.batch_size as u64) as usize, chunk.data, @@ -267,7 +279,19 @@ impl FlowWrite for FlowStore { } metrics::APPEND_ENTRIES.update_since(start_time); - self.data_db.put_entry_batch_list(batch_list) + let res = self.data_db.put_entry_batch_list(batch_list); + + let batch_data_12288 = self + .data_db + .get_entry_batch(12288)? + .unwrap_or_else(|| EntryBatch::new(12288)); + + debug!( + "Entry batch data at 12288 after insert: {:?}", + batch_data_12288.as_ssz_bytes() + ); + + res } fn truncate(&self, start_index: u64) -> crate::error::Result<()> { @@ -393,6 +417,7 @@ impl FlowDBStore { let start_time = Instant::now(); let mut completed_batches = Vec::new(); let mut tx = self.kvdb.transaction(); + for (batch_index, batch) in batch_list { tx.put( COL_ENTRY_BATCH, diff --git a/node/storage/src/log_store/log_manager.rs b/node/storage/src/log_store/log_manager.rs index 767af16..c3bdc0f 100644 --- a/node/storage/src/log_store/log_manager.rs +++ b/node/storage/src/log_store/log_manager.rs @@ -564,11 +564,9 @@ impl LogStoreRead for LogManager { self.tx_store.get_tx_by_seq_number(seq) } - fn get_node_hash_by_index(&self, index:u64) -> crate::error::Result { + fn get_node_hash_by_index(&self, index: u64) -> crate::error::Result { let merkle = self.merkle.read(); - let opt = merkle - .pora_chunks_merkle - .get_node_hash_by_index(index as usize)?; + let opt = merkle.pora_chunks_merkle.leaf_at(index as usize)?; opt.ok_or_else(|| anyhow!("node hash not found at index {}", index)) } @@ -1123,7 +1121,7 @@ impl LogManager { .pora_chunks_merkle .update_last(merkle.last_chunk_merkle.root()); } - + let chunk_roots = self.flow_store.append_entries(flow_entry_array)?; debug!("fill leaf for pora_chunks_merkle"); for (chunk_index, chunk_root) in chunk_roots { @@ -1180,10 +1178,13 @@ impl LogManager { ); // Padding should only start after real data ends - let real_data_end_index = (segments_for_file - 1) * PORA_CHUNK_SIZE + last_segment_size_for_file; + let real_data_end_index = + (segments_for_file - 1) * PORA_CHUNK_SIZE + last_segment_size_for_file; debug!( "Padding: real_data_end_index={}, padding_start_segment={}, padding_start_offset={}", - real_data_end_index, segments_for_file - 1, last_segment_size_for_file + real_data_end_index, + segments_for_file - 1, + last_segment_size_for_file ); while segments_for_file <= segments_for_proof { @@ -1193,12 +1194,15 @@ impl LogManager { (PORA_CHUNK_SIZE - last_segment_size_for_file) * ENTRY_SIZE }; - let padding_start_index = ((segments_for_file - 1) * PORA_CHUNK_SIZE - + last_segment_size_for_file) as u64; + let padding_start_index = + ((segments_for_file - 1) * PORA_CHUNK_SIZE + last_segment_size_for_file) as u64; debug!( "Padding iteration: segment={}, offset={}, padding_size={}, start_index={}", - segments_for_file - 1, last_segment_size_for_file, padding_size, padding_start_index + segments_for_file - 1, + last_segment_size_for_file, + padding_size, + padding_start_index ); if padding_size > 0 { diff --git a/node/storage/src/log_store/mod.rs b/node/storage/src/log_store/mod.rs index 45b7622..4ac1b77 100644 --- a/node/storage/src/log_store/mod.rs +++ b/node/storage/src/log_store/mod.rs @@ -31,7 +31,7 @@ pub trait LogStoreRead: LogStoreChunkRead { /// Get a transaction by its global log sequence number. fn get_tx_by_seq_number(&self, seq: u64) -> Result>; - fn get_node_hash_by_index(&self, index:u64) -> Result; + fn get_node_hash_by_index(&self, index: u64) -> Result; /// Get a transaction by the data root of its data. /// If all txs are not finalized, return the first one if need available is false.