smeb_y/0g-storage-node
1
0
Fork 0
mirror of https://github.com/0glabs/0g-storage-node.git synced 2024-11-20 15:05:19 +00:00
Code Issues Actions 4 Packages Projects Releases Wiki Activity

Merge 55a4270623 into baf0521c99

Browse Source
This commit is contained in:
0g-peterzhb 2024-11-06 14:02:06 +08:00 committed by GitHub
commit 70ba30599e
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
15 changed files with 190 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
Cargo.lock generated
View File

@ -226,6 +226,7 @@ dependencies = [
"itertools 0.13.0",
"lazy_static",
"lru 0.12.5",
"metrics",
"once_cell",
"serde",
"tiny-keccak",
@ -7300,8 +7301,10 @@ dependencies = [
"kvdb",
"kvdb-memorydb",
"kvdb-rocksdb",
"lazy_static",
"merkle_light",
"merkle_tree",
"metrics",
"once_cell",
"parking_lot 0.12.3",
"rand 0.8.5",
@ -7324,6 +7327,7 @@ name = "storage-async"
version = "0.1.0"
dependencies = [
"anyhow",
"backtrace",
"eth2_ssz",
"shared_types",
"storage",

3
common/append_merkle/Cargo.toml
View File

@ -13,5 +13,8 @@ serde = { version = "1.0.137", features = ["derive"] }
lazy_static = "1.4.0"
tracing = "0.1.36"
once_cell = "1.19.0"
metrics = { workspace = true }
itertools = "0.13.0"
lru = "0.12.5"

15
common/append_merkle/src/lib.rs
View File

@ -1,4 +1,5 @@
mod merkle_tree;
mod metrics;
mod node_manager;
mod proof;
mod sha3;
@ -10,6 +11,7 @@ use std::collections::{BTreeMap, HashMap};
use std::fmt::Debug;
use std::marker::PhantomData;
use std::sync::Arc;
use std::time::Instant;
use tracing::{trace, warn};
use crate::merkle_tree::MerkleTreeWrite;
@ -145,6 +147,7 @@ impl<E: HashElement, A: Algorithm<E>> AppendMerkleTree<E, A> {
}
pub fn append(&mut self, new_leaf: E) {
let start_time = Instant::now();
if new_leaf == E::null() {
// appending null is not allowed.
return;
@ -152,10 +155,13 @@ impl<E: HashElement, A: Algorithm<E>> AppendMerkleTree<E, A> {
self.node_manager.start_transaction();
self.node_manager.push_node(0, new_leaf);
self.recompute_after_append_leaves(self.leaves() - 1);
self.node_manager.commit();
metrics::APPEND.update_since(start_time);
}
pub fn append_list(&mut self, leaf_list: Vec<E>) {
let start_time = Instant::now();
if leaf_list.contains(&E::null()) {
// appending null is not allowed.
return;
@ -165,6 +171,7 @@ impl<E: HashElement, A: Algorithm<E>> AppendMerkleTree<E, A> {
self.node_manager.append_nodes(0, &leaf_list);
self.recompute_after_append_leaves(start_index);
self.node_manager.commit();
metrics::APPEND_LIST.update_since(start_time);
}
/// Append a leaf list by providing their intermediate node hash.
@ -173,6 +180,7 @@ impl<E: HashElement, A: Algorithm<E>> AppendMerkleTree<E, A> {
/// Other nodes in the subtree will be set to `null` nodes.
/// TODO: Optimize to avoid storing the `null` nodes?
pub fn append_subtree(&mut self, subtree_depth: usize, subtree_root: E) -> Result<()> {
let start_time = Instant::now();
if subtree_root == E::null() {
// appending null is not allowed.
bail!("subtree_root is null");
@ -182,10 +190,13 @@ impl<E: HashElement, A: Algorithm<E>> AppendMerkleTree<E, A> {
self.append_subtree_inner(subtree_depth, subtree_root)?;
self.recompute_after_append_subtree(start_index, subtree_depth - 1);
self.node_manager.commit();
metrics::APPEND_SUBTREE.update_since(start_time);
Ok(())
}
pub fn append_subtree_list(&mut self, subtree_list: Vec<(usize, E)>) -> Result<()> {
let start_time = Instant::now();
if subtree_list.iter().any(|(_, root)| root == &E::null()) {
// appending null is not allowed.
bail!("subtree_list contains null");
@ -197,12 +208,15 @@ impl<E: HashElement, A: Algorithm<E>> AppendMerkleTree<E, A> {
self.recompute_after_append_subtree(start_index, subtree_depth - 1);
}
self.node_manager.commit();
metrics::APPEND_SUBTREE_LIST.update_since(start_time);
Ok(())
}
/// Change the value of the last leaf and return the new merkle root.
/// This is needed if our merkle-tree in memory only keeps intermediate nodes instead of real leaves.
pub fn update_last(&mut self, updated_leaf: E) {
let start_time = Instant::now();
if updated_leaf == E::null() {
// updating to null is not allowed.
return;
@ -216,6 +230,7 @@ impl<E: HashElement, A: Algorithm<E>> AppendMerkleTree<E, A> {
}
self.recompute_after_append_leaves(self.leaves() - 1);
self.node_manager.commit();
metrics::UPDATE_LAST.update_since(start_time);
}
/// Fill an unknown `null` leaf with its real value.

11
common/append_merkle/src/metrics.rs Normal file
View File

@ -0,0 +1,11 @@
use std::sync::Arc;
use metrics::{register_timer, Timer};
lazy_static::lazy_static! {
pub static ref APPEND: Arc<dyn Timer> = register_timer("append_merkle_append");
pub static ref APPEND_LIST: Arc<dyn Timer> = register_timer("append_merkle_append_list");
pub static ref APPEND_SUBTREE: Arc<dyn Timer> = register_timer("append_merkle_append_subtree");
pub static ref APPEND_SUBTREE_LIST: Arc<dyn Timer> = register_timer("append_merkle_append_subtree_list");
pub static ref UPDATE_LAST: Arc<dyn Timer> = register_timer("append_merkle_update_last");
}

15
node/log_entry_sync/src/sync_manager/log_entry_fetcher.rs
View File

@ -1,5 +1,5 @@
use crate::sync_manager::log_query::LogQuery;
use crate::sync_manager::RETRY_WAIT_MS;
use crate::sync_manager::{metrics, RETRY_WAIT_MS};
use crate::ContractAddress;
use anyhow::{anyhow, bail, Result};
use append_merkle::{Algorithm, Sha3Algorithm};
@ -14,15 +14,12 @@ use shared_types::{DataRoot, Transaction};
use std::collections::{BTreeMap, HashMap};
use std::str::FromStr;
use std::sync::Arc;
use std::time::Duration;
use std::time::{Duration, Instant};
use storage::log_store::{tx_store::BlockHashAndSubmissionIndex, Store};
use task_executor::TaskExecutor;
use tokio::{
sync::{
mpsc::{UnboundedReceiver, UnboundedSender},
RwLock,
},
time::Instant,
use tokio::sync::{
mpsc::{UnboundedReceiver, UnboundedSender},
RwLock,
};
pub struct LogEntryFetcher {
@ -242,6 +239,7 @@ impl LogEntryFetcher {
);
let (mut block_hash_sent, mut block_number_sent) = (None, None);
while let Some(maybe_log) = stream.next().await {
let start_time = Instant::now();
match maybe_log {
Ok(log) => {
let sync_progress =
@ -301,6 +299,7 @@ impl LogEntryFetcher {
tokio::time::sleep(Duration::from_millis(RETRY_WAIT_MS)).await;
}
}
metrics::RECOVER_LOG.update_since(start_time);
}
info!("log recover end");

10
node/log_entry_sync/src/sync_manager/metrics.rs
View File

@ -1,7 +1,13 @@
use std::sync::Arc;
use metrics::{register_timer, Timer};
use metrics::{register_timer, Gauge, GaugeUsize, Timer};
lazy_static::lazy_static! {
pub static ref STORE_PUT_TX: Arc<dyn Timer> = register_timer("log_entry_sync_store_put_tx");
pub static ref LOG_MANAGER_HANDLE_DATA_TRANSACTION: Arc<dyn Timer> = register_timer("log_manager_handle_data_transaction");
pub static ref STORE_PUT_TX: Arc<dyn Timer> = register_timer("log_entry_sync_manager_put_tx_inner");
pub static ref STORE_PUT_TX_SPEED_IN_BYTES: Arc<dyn Gauge<usize>> = GaugeUsize::register("log_entry_sync_manager_put_tx_speed_in_bytes");
pub static ref RECOVER_LOG: Arc<dyn Timer> = register_timer("log_entry_sync_manager_recover_log");
}

62
node/log_entry_sync/src/sync_manager/mod.rs
View File

@ -26,6 +26,7 @@ const RETRY_WAIT_MS: u64 = 500;
// Each tx has less than 10KB, so the cache size should be acceptable.
const BROADCAST_CHANNEL_CAPACITY: usize = 25000;
const CATCH_UP_END_GAP: u64 = 10;
const CHECK_ROOT_INTERVAL: u64 = 500;
/// Errors while handle data
#[derive(Error, Debug)]
@ -408,6 +409,7 @@ impl LogSyncManager {
}
LogFetchProgress::Transaction((tx, block_number)) => {
let mut stop = false;
let start_time = Instant::now();
match self.put_tx(tx.clone()).await {
Some(false) => stop = true,
Some(true) => {
@ -441,6 +443,8 @@ impl LogSyncManager {
// no receivers will be created.
warn!("log sync broadcast error, error={:?}", e);
}
metrics::LOG_MANAGER_HANDLE_DATA_TRANSACTION.update_since(start_time);
}
LogFetchProgress::Reverted(reverted) => {
self.process_reverted(reverted).await;
@ -453,7 +457,6 @@ impl LogSyncManager {
async fn put_tx_inner(&mut self, tx: Transaction) -> bool {
let start_time = Instant::now();
let result = self.store.put_tx(tx.clone());
metrics::STORE_PUT_TX.update_since(start_time);
if let Err(e) = result {
error!("put_tx error: e={:?}", e);
@ -513,38 +516,45 @@ impl LogSyncManager {
// Check if the computed data root matches on-chain state.
// If the call fails, we won't check the root here and return `true` directly.
let flow_contract = self.log_fetcher.flow_contract();
match flow_contract
.get_flow_root_by_tx_seq(tx.seq.into())
.call()
.await
{
Ok(contract_root_bytes) => {
let contract_root = H256::from_slice(&contract_root_bytes);
// contract_root is zero for tx submitted before upgrading.
if !contract_root.is_zero() {
match self.store.get_context() {
Ok((local_root, _)) => {
if contract_root != local_root {
error!(
?contract_root,
?local_root,
"local flow root and on-chain flow root mismatch"
);
return false;
if self.next_tx_seq % CHECK_ROOT_INTERVAL == 0 {
let flow_contract = self.log_fetcher.flow_contract();
match flow_contract
.get_flow_root_by_tx_seq(tx.seq.into())
.call()
.await
{
Ok(contract_root_bytes) => {
let contract_root = H256::from_slice(&contract_root_bytes);
// contract_root is zero for tx submitted before upgrading.
if !contract_root.is_zero() {
match self.store.get_context() {
Ok((local_root, _)) => {
if contract_root != local_root {
error!(
?contract_root,
?local_root,
"local flow root and on-chain flow root mismatch"
);
return false;
}
}
Err(e) => {
warn!(?e, "fail to read the local flow root");
}
}
Err(e) => {
warn!(?e, "fail to read the local flow root");
}
}
}
}
Err(e) => {
warn!(?e, "fail to read the on-chain flow root");
Err(e) => {
warn!(?e, "fail to read the on-chain flow root");
}
}
}
metrics::STORE_PUT_TX_SPEED_IN_BYTES
.update((tx.size * 1000 / start_time.elapsed().as_micros() as u64) as usize);
metrics::STORE_PUT_TX.update_since(start_time);
true
}
}

3
node/storage-async/Cargo.toml
View File

@ -10,4 +10,5 @@ storage = { path = "../storage" }
task_executor = { path = "../../common/task_executor" }
tokio = { version = "1.19.2", features = ["sync"] }
tracing = "0.1.35"
eth2_ssz = "0.4.0"
eth2_ssz = "0.4.0"
backtrace = "0.3"

13
node/storage-async/src/lib.rs
View File

@ -2,6 +2,7 @@
extern crate tracing;
use anyhow::bail;
use backtrace::Backtrace;
use shared_types::{
Chunk, ChunkArray, ChunkArrayWithProof, DataRoot, FlowProof, FlowRangeProof, Transaction,
};
@ -148,8 +149,16 @@ impl Store {
WORKER_TASK_NAME,
);
rx.await
.unwrap_or_else(|_| bail!(error::Error::Custom("Receiver error".to_string())))
rx.await.unwrap_or_else(|_| {
let mut backtrace = Backtrace::new();
let mut frames = backtrace.frames().to_vec();
if frames.len() > 3 {
frames.drain(0..); //Remove the first 3 unnecessary frames to simplify// backtrace
backtrace = frames.into();
Some(format!("{:?}", backtrace));
}
bail!(error::Error::Custom("Receiver error".to_string()))
})
}
// FIXME(zz): Refactor the lock and async call here.

2
node/storage/Cargo.toml
View File

@ -31,6 +31,8 @@ parking_lot = "0.12.3"
serde_json = "1.0.127"
tokio = { version = "1.38.0", features = ["full"] }
task_executor = { path = "../../common/task_executor" }
lazy_static = "1.4.0"
metrics = { workspace = true }
once_cell = { version = "1.19.0", features = [] }
[dev-dependencies]

18
node/storage/src/log_store/flow_store.rs
View File

@ -1,12 +1,13 @@
use super::load_chunk::EntryBatch;
use super::seal_task_manager::SealTaskManager;
use super::{MineLoadChunk, SealAnswer, SealTask};
use crate::config::ShardConfig;
use crate::error::Error;
use crate::log_store::load_chunk::EntryBatch;
use crate::log_store::log_manager::{
bytes_to_entries, COL_ENTRY_BATCH, COL_FLOW_MPT_NODES, PORA_CHUNK_SIZE,
};
use crate::log_store::{FlowRead, FlowSeal, FlowWrite};
use crate::log_store::seal_task_manager::SealTaskManager;
use crate::log_store::{
metrics, FlowRead, FlowSeal, FlowWrite, MineLoadChunk, SealAnswer, SealTask,
};
use crate::{try_option, ZgsKeyValueDB};
use any::Any;
use anyhow::{anyhow, bail, Result};
@ -20,6 +21,7 @@ use ssz_derive::{Decode as DeriveDecode, Encode as DeriveEncode};
use std::fmt::Debug;
use std::sync::Arc;
use std::time::Instant;
use std::{any, cmp};
use tracing::{debug, error, trace};
use zgs_spec::{BYTES_PER_SECTOR, SEALS_PER_LOAD, SECTORS_PER_LOAD, SECTORS_PER_SEAL};
@ -44,6 +46,7 @@ impl FlowStore {
batch_index: usize,
subtree_list: Vec<(usize, usize, DataRoot)>,
) -> Result<()> {
let start_time = Instant::now();
let mut batch = self
.data_db
.get_entry_batch(batch_index as u64)?
@ -51,7 +54,7 @@ impl FlowStore {
batch.set_subtree_list(subtree_list);
self.data_db
.put_entry_raw(vec![(batch_index as u64, batch)])?;
metrics::INSERT_SUBTREE_LIST.update_since(start_time);
Ok(())
}
@ -205,6 +208,7 @@ impl FlowWrite for FlowStore {
/// Return the roots of completed chunks. The order is guaranteed to be increasing
/// by chunk index.
fn append_entries(&self, data: ChunkArray) -> Result<Vec<(u64, DataRoot)>> {
let start_time = Instant::now();
let mut to_seal_set = self.seal_manager.to_seal_set.write();
trace!("append_entries: {} {}", data.start_index, data.data.len());
if data.data.len() % BYTES_PER_SECTOR != 0 {
@ -247,6 +251,8 @@ impl FlowWrite for FlowStore {
batch_list.push((chunk_index, batch));
}
metrics::APPEND_ENTRIES.update_since(start_time);
self.data_db.put_entry_batch_list(batch_list)
}
@ -356,6 +362,7 @@ impl FlowDBStore {
&self,
batch_list: Vec<(u64, EntryBatch)>,
) -> Result<Vec<(u64, DataRoot)>> {
let start_time = Instant::now();
let mut completed_batches = Vec::new();
let mut tx = self.kvdb.transaction();
for (batch_index, batch) in batch_list {
@ -370,6 +377,7 @@ impl FlowDBStore {
}
}
self.kvdb.write(tx)?;
metrics::PUT_ENTRY_BATCH_LIST.update_since(start_time);
Ok(completed_batches)
}

27
node/storage/src/log_store/log_manager.rs
View File

@ -1,5 +1,3 @@
use super::tx_store::BlockHashAndSubmissionIndex;
use super::{FlowSeal, MineLoadChunk, SealAnswer, SealTask};
use crate::config::ShardConfig;
use crate::log_store::flow_store::{batch_iter_sharded, FlowConfig, FlowDBStore, FlowStore};
use crate::log_store::tx_store::TransactionStore;
@ -26,8 +24,13 @@ use std::cmp::Ordering;
use std::path::Path;
use std::sync::mpsc;
use std::sync::Arc;
use std::time::Instant;
use tracing::{debug, error, info, instrument, trace, warn};
use crate::log_store::metrics;
use crate::log_store::tx_store::BlockHashAndSubmissionIndex;
use crate::log_store::{FlowSeal, MineLoadChunk, SealAnswer, SealTask};
/// 256 Bytes
pub const ENTRY_SIZE: usize = 256;
/// 1024 Entries.
@ -188,6 +191,7 @@ impl LogStoreChunkWrite for LogManager {
chunks: ChunkArray,
maybe_file_proof: Option<FlowProof>,
) -> Result<bool> {
let start_time = Instant::now();
let mut merkle = self.merkle.write();
let tx = self
.tx_store
@ -219,6 +223,7 @@ impl LogStoreChunkWrite for LogManager {
tx.start_entry_index,
)?;
}
metrics::PUT_CHUNKS.update_since(start_time);
Ok(true)
}
@ -249,6 +254,7 @@ impl LogStoreWrite for LogManager {
/// `put_tx` for the last tx when we restart the node to ensure that it succeeds.
///
fn put_tx(&self, tx: Transaction) -> Result<()> {
let start_time = Instant::now();
let mut merkle = self.merkle.write();
debug!("put_tx: tx={:?}", tx);
let expected_seq = self.tx_store.next_tx_seq();
@ -278,6 +284,7 @@ impl LogStoreWrite for LogManager {
self.copy_tx_and_finalize(old_tx_seq, vec![tx.seq])?;
}
}
metrics::PUT_TX.update_since(start_time);
Ok(())
}
@ -308,6 +315,7 @@ impl LogStoreWrite for LogManager {
}
fn finalize_tx_with_hash(&self, tx_seq: u64, tx_hash: H256) -> crate::error::Result<bool> {
let start_time = Instant::now();
trace!(
"finalize_tx_with_hash: tx_seq={} tx_hash={:?}",
tx_seq,
@ -336,6 +344,7 @@ impl LogStoreWrite for LogManager {
if same_root_seq_list.first() == Some(&tx_seq) {
self.copy_tx_and_finalize(tx_seq, same_root_seq_list[1..].to_vec())?;
}
metrics::FINALIZE_TX_WITH_HASH.update_since(start_time);
Ok(true)
} else {
bail!("finalize tx hash with data missing: tx_seq={}", tx_seq)
@ -861,6 +870,7 @@ impl LogManager {
if merkle_list.is_empty() {
return Ok(());
}
let start_time = Instant::now();
self.pad_tx(tx_start_index, &mut *merkle)?;
@ -896,12 +906,15 @@ impl LogManager {
.append_subtree(subtree_depth - log2_pow2(PORA_CHUNK_SIZE), subtree_root)?;
}
}
metrics::APPEND_SUBTREE_LIST.update_since(start_time);
Ok(())
}
#[instrument(skip(self, merkle))]
fn pad_tx(&self, tx_start_index: u64, merkle: &mut MerkleManager) -> Result<()> {
// Check if we need to pad the flow.
let start_time = Instant::now();
let mut tx_start_flow_index =
merkle.last_chunk_start_index() + merkle.last_chunk_merkle.leaves() as u64;
let pad_size = tx_start_index - tx_start_flow_index;
@ -979,6 +992,8 @@ impl LogManager {
merkle.pora_chunks_merkle.leaves(),
merkle.last_chunk_merkle.leaves()
);
metrics::PAD_TX.update_since(start_time);
Ok(())
}
@ -1107,6 +1122,8 @@ impl LogManager {
}
fn copy_tx_and_finalize(&self, from_tx_seq: u64, to_tx_seq_list: Vec<u64>) -> Result<()> {
let start_time = Instant::now();
let mut merkle = self.merkle.write();
let shard_config = self.flow_store.get_shard_config();
// We have all the data need for this tx, so just copy them.
@ -1155,6 +1172,8 @@ impl LogManager {
for (seq, _) in to_tx_offset_list {
self.tx_store.finalize_tx(seq)?;
}
metrics::COPY_TX_AND_FINALIZE.update_since(start_time);
Ok(())
}
@ -1227,6 +1246,7 @@ pub fn sub_merkle_tree(leaf_data: &[u8]) -> Result<FileMerkleTree> {
}
pub fn data_to_merkle_leaves(leaf_data: &[u8]) -> Result<Vec<H256>> {
let start_time = Instant::now();
if leaf_data.len() % ENTRY_SIZE != 0 {
bail!("merkle_tree: mismatched data size");
}
@ -1242,6 +1262,9 @@ pub fn data_to_merkle_leaves(leaf_data: &[u8]) -> Result<Vec<H256>> {
.map(Sha3Algorithm::leaf)
.collect()
};
metrics::DATA_TO_MERKLE_LEAVES_SIZE.update(leaf_data.len() as usize);
metrics::DATA_TO_MERKLE_LEAVES.update_since(start_time);
Ok(r)
}

41
node/storage/src/log_store/metrics.rs Normal file
View File

@ -0,0 +1,41 @@
use std::sync::Arc;
use metrics::{register_timer, Gauge, GaugeUsize, Timer};
lazy_static::lazy_static! {
pub static ref PUT_TX: Arc<dyn Timer> = register_timer("log_store_put_tx");
pub static ref PUT_CHUNKS: Arc<dyn Timer> = register_timer("log_store_put_chunks");
pub static ref TX_STORE_PUT: Arc<dyn Timer> = register_timer("log_store_tx_store_put_tx");
pub static ref CHECK_TX_COMPLETED: Arc<dyn Timer> =
register_timer("log_store_log_manager_check_tx_completed");
pub static ref APPEND_SUBTREE_LIST: Arc<dyn Timer> =
register_timer("log_store_log_manager_append_subtree_list");
pub static ref DATA_TO_MERKLE_LEAVES: Arc<dyn Timer> =
register_timer("log_store_log_manager_data_to_merkle_leaves");
pub static ref COPY_TX_AND_FINALIZE: Arc<dyn Timer> =
register_timer("log_store_log_manager_copy_tx_and_finalize");
pub static ref PAD_TX: Arc<dyn Timer> = register_timer("log_store_log_manager_pad_tx");
pub static ref PUT_BATCH_ROOT_LIST: Arc<dyn Timer> = register_timer("log_store_flow_store_put_batch_root_list");
pub static ref INSERT_SUBTREE_LIST: Arc<dyn Timer> =
register_timer("log_store_flow_store_insert_subtree_list");
pub static ref PUT_MPT_NODE: Arc<dyn Timer> = register_timer("log_store_flow_store_put_mpt_node");
pub static ref PUT_ENTRY_BATCH_LIST: Arc<dyn Timer> =
register_timer("log_store_flow_store_put_entry_batch_list");
pub static ref APPEND_ENTRIES: Arc<dyn Timer> = register_timer("log_store_flow_store_append_entries");
pub static ref FINALIZE_TX_WITH_HASH: Arc<dyn Timer> = register_timer("log_store_log_manager_finalize_tx_with_hash");
pub static ref DATA_TO_MERKLE_LEAVES_SIZE: Arc<dyn Gauge<usize>> = GaugeUsize::register("log_store_data_to_merkle_leaves_size");
}

1
node/storage/src/log_store/mod.rs
View File

@ -15,6 +15,7 @@ pub mod config;
mod flow_store;
mod load_chunk;
pub mod log_manager;
mod metrics;
mod seal_task_manager;
#[cfg(test)]
mod tests;

13
node/storage/src/log_store/tx_store.rs
View File

@ -3,6 +3,7 @@ use crate::log_store::log_manager::{
data_to_merkle_leaves, sub_merkle_tree, COL_BLOCK_PROGRESS, COL_MISC, COL_TX, COL_TX_COMPLETED,
COL_TX_DATA_ROOT_INDEX, ENTRY_SIZE, PORA_CHUNK_SIZE,
};
use crate::log_store::metrics;
use crate::{try_option, LogManager, ZgsKeyValueDB};
use anyhow::{anyhow, Result};
use append_merkle::{AppendMerkleTree, MerkleTreeRead, Sha3Algorithm};
@ -15,6 +16,7 @@ use std::collections::hash_map::Entry;
use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Instant;
use tracing::{error, instrument};
const LOG_SYNC_PROGRESS_KEY: &str = "log_sync_progress";
@ -51,6 +53,8 @@ impl TransactionStore {
#[instrument(skip(self))]
/// Return `Ok(Some(tx_seq))` if a previous transaction has the same tx root.
pub fn put_tx(&self, mut tx: Transaction) -> Result<Vec<u64>> {
let start_time = Instant::now();
let old_tx_seq_list = self.get_tx_seq_list_by_data_root(&tx.data_merkle_root)?;
if old_tx_seq_list.last().is_some_and(|seq| *seq == tx.seq) {
// The last tx is inserted again, so no need to process it.
@ -86,6 +90,7 @@ impl TransactionStore {
);
self.next_tx_seq.store(tx.seq + 1, Ordering::SeqCst);
self.kvdb.write(db_tx)?;
metrics::TX_STORE_PUT.update_since(start_time);
Ok(old_tx_seq_list)
}
@ -175,8 +180,12 @@ impl TransactionStore {
}
pub fn check_tx_completed(&self, tx_seq: u64) -> Result<bool> {
Ok(self.kvdb.get(COL_TX_COMPLETED, &tx_seq.to_be_bytes())?
== Some(vec![TX_STATUS_FINALIZED]))
let start_time = Instant::now();
let res = self.kvdb.get(COL_TX_COMPLETED, &tx_seq.to_be_bytes())?
== Some(vec![TX_STATUS_FINALIZED]);
metrics::CHECK_TX_COMPLETED.update_since(start_time);
Ok(res)
}
pub fn check_tx_pruned(&self, tx_seq: u64) -> Result<bool> {