add detailed metrics in slow operations

* Add new P2P protocol NewFile

* Publish NewFile message when any file finalized

* handle NewFile message in router

* handle NewFile in sync servic to write in db

* use propagation source to handle NewFile message

* Disable sequential sync and store new file in v2 sync store

* Add shard config in FindFile

* Add AnnounceFile RPC message in network layer

* do not propagate FindFile to whole network

* Mark peer connected if FileAnnouncement RPC message received

* fix unit test failures

* Change P2P protocol version

* Ignore py tests of sequential auto sync

* Add py test for auto sync v2

* fmt code

* remove dummy code in py test

* fix random test failure

* Add comments

* Enable file sync protocol v2 in config file by default

Cargo.lock

@@ -223,7 +223,10 @@ dependencies = [
  "eth2_ssz",
  "eth2_ssz_derive",
  "ethereum-types 0.14.1",
+ "itertools 0.13.0",
  "lazy_static",
+ "lru 0.12.5",
+ "metrics",
  "once_cell",
  "serde",
  "tiny-keccak",
@@ -1673,7 +1676,7 @@ dependencies = [
  "hkdf",
  "lazy_static",
  "libp2p-core 0.30.2",
- "lru",
+ "lru 0.7.8",
  "parking_lot 0.11.2",
  "rand 0.8.5",
  "rlp",
@@ -2514,6 +2517,12 @@ version = "1.0.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1"
 
+[[package]]
+name = "foldhash"
+version = "0.1.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f81ec6369c545a7d40e4589b5597581fa1c441fe1cce96dd1de43159910a36a2"
+
 [[package]]
 name = "foreign-types"
 version = "0.3.2"
@@ -2946,6 +2955,17 @@ dependencies = [
  "allocator-api2",
 ]
 
+[[package]]
+name = "hashbrown"
+version = "0.15.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1e087f84d4f86bf4b218b927129862374b72199ae7d8657835f1e89000eea4fb"
+dependencies = [
+ "allocator-api2",
+ "equivalent",
+ "foldhash",
+]
+
 [[package]]
 name = "hashers"
 version = "1.0.1"
@@ -4117,7 +4137,7 @@ dependencies = [
  "libp2p-core 0.33.0",
  "libp2p-swarm",
  "log",
- "lru",
+ "lru 0.7.8",
  "prost 0.10.4",
  "prost-build 0.10.4",
  "prost-codec",
@@ -4650,6 +4670,15 @@ dependencies = [
  "hashbrown 0.12.3",
 ]
 
+[[package]]
+name = "lru"
+version = "0.12.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "234cf4f4a04dc1f57e24b96cc0cd600cf2af460d4161ac5ecdd0af8e1f3b2a38"
+dependencies = [
+ "hashbrown 0.15.0",
+]
+
 [[package]]
 name = "lru-cache"
 version = "0.1.2"
@@ -5019,7 +5048,7 @@ dependencies = [
  "lazy_static",
  "libp2p",
  "lighthouse_metrics",
- "lru",
+ "lru 0.7.8",
  "parking_lot 0.12.3",
  "rand 0.8.5",
  "regex",
@@ -7272,8 +7301,11 @@ dependencies = [
  "kvdb",
  "kvdb-memorydb",
  "kvdb-rocksdb",
+ "lazy_static",
  "merkle_light",
  "merkle_tree",
+ "metrics",
+ "once_cell",
  "parking_lot 0.12.3",
  "rand 0.8.5",
  "rayon",

Cargo.toml

@@ -37,3 +37,7 @@ enr = { path = "version-meld/enr" }
 
 [profile.bench.package.'storage']
 debug = true
+
+[profile.dev]
+# enabling debug_assertions will make node fail to start because of checks in `clap`.
+debug-assertions = false

common/append_merkle/Cargo.toml

@@ -13,3 +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"

common/append_merkle/src/lib.rs

@@ -1,23 +1,30 @@
 mod merkle_tree;
+mod metrics;
+mod node_manager;
 mod proof;
 mod sha3;
 
 use anyhow::{anyhow, bail, Result};
+use itertools::Itertools;
 use std::cmp::Ordering;
 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;
 pub use crate::merkle_tree::{
     Algorithm, HashElement, MerkleTreeInitialData, MerkleTreeRead, ZERO_HASHES,
 };
+pub use crate::node_manager::{EmptyNodeDatabase, NodeDatabase, NodeManager, NodeTransaction};
 pub use proof::{Proof, RangeProof};
 pub use sha3::Sha3Algorithm;
 
 pub struct AppendMerkleTree<E: HashElement, A: Algorithm<E>> {
     /// Keep all the nodes in the latest version. `layers[0]` is the layer of leaves.
-    layers: Vec<Vec<E>>,
+    node_manager: NodeManager<E>,
     /// Keep the delta nodes that can be used to construct a history tree.
     /// The key is the root node of that version.
     delta_nodes_map: BTreeMap<u64, DeltaNodes<E>>,
@@ -35,13 +42,16 @@ pub struct AppendMerkleTree<E: HashElement, A: Algorithm<E>> {
 impl<E: HashElement, A: Algorithm<E>> AppendMerkleTree<E, A> {
     pub fn new(leaves: Vec<E>, leaf_height: usize, start_tx_seq: Option<u64>) -> Self {
         let mut merkle = Self {
-            layers: vec![leaves],
+            node_manager: NodeManager::new_dummy(),
             delta_nodes_map: BTreeMap::new(),
             root_to_tx_seq_map: HashMap::new(),
             min_depth: None,
             leaf_height,
             _a: Default::default(),
         };
+        merkle.node_manager.start_transaction();
+        merkle.node_manager.add_layer();
+        merkle.node_manager.append_nodes(0, &leaves);
         if merkle.leaves() == 0 {
             if let Some(seq) = start_tx_seq {
                 merkle.delta_nodes_map.insert(
@@ -51,10 +61,12 @@ impl<E: HashElement, A: Algorithm<E>> AppendMerkleTree<E, A> {
                     },
                 );
             }
+            merkle.node_manager.commit();
             return merkle;
         }
         // Reconstruct the whole tree.
         merkle.recompute(0, 0, None);
+        merkle.node_manager.commit();
         // Commit the first version in memory.
         // TODO(zz): Check when the roots become available.
         merkle.commit(start_tx_seq);
@@ -62,53 +74,44 @@ impl<E: HashElement, A: Algorithm<E>> AppendMerkleTree<E, A> {
     }
 
     pub fn new_with_subtrees(
-        initial_data: MerkleTreeInitialData<E>,
+        node_db: Arc<dyn NodeDatabase<E>>,
+        node_cache_capacity: usize,
         leaf_height: usize,
-        start_tx_seq: Option<u64>,
     ) -> Result<Self> {
         let mut merkle = Self {
-            layers: vec![vec![]],
+            node_manager: NodeManager::new(node_db, node_cache_capacity)?,
             delta_nodes_map: BTreeMap::new(),
             root_to_tx_seq_map: HashMap::new(),
             min_depth: None,
             leaf_height,
             _a: Default::default(),
         };
-        if initial_data.subtree_list.is_empty() {
+        if merkle.height() == 0 {
-            if let Some(seq) = start_tx_seq {
+            merkle.node_manager.start_transaction();
-                merkle.delta_nodes_map.insert(
+            merkle.node_manager.add_layer();
-                    seq,
+            merkle.node_manager.commit();
-                    DeltaNodes {
-                        right_most_nodes: vec![],
-                    },
-                );
-            }
-            return Ok(merkle);
-        }
-        merkle.append_subtree_list(initial_data.subtree_list)?;
-        merkle.commit(start_tx_seq);
-        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)
     }
 
     /// This is only used for the last chunk, so `leaf_height` is always 0 so far.
     pub fn new_with_depth(leaves: Vec<E>, depth: usize, start_tx_seq: Option<u64>) -> Self {
+        let mut node_manager = NodeManager::new_dummy();
+        node_manager.start_transaction();
         if leaves.is_empty() {
             // Create an empty merkle tree with `depth`.
             let mut merkle = Self {
-                layers: vec![vec![]; depth],
+                // dummy node manager for the last chunk.
+                node_manager,
                 delta_nodes_map: BTreeMap::new(),
                 root_to_tx_seq_map: HashMap::new(),
                 min_depth: Some(depth),
                 leaf_height: 0,
                 _a: Default::default(),
             };
+            for _ in 0..depth {
+                merkle.node_manager.add_layer();
+            }
             if let Some(seq) = start_tx_seq {
                 merkle.delta_nodes_map.insert(
                     seq,
@@ -117,20 +120,26 @@ impl<E: HashElement, A: Algorithm<E>> AppendMerkleTree<E, A> {
                     },
                 );
             }
+            merkle.node_manager.commit();
             merkle
         } else {
-            let mut layers = vec![vec![]; depth];
-            layers[0] = leaves;
             let mut merkle = Self {
-                layers,
+                // dummy node manager for the last chunk.
+                node_manager,
                 delta_nodes_map: BTreeMap::new(),
                 root_to_tx_seq_map: HashMap::new(),
                 min_depth: Some(depth),
                 leaf_height: 0,
                 _a: Default::default(),
             };
+            merkle.node_manager.add_layer();
+            merkle.append_nodes(0, &leaves);
+            for _ in 1..depth {
+                merkle.node_manager.add_layer();
+            }
             // Reconstruct the whole tree.
             merkle.recompute(0, 0, None);
+            merkle.node_manager.commit();
             // Commit the first version in memory.
             merkle.commit(start_tx_seq);
             merkle
@@ -138,22 +147,31 @@ 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;
         }
-        self.layers[0].push(new_leaf);
+        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, mut leaf_list: Vec<E>) {
+    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;
         }
+        self.node_manager.start_transaction();
         let start_index = self.leaves();
-        self.layers[0].append(&mut leaf_list);
+        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.
@@ -162,43 +180,57 @@ 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");
         }
+        self.node_manager.start_transaction();
         let start_index = self.leaves();
         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");
         }
+        self.node_manager.start_transaction();
         for (subtree_depth, subtree_root) in subtree_list {
             let start_index = self.leaves();
             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_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;
         }
-        if self.layers[0].is_empty() {
+        self.node_manager.start_transaction();
+        if self.layer_len(0) == 0 {
             // Special case for the first data.
-            self.layers[0].push(updated_leaf);
+            self.push_node(0, updated_leaf);
         } else {
-            *self.layers[0].last_mut().unwrap() = updated_leaf;
+            self.update_node(0, self.layer_len(0) - 1, updated_leaf);
         }
         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.
@@ -207,13 +239,17 @@ impl<E: HashElement, A: Algorithm<E>> AppendMerkleTree<E, A> {
     pub fn fill_leaf(&mut self, index: usize, leaf: E) {
         if leaf == E::null() {
             // fill leaf with null is not allowed.
-        } else if self.layers[0][index] == E::null() {
+        } else if self.node(0, index) == E::null() {
-            self.layers[0][index] = leaf;
+            self.node_manager.start_transaction();
+            self.update_node(0, index, leaf);
             self.recompute_after_fill_leaves(index, index + 1);
-        } else if self.layers[0][index] != leaf {
+            self.node_manager.commit();
+        } else if self.node(0, index) != leaf {
             panic!(
                 "Fill with invalid leaf, index={} was={:?} get={:?}",
-                index, self.layers[0][index], leaf
+                index,
+                self.node(0, index),
+                leaf
             );
         }
     }
@@ -226,6 +262,7 @@ impl<E: HashElement, A: Algorithm<E>> AppendMerkleTree<E, A> {
         &mut self,
         proof: RangeProof<E>,
     ) -> Result<Vec<(usize, usize, E)>> {
+        self.node_manager.start_transaction();
         let mut updated_nodes = Vec::new();
         let mut left_nodes = proof.left_proof.proof_nodes_in_tree();
         if left_nodes.len() >= self.leaf_height {
@@ -237,6 +274,7 @@ impl<E: HashElement, A: Algorithm<E>> AppendMerkleTree<E, A> {
             updated_nodes
                 .append(&mut self.fill_with_proof(right_nodes.split_off(self.leaf_height))?);
         }
+        self.node_manager.commit();
         Ok(updated_nodes)
     }
 
@@ -262,13 +300,16 @@ impl<E: HashElement, A: Algorithm<E>> AppendMerkleTree<E, A> {
         if tx_merkle_nodes.is_empty() {
             return Ok(Vec::new());
         }
+        self.node_manager.start_transaction();
         let mut position_and_data =
             proof.file_proof_nodes_in_tree(tx_merkle_nodes, tx_merkle_nodes_size);
         let start_index = (start_index >> self.leaf_height) as usize;
         for (i, (position, _)) in position_and_data.iter_mut().enumerate() {
             *position += start_index >> i;
         }
-        self.fill_with_proof(position_and_data)
+        let updated_nodes = self.fill_with_proof(position_and_data)?;
+        self.node_manager.commit();
+        Ok(updated_nodes)
     }
 
     /// This assumes that the proof leaf is no lower than the tree leaf. It holds for both SegmentProof and ChunkProof.
@@ -280,28 +321,27 @@ impl<E: HashElement, A: Algorithm<E>> AppendMerkleTree<E, A> {
         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];
+            if position > self.layer_len(i) {
-            if position > layer.len() {
                 bail!(
                     "proof position out of range, position={} layer.len()={}",
                     position,
-                    layer.len()
+                    self.layer_len(i)
                 );
             }
-            if position == layer.len() {
+            if position == self.layer_len(i) {
                 // skip padding node.
                 continue;
             }
-            if layer[position] == E::null() {
+            if self.node(i, position) == E::null() {
-                layer[position] = data.clone();
+                self.update_node(i, position, data.clone());
                 updated_nodes.push((i, position, data))
-            } else if layer[position] != data {
+            } else if self.node(i, position) != data {
                 // The last node in each layer may have changed in the tree.
                 trace!(
                     "conflict data layer={} position={} tree_data={:?} proof_data={:?}",
                     i,
                     position,
-                    layer[position],
+                    self.node(i, position),
                     data
                 );
             }
@@ -317,8 +357,8 @@ impl<E: HashElement, A: Algorithm<E>> AppendMerkleTree<E, A> {
         if position >= self.leaves() {
             bail!("Out of bound: position={} end={}", position, self.leaves());
         }
-        if self.layers[0][position] != E::null() {
+        if self.node(0, position) != E::null() {
-            Ok(Some(self.layers[0][position].clone()))
+            Ok(Some(self.node(0, position)))
         } else {
             // The leaf hash is unknown.
             Ok(None)
@@ -366,10 +406,11 @@ impl<E: HashElement, A: Algorithm<E>> AppendMerkleTree<E, A> {
                 return;
             }
             let mut right_most_nodes = Vec::new();
-            for layer in &self.layers {
+            for height in 0..self.height() {
-                right_most_nodes.push((layer.len() - 1, layer.last().unwrap().clone()));
+                let pos = self.layer_len(height) - 1;
+                right_most_nodes.push((pos, self.node(height, pos)));
             }
-            let root = self.root().clone();
+            let root = self.root();
             self.delta_nodes_map
                 .insert(tx_seq, DeltaNodes::new(right_most_nodes));
             self.root_to_tx_seq_map.insert(root, tx_seq);
@@ -377,8 +418,8 @@ impl<E: HashElement, A: Algorithm<E>> AppendMerkleTree<E, A> {
     }
 
     fn before_extend_layer(&mut self, height: usize) {
-        if height == self.layers.len() {
+        if height == self.height() {
-            self.layers.push(Vec::new());
+            self.node_manager.add_layer()
         }
     }
 
@@ -395,7 +436,6 @@ impl<E: HashElement, A: Algorithm<E>> AppendMerkleTree<E, A> {
     }
 
     /// Given a range of changed leaf nodes and recompute the tree.
-    /// Since this tree is append-only, we always compute to the end.
     fn recompute(
         &mut self,
         mut start_index: usize,
@@ -405,42 +445,51 @@ impl<E: HashElement, A: Algorithm<E>> AppendMerkleTree<E, A> {
         start_index >>= height;
         maybe_end_index = maybe_end_index.map(|end| end >> height);
         // Loop until we compute the new root and reach `tree_depth`.
-        while self.layers[height].len() > 1 || height < self.layers.len() - 1 {
+        while self.layer_len(height) > 1 || height < self.height() - 1 {
             let next_layer_start_index = start_index >> 1;
             if start_index % 2 == 1 {
                 start_index -= 1;
             }
 
-            let mut end_index = maybe_end_index.unwrap_or(self.layers[height].len());
+            let mut end_index = maybe_end_index.unwrap_or(self.layer_len(height));
-            if end_index % 2 == 1 && end_index != self.layers[height].len() {
+            if end_index % 2 == 1 && end_index != self.layer_len(height) {
                 end_index += 1;
             }
             let mut i = 0;
-            let mut iter = self.layers[height][start_index..end_index].chunks_exact(2);
+            let iter = self
+                .node_manager
+                .get_nodes(height, start_index, end_index)
+                .chunks(2);
             // We cannot modify the parent layer while iterating the child layer,
             // so just keep the changes and update them later.
             let mut parent_update = Vec::new();
-            while let Some([left, right]) = iter.next() {
+            for chunk_iter in &iter {
-                // If either left or right is null (unknown), we cannot compute the parent hash.
+                let chunk: Vec<_> = chunk_iter.collect();
-                // Note that if we are recompute a range of an existing tree,
+                if chunk.len() == 2 {
-                // we do not need to keep these possibly null parent. This is only saved
+                    let left = &chunk[0];
-                // for the case of constructing a new tree from the leaves.
+                    let right = &chunk[1];
-                let parent = if *left == E::null() || *right == E::null() {
+                    // If either left or right is null (unknown), we cannot compute the parent hash.
-                    E::null()
+                    // Note that if we are recompute a range of an existing tree,
+                    // we do not need to keep these possibly null parent. This is only saved
+                    // for the case of constructing a new tree from the leaves.
+                    let parent = if *left == E::null() || *right == E::null() {
+                        E::null()
+                    } else {
+                        A::parent(left, right)
+                    };
+                    parent_update.push((next_layer_start_index + i, parent));
+                    i += 1;
                 } else {
-                    A::parent(left, right)
+                    assert_eq!(chunk.len(), 1);
-                };
+                    let r = &chunk[0];
-                parent_update.push((next_layer_start_index + i, parent));
+                    // Same as above.
-                i += 1;
+                    let parent = if *r == E::null() {
-            }
+                        E::null()
-            if let [r] = iter.remainder() {
+                    } else {
-                // Same as above.
+                        A::parent_single(r, height + self.leaf_height)
-                let parent = if *r == E::null() {
+                    };
-                    E::null()
+                    parent_update.push((next_layer_start_index + i, parent));
-                } else {
+                }
-                    A::parent_single(r, height + self.leaf_height)
-                };
-                parent_update.push((next_layer_start_index + i, parent));
             }
             if !parent_update.is_empty() {
                 self.before_extend_layer(height + 1);
@@ -449,27 +498,27 @@ impl<E: HashElement, A: Algorithm<E>> AppendMerkleTree<E, A> {
             // we can just overwrite `last_changed_parent_index` with new values.
             let mut last_changed_parent_index = None;
             for (parent_index, parent) in parent_update {
-                match parent_index.cmp(&self.layers[height + 1].len()) {
+                match parent_index.cmp(&self.layer_len(height + 1)) {
                     Ordering::Less => {
                         // We do not overwrite with null.
                         if parent != E::null() {
-                            if self.layers[height + 1][parent_index] == E::null()
+                            if self.node(height + 1, parent_index) == E::null()
                                 // The last node in a layer can be updated.
-                                || (self.layers[height + 1][parent_index] != parent
+                                || (self.node(height + 1, parent_index) != parent
-                                    && parent_index == self.layers[height + 1].len() - 1)
+                                    && parent_index == self.layer_len(height + 1) - 1)
                             {
-                                self.layers[height + 1][parent_index] = parent;
+                                self.update_node(height + 1, parent_index, parent);
                                 last_changed_parent_index = Some(parent_index);
-                            } else if self.layers[height + 1][parent_index] != parent {
+                            } else if self.node(height + 1, parent_index) != parent {
                                 // Recompute changes a node in the middle. This should be impossible
                                 // if the inputs are valid.
                                 panic!("Invalid append merkle tree! height={} index={} expected={:?} get={:?}",
-                                       height + 1, parent_index, self.layers[height + 1][parent_index], parent);
+                                       height + 1, parent_index, self.node(height + 1, parent_index), parent);
                             }
                         }
                     }
                     Ordering::Equal => {
-                        self.layers[height + 1].push(parent);
+                        self.push_node(height + 1, parent);
                         last_changed_parent_index = Some(parent_index);
                     }
                     Ordering::Greater => {
@@ -500,10 +549,10 @@ impl<E: HashElement, A: Algorithm<E>> AppendMerkleTree<E, A> {
         for height in 0..(subtree_depth - 1) {
             self.before_extend_layer(height);
             let subtree_layer_size = 1 << (subtree_depth - 1 - height);
-            self.layers[height].append(&mut vec![E::null(); subtree_layer_size]);
+            self.append_nodes(height, &vec![E::null(); subtree_layer_size]);
         }
         self.before_extend_layer(subtree_depth - 1);
-        self.layers[subtree_depth - 1].push(subtree_root);
+        self.push_node(subtree_depth - 1, subtree_root);
         Ok(())
     }
 
@@ -514,23 +563,45 @@ impl<E: HashElement, A: Algorithm<E>> AppendMerkleTree<E, A> {
     }
 
     pub fn revert_to(&mut self, tx_seq: u64) -> Result<()> {
-        if self.layers[0].is_empty() {
+        if self.layer_len(0) == 0 {
             // Any previous state of an empty tree is always empty.
             return Ok(());
         }
+        self.node_manager.start_transaction();
         let delta_nodes = self
             .delta_nodes_map
             .get(&tx_seq)
-            .ok_or_else(|| anyhow!("tx_seq unavailable, root={:?}", tx_seq))?;
+            .ok_or_else(|| anyhow!("tx_seq unavailable, root={:?}", tx_seq))?
+            .clone();
         // Dropping the upper layers that are not in the old merkle tree.
-        self.layers.truncate(delta_nodes.right_most_nodes.len());
+        for height in (delta_nodes.right_most_nodes.len()..self.height()).rev() {
+            self.node_manager.truncate_layer(height);
+        }
         for (height, (last_index, right_most_node)) in
             delta_nodes.right_most_nodes.iter().enumerate()
         {
-            self.layers[height].truncate(*last_index + 1);
+            self.node_manager.truncate_nodes(height, *last_index + 1);
-            self.layers[height][*last_index] = right_most_node.clone();
+            self.update_node(height, *last_index, right_most_node.clone())
         }
         self.clear_after(tx_seq);
+        self.node_manager.commit();
+        Ok(())
+    }
+
+    // Revert to a tx_seq not in `delta_nodes_map`.
+    // This is needed to revert the last unfinished tx after restart.
+    pub fn revert_to_leaves(&mut self, leaves: usize) -> Result<()> {
+        self.node_manager.start_transaction();
+        for height in (0..self.height()).rev() {
+            let kept_nodes = leaves >> height;
+            if kept_nodes == 0 {
+                self.node_manager.truncate_layer(height);
+            } else {
+                self.node_manager.truncate_nodes(height, kept_nodes + 1);
+            }
+        }
+        self.recompute_after_append_leaves(leaves);
+        self.node_manager.commit();
         Ok(())
     }
 
@@ -550,17 +621,25 @@ impl<E: HashElement, A: Algorithm<E>> AppendMerkleTree<E, A> {
             bail!("empty tree");
         }
         Ok(HistoryTree {
-            layers: &self.layers,
+            node_manager: &self.node_manager,
             delta_nodes,
             leaf_height: self.leaf_height,
         })
     }
 
     pub fn reset(&mut self) {
-        self.layers = match self.min_depth {
+        self.node_manager.start_transaction();
-            None => vec![vec![]],
+        for height in (0..self.height()).rev() {
-            Some(depth) => vec![vec![]; depth],
+            self.node_manager.truncate_layer(height);
-        };
+        }
+        if let Some(depth) = self.min_depth {
+            for _ in 0..depth {
+                self.node_manager.add_layer();
+            }
+        } else {
+            self.node_manager.add_layer();
+        }
+        self.node_manager.commit();
     }
 
     fn clear_after(&mut self, tx_seq: u64) {
@@ -580,10 +659,10 @@ impl<E: HashElement, A: Algorithm<E>> AppendMerkleTree<E, A> {
     fn first_known_root_at(&self, index: usize) -> (usize, E) {
         let mut height = 0;
         let mut index_in_layer = index;
-        while height < self.layers.len() {
+        while height < self.height() {
             let node = self.node(height, index_in_layer);
             if !node.is_null() {
-                return (height + 1, node.clone());
+                return (height + 1, node);
             }
             height += 1;
             index_in_layer /= 2;
@@ -628,7 +707,7 @@ impl<E: HashElement> DeltaNodes<E> {
 
 pub struct HistoryTree<'m, E: HashElement> {
     /// A reference to the global tree nodes.
-    layers: &'m Vec<Vec<E>>,
+    node_manager: &'m NodeManager<E>,
     /// The delta nodes that are difference from `layers`.
     /// This could be a reference, we just take ownership for convenience.
     delta_nodes: &'m DeltaNodes<E>,
@@ -639,16 +718,18 @@ pub struct HistoryTree<'m, E: HashElement> {
 impl<E: HashElement, A: Algorithm<E>> MerkleTreeRead for AppendMerkleTree<E, A> {
     type E = E;
 
-    fn node(&self, layer: usize, index: usize) -> &Self::E {
+    fn node(&self, layer: usize, index: usize) -> Self::E {
-        &self.layers[layer][index]
+        self.node_manager
+            .get_node(layer, index)
+            .expect("index checked")
     }
 
     fn height(&self) -> usize {
-        self.layers.len()
+        self.node_manager.num_layers()
     }
 
     fn layer_len(&self, layer_height: usize) -> usize {
-        self.layers[layer_height].len()
+        self.node_manager.layer_size(layer_height)
     }
 
     fn padding_node(&self, height: usize) -> Self::E {
@@ -658,10 +739,13 @@ impl<E: HashElement, A: Algorithm<E>> MerkleTreeRead for AppendMerkleTree<E, A>
 
 impl<'a, E: HashElement> MerkleTreeRead for HistoryTree<'a, E> {
     type E = E;
-    fn node(&self, layer: usize, index: usize) -> &Self::E {
+    fn node(&self, layer: usize, index: usize) -> Self::E {
         match self.delta_nodes.get(layer, index).expect("range checked") {
-            Some(node) if *node != E::null() => node,
+            Some(node) if *node != E::null() => node.clone(),
-            _ => &self.layers[layer][index],
+            _ => self
+                .node_manager
+                .get_node(layer, index)
+                .expect("index checked"),
         }
     }
 
@@ -678,6 +762,22 @@ impl<'a, E: HashElement> MerkleTreeRead for HistoryTree<'a, E> {
     }
 }
 
+impl<E: HashElement, A: Algorithm<E>> MerkleTreeWrite for AppendMerkleTree<E, A> {
+    type E = E;
+
+    fn push_node(&mut self, layer: usize, node: Self::E) {
+        self.node_manager.push_node(layer, node);
+    }
+
+    fn append_nodes(&mut self, layer: usize, nodes: &[Self::E]) {
+        self.node_manager.append_nodes(layer, nodes);
+    }
+
+    fn update_node(&mut self, layer: usize, pos: usize, node: Self::E) {
+        self.node_manager.add_node(layer, pos, node);
+    }
+}
+
 #[macro_export]
 macro_rules! ensure_eq {
     ($given:expr, $expected:expr) => {
@@ -699,6 +799,7 @@ macro_rules! ensure_eq {
 #[cfg(test)]
 mod tests {
     use crate::merkle_tree::MerkleTreeRead;
+
     use crate::sha3::Sha3Algorithm;
     use crate::AppendMerkleTree;
     use ethereum_types::H256;

common/append_merkle/src/merkle_tree.rs

@@ -49,7 +49,7 @@ pub trait Algorithm<E: HashElement> {
 
 pub trait MerkleTreeRead {
     type E: HashElement;
-    fn node(&self, layer: usize, index: usize) -> &Self::E;
+    fn node(&self, layer: usize, index: usize) -> Self::E;
     fn height(&self) -> usize;
     fn layer_len(&self, layer_height: usize) -> usize;
     fn padding_node(&self, height: usize) -> Self::E;
@@ -58,7 +58,7 @@ pub trait MerkleTreeRead {
         self.layer_len(0)
     }
 
-    fn root(&self) -> &Self::E {
+    fn root(&self) -> Self::E {
         self.node(self.height() - 1, 0)
     }
 
@@ -70,16 +70,16 @@ pub trait MerkleTreeRead {
                 self.leaves()
             );
         }
-        if self.node(0, leaf_index) == &Self::E::null() {
+        if self.node(0, leaf_index) == Self::E::null() {
             bail!("Not ready to generate proof for leaf_index={}", leaf_index);
         }
         if self.height() == 1 {
-            return Proof::new(vec![self.root().clone(), self.root().clone()], vec![]);
+            return Proof::new(vec![self.root(), self.root().clone()], vec![]);
         }
         let mut lemma: Vec<Self::E> = Vec::with_capacity(self.height()); // path + root
         let mut path: Vec<bool> = Vec::with_capacity(self.height() - 2); // path - 1
         let mut index_in_layer = leaf_index;
-        lemma.push(self.node(0, leaf_index).clone());
+        lemma.push(self.node(0, leaf_index));
         for height in 0..(self.height() - 1) {
             trace!(
                 "gen_proof: height={} index={} hash={:?}",
@@ -93,15 +93,15 @@ pub trait MerkleTreeRead {
                     // TODO: This can be skipped if the tree size is available in validation.
                     lemma.push(self.padding_node(height));
                 } else {
-                    lemma.push(self.node(height, index_in_layer + 1).clone());
+                    lemma.push(self.node(height, index_in_layer + 1));
                 }
             } else {
                 path.push(false);
-                lemma.push(self.node(height, index_in_layer - 1).clone());
+                lemma.push(self.node(height, index_in_layer - 1));
             }
             index_in_layer >>= 1;
         }
-        lemma.push(self.root().clone());
+        lemma.push(self.root());
         if lemma.contains(&Self::E::null()) {
             bail!(
                 "Not enough data to generate proof, lemma={:?} path={:?}",
@@ -130,6 +130,13 @@ pub trait MerkleTreeRead {
     }
 }
 
+pub trait MerkleTreeWrite {
+    type E: HashElement;
+    fn push_node(&mut self, layer: usize, node: Self::E);
+    fn append_nodes(&mut self, layer: usize, nodes: &[Self::E]);
+    fn update_node(&mut self, layer: usize, pos: usize, node: Self::E);
+}
+
 /// This includes the data to reconstruct an `AppendMerkleTree` root where some nodes
 /// are `null`. Other intermediate nodes will be computed based on these known nodes.
 pub struct MerkleTreeInitialData<E: HashElement> {

common/append_merkle/src/metrics.rs

@@ -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");
+}

common/append_merkle/src/node_manager.rs

@@ -0,0 +1,219 @@
+use crate::HashElement;
+use anyhow::Result;
+use lru::LruCache;
+use std::any::Any;
+use std::num::NonZeroUsize;
+use std::sync::Arc;
+use tracing::error;
+
+pub struct NodeManager<E: HashElement> {
+    cache: LruCache<(usize, usize), E>,
+    layer_size: Vec<usize>,
+    db: Arc<dyn NodeDatabase<E>>,
+    db_tx: Option<Box<dyn NodeTransaction<E>>>,
+}
+
+impl<E: HashElement> NodeManager<E> {
+    pub fn new(db: Arc<dyn NodeDatabase<E>>, capacity: usize) -> Result<Self> {
+        let mut layer = 0;
+        let mut layer_size = Vec::new();
+        while let Some(size) = db.get_layer_size(layer)? {
+            layer_size.push(size);
+            layer += 1;
+        }
+        Ok(Self {
+            cache: LruCache::new(NonZeroUsize::new(capacity).expect("capacity should be non-zero")),
+            layer_size,
+            db,
+            db_tx: None,
+        })
+    }
+
+    pub fn new_dummy() -> Self {
+        Self {
+            cache: LruCache::unbounded(),
+            layer_size: vec![],
+            db: Arc::new(EmptyNodeDatabase {}),
+            db_tx: None,
+        }
+    }
+
+    pub fn push_node(&mut self, layer: usize, node: E) {
+        self.add_node(layer, self.layer_size[layer], node);
+        self.set_layer_size(layer, self.layer_size[layer] + 1);
+    }
+
+    pub fn append_nodes(&mut self, layer: usize, nodes: &[E]) {
+        let mut pos = self.layer_size[layer];
+        let mut saved_nodes = Vec::with_capacity(nodes.len());
+        for node in nodes {
+            self.cache.put((layer, pos), node.clone());
+            saved_nodes.push((layer, pos, node));
+            pos += 1;
+        }
+        self.set_layer_size(layer, pos);
+        self.db_tx().save_node_list(&saved_nodes);
+    }
+
+    pub fn get_node(&self, layer: usize, pos: usize) -> Option<E> {
+        match self.cache.peek(&(layer, pos)) {
+            Some(node) => Some(node.clone()),
+            None => self.db.get_node(layer, pos).unwrap_or_else(|e| {
+                error!("Failed to get node: {}", e);
+                None
+            }),
+        }
+    }
+
+    pub fn get_nodes(&self, layer: usize, start_pos: usize, end_pos: usize) -> NodeIterator<E> {
+        NodeIterator {
+            node_manager: self,
+            layer,
+            start_pos,
+            end_pos,
+        }
+    }
+
+    pub fn add_node(&mut self, layer: usize, pos: usize, node: E) {
+        // No need to insert if the value is unchanged.
+        if self.cache.get(&(layer, pos)) != Some(&node) {
+            self.db_tx().save_node(layer, pos, &node);
+            self.cache.put((layer, pos), node);
+        }
+    }
+
+    pub fn add_layer(&mut self) {
+        self.layer_size.push(0);
+        let layer = self.layer_size.len() - 1;
+        self.db_tx().save_layer_size(layer, 0);
+    }
+
+    pub fn layer_size(&self, layer: usize) -> usize {
+        self.layer_size[layer]
+    }
+
+    pub fn num_layers(&self) -> usize {
+        self.layer_size.len()
+    }
+
+    pub fn truncate_nodes(&mut self, layer: usize, pos_end: usize) {
+        let mut removed_nodes = Vec::new();
+        for pos in pos_end..self.layer_size[layer] {
+            self.cache.pop(&(layer, pos));
+            removed_nodes.push((layer, pos));
+        }
+        self.db_tx().remove_node_list(&removed_nodes);
+        self.set_layer_size(layer, pos_end);
+    }
+
+    pub fn truncate_layer(&mut self, layer: usize) {
+        self.truncate_nodes(layer, 0);
+        if layer == self.num_layers() - 1 {
+            self.layer_size.pop();
+            self.db_tx().remove_layer_size(layer);
+        }
+    }
+
+    pub fn start_transaction(&mut self) {
+        if self.db_tx.is_some() {
+            error!("start new tx before commit");
+            panic!("start new tx before commit");
+        }
+        self.db_tx = Some(self.db.start_transaction());
+    }
+
+    pub fn commit(&mut self) {
+        let tx = match self.db_tx.take() {
+            Some(tx) => tx,
+            None => {
+                error!("db_tx is None");
+                return;
+            }
+        };
+        if let Err(e) = self.db.commit(tx) {
+            error!("Failed to commit db transaction: {}", e);
+        }
+    }
+
+    fn db_tx(&mut self) -> &mut dyn NodeTransaction<E> {
+        (*self.db_tx.as_mut().expect("tx checked")).as_mut()
+    }
+
+    fn set_layer_size(&mut self, layer: usize, size: usize) {
+        self.layer_size[layer] = size;
+        self.db_tx().save_layer_size(layer, size);
+    }
+}
+
+pub struct NodeIterator<'a, E: HashElement> {
+    node_manager: &'a NodeManager<E>,
+    layer: usize,
+    start_pos: usize,
+    end_pos: usize,
+}
+
+impl<'a, E: HashElement> Iterator for NodeIterator<'a, E> {
+    type Item = E;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        if self.start_pos < self.end_pos {
+            let r = self.node_manager.get_node(self.layer, self.start_pos);
+            self.start_pos += 1;
+            r
+        } else {
+            None
+        }
+    }
+}
+
+pub trait NodeDatabase<E: HashElement>: Send + Sync {
+    fn get_node(&self, layer: usize, pos: usize) -> Result<Option<E>>;
+    fn get_layer_size(&self, layer: usize) -> Result<Option<usize>>;
+    fn start_transaction(&self) -> Box<dyn NodeTransaction<E>>;
+    fn commit(&self, tx: Box<dyn NodeTransaction<E>>) -> Result<()>;
+}
+
+pub trait NodeTransaction<E: HashElement>: Send + Sync {
+    fn save_node(&mut self, layer: usize, pos: usize, node: &E);
+    /// `nodes` are a list of tuples `(layer, pos, node)`.
+    fn save_node_list(&mut self, nodes: &[(usize, usize, &E)]);
+    fn remove_node_list(&mut self, nodes: &[(usize, usize)]);
+    fn save_layer_size(&mut self, layer: usize, size: usize);
+    fn remove_layer_size(&mut self, layer: usize);
+
+    fn into_any(self: Box<Self>) -> Box<dyn Any>;
+}
+
+/// A dummy database structure for in-memory merkle tree that will not read/write db.
+pub struct EmptyNodeDatabase {}
+pub struct EmptyNodeTransaction {}
+impl<E: HashElement> NodeDatabase<E> for EmptyNodeDatabase {
+    fn get_node(&self, _layer: usize, _pos: usize) -> Result<Option<E>> {
+        Ok(None)
+    }
+    fn get_layer_size(&self, _layer: usize) -> Result<Option<usize>> {
+        Ok(None)
+    }
+    fn start_transaction(&self) -> Box<dyn NodeTransaction<E>> {
+        Box::new(EmptyNodeTransaction {})
+    }
+    fn commit(&self, _tx: Box<dyn NodeTransaction<E>>) -> Result<()> {
+        Ok(())
+    }
+}
+
+impl<E: HashElement> NodeTransaction<E> for EmptyNodeTransaction {
+    fn save_node(&mut self, _layer: usize, _pos: usize, _node: &E) {}
+
+    fn save_node_list(&mut self, _nodes: &[(usize, usize, &E)]) {}
+
+    fn remove_node_list(&mut self, _nodes: &[(usize, usize)]) {}
+
+    fn save_layer_size(&mut self, _layer: usize, _size: usize) {}
+
+    fn remove_layer_size(&mut self, _layer: usize) {}
+
+    fn into_any(self: Box<Self>) -> Box<dyn Any> {
+        self
+    }
+}

node/network/src/behaviour/gossip_cache.rs

@@ -20,6 +20,8 @@ pub struct GossipCache {
     topic_msgs: HashMap<GossipTopic, HashMap<Vec<u8>, Key>>,
     /// Timeout for Example messages.
     example: Option<Duration>,
+    /// Timeout for NewFile messages.
+    new_file: Option<Duration>,
     /// Timeout for FindFile messages.
     find_file: Option<Duration>,
     /// Timeout for FindChunks messages.
@@ -37,6 +39,8 @@ pub struct GossipCacheBuilder {
     default_timeout: Option<Duration>,
     /// Timeout for Example messages.
     example: Option<Duration>,
+    /// Timeout for NewFile messages.
+    new_file: Option<Duration>,
     /// Timeout for blocks FindFile messages.
     find_file: Option<Duration>,
     /// Timeout for blocks FindChunks messages.
@@ -64,6 +68,12 @@ impl GossipCacheBuilder {
         self
     }
 
+    /// Timeout for NewFile messages.
+    pub fn new_file_timeout(mut self, timeout: Duration) -> Self {
+        self.new_file = Some(timeout);
+        self
+    }
+
     /// Timeout for FindFile messages.
     pub fn find_file_timeout(mut self, timeout: Duration) -> Self {
         self.find_file = Some(timeout);
@@ -98,6 +108,7 @@ impl GossipCacheBuilder {
         let GossipCacheBuilder {
             default_timeout,
             example,
+            new_file,
             find_file,
             find_chunks,
             announce_file,
@@ -109,6 +120,7 @@ impl GossipCacheBuilder {
             expirations: DelayQueue::default(),
             topic_msgs: HashMap::default(),
             example: example.or(default_timeout),
+            new_file: new_file.or(default_timeout),
             find_file: find_file.or(default_timeout),
             find_chunks: find_chunks.or(default_timeout),
             announce_file: announce_file.or(default_timeout),
@@ -129,6 +141,7 @@ impl GossipCache {
     pub fn insert(&mut self, topic: GossipTopic, data: Vec<u8>) {
         let expire_timeout = match topic.kind() {
             GossipKind::Example => self.example,
+            GossipKind::NewFile => self.new_file,
             GossipKind::FindFile => self.find_file,
             GossipKind::FindChunks => self.find_chunks,
             GossipKind::AnnounceFile => self.announce_file,

node/network/src/behaviour/mod.rs

@@ -6,6 +6,7 @@ use crate::peer_manager::{
     ConnectionDirection, PeerManager, PeerManagerEvent,
 };
 use crate::rpc::methods::DataByHashRequest;
+use crate::rpc::methods::FileAnnouncement;
 use crate::rpc::methods::GetChunksRequest;
 use crate::rpc::*;
 use crate::service::Context as ServiceContext;
@@ -232,6 +233,9 @@ impl<AppReqId: ReqId> Behaviour<AppReqId> {
             let topic: Topic = GossipTopic::new(kind, GossipEncoding::default()).into();
             topic.hash()
         };
+        params
+            .topics
+            .insert(get_hash(GossipKind::NewFile), TopicScoreParams::default());
         params
             .topics
             .insert(get_hash(GossipKind::FindFile), TopicScoreParams::default());
@@ -543,6 +547,9 @@ impl<AppReqId: ReqId> Behaviour<AppReqId> {
             Request::DataByHash { .. } => {
                 metrics::inc_counter_vec(&metrics::TOTAL_RPC_REQUESTS, &["data_by_hash"])
             }
+            Request::AnnounceFile { .. } => {
+                metrics::inc_counter_vec(&metrics::TOTAL_RPC_REQUESTS, &["announce_file"])
+            }
             Request::GetChunks { .. } => {
                 metrics::inc_counter_vec(&metrics::TOTAL_RPC_REQUESTS, &["get_chunks"])
             }
@@ -755,6 +762,9 @@ where
                     InboundRequest::DataByHash(req) => {
                         self.propagate_request(peer_request_id, peer_id, Request::DataByHash(req))
                     }
+                    InboundRequest::AnnounceFile(req) => {
+                        self.propagate_request(peer_request_id, peer_id, Request::AnnounceFile(req))
+                    }
                     InboundRequest::GetChunks(req) => {
                         self.propagate_request(peer_request_id, peer_id, Request::GetChunks(req))
                     }
@@ -969,6 +979,8 @@ pub enum Request {
     Status(StatusMessage),
     /// A data by hash request.
     DataByHash(DataByHashRequest),
+    /// An AnnounceFile message.
+    AnnounceFile(FileAnnouncement),
     /// A GetChunks request.
     GetChunks(GetChunksRequest),
 }
@@ -978,6 +990,7 @@ impl std::convert::From<Request> for OutboundRequest {
         match req {
             Request::Status(s) => OutboundRequest::Status(s),
             Request::DataByHash(r) => OutboundRequest::DataByHash(r),
+            Request::AnnounceFile(r) => OutboundRequest::AnnounceFile(r),
             Request::GetChunks(r) => OutboundRequest::GetChunks(r),
         }
     }

node/network/src/lib.rs

@@ -93,7 +93,10 @@ pub use peer_manager::{
 };
 pub use service::{load_private_key, Context, Libp2pEvent, Service, NETWORK_KEY_FILENAME};
 
-pub const PROTOCOL_VERSION: [u8; 3] = [0, 1, 0];
+/// Defines the current P2P protocol version.
+/// - v1: Broadcast FindFile & AnnounceFile messages in the whole network, which caused network too heavey.
+/// - v2: Publish NewFile to neighbors only and announce file via RPC message.
+pub const PROTOCOL_VERSION: [u8; 3] = [0, 2, 0];
 
 /// Application level requests sent to the network.
 #[derive(Debug, Clone, Copy)]

node/network/src/peer_manager/mod.rs

@@ -460,6 +460,7 @@ impl PeerManager {
                     Protocol::Goodbye => PeerAction::LowToleranceError,
                     Protocol::Status => PeerAction::LowToleranceError,
                     Protocol::DataByHash => PeerAction::MidToleranceError,
+                    Protocol::AnnounceFile => PeerAction::MidToleranceError,
                     Protocol::GetChunks => PeerAction::MidToleranceError,
                 },
             },
@@ -474,6 +475,7 @@ impl PeerManager {
                     Protocol::Goodbye => return,
                     Protocol::Status => PeerAction::LowToleranceError,
                     Protocol::DataByHash => return,
+                    Protocol::AnnounceFile => return,
                     Protocol::GetChunks => return,
                 }
             }
@@ -488,6 +490,7 @@ impl PeerManager {
                     Protocol::Goodbye => return,
                     Protocol::Status => return,
                     Protocol::DataByHash => PeerAction::MidToleranceError,
+                    Protocol::AnnounceFile => PeerAction::MidToleranceError,
                     Protocol::GetChunks => PeerAction::MidToleranceError,
                 },
             },

node/network/src/rpc/codec/ssz_snappy.rs

@@ -159,6 +159,7 @@ impl Encoder<OutboundRequest> for SSZSnappyOutboundCodec {
             OutboundRequest::Goodbye(req) => req.as_ssz_bytes(),
             OutboundRequest::Ping(req) => req.as_ssz_bytes(),
             OutboundRequest::DataByHash(req) => req.hashes.as_ssz_bytes(),
+            OutboundRequest::AnnounceFile(req) => req.as_ssz_bytes(),
             OutboundRequest::GetChunks(req) => req.as_ssz_bytes(),
         };
         // SSZ encoded bytes should be within `max_packet_size`
@@ -346,6 +347,9 @@ fn handle_v1_request(
         Protocol::DataByHash => Ok(Some(InboundRequest::DataByHash(DataByHashRequest {
             hashes: VariableList::from_ssz_bytes(decoded_buffer)?,
         }))),
+        Protocol::AnnounceFile => Ok(Some(InboundRequest::AnnounceFile(
+            FileAnnouncement::from_ssz_bytes(decoded_buffer)?,
+        ))),
         Protocol::GetChunks => Ok(Some(InboundRequest::GetChunks(
             GetChunksRequest::from_ssz_bytes(decoded_buffer)?,
         ))),
@@ -373,6 +377,10 @@ fn handle_v1_response(
         Protocol::DataByHash => Ok(Some(RPCResponse::DataByHash(Box::new(
             ZgsData::from_ssz_bytes(decoded_buffer)?,
         )))),
+        // This case should be unreachable as `AnnounceFile` has no response.
+        Protocol::AnnounceFile => Err(RPCError::InvalidData(
+            "AnnounceFile RPC message has no valid response".to_string(),
+        )),
         Protocol::GetChunks => Ok(Some(RPCResponse::Chunks(
             ChunkArrayWithProof::from_ssz_bytes(decoded_buffer)?,
         ))),

node/network/src/rpc/methods.rs

@@ -178,6 +178,14 @@ pub struct DataByHashRequest {
     pub hashes: VariableList<Hash256, MaxRequestBlocks>,
 }
 
+// The message of `AnnounceFile` RPC message.
+#[derive(Encode, Decode, Clone, Debug, PartialEq, Eq)]
+pub struct FileAnnouncement {
+    pub tx_id: TxID,
+    pub num_shard: usize,
+    pub shard_id: usize,
+}
+
 /// Request a chunk array from a peer.
 #[derive(Encode, Decode, Clone, Debug, PartialEq, Eq)]
 pub struct GetChunksRequest {

node/network/src/rpc/mod.rs

@@ -118,6 +118,7 @@ impl<Id: ReqId> RPC<Id> {
             .n_every(Protocol::Status, 5, Duration::from_secs(15))
             .one_every(Protocol::Goodbye, Duration::from_secs(10))
             .n_every(Protocol::DataByHash, 128, Duration::from_secs(10))
+            .n_every(Protocol::AnnounceFile, 256, Duration::from_secs(10))
             .n_every(Protocol::GetChunks, 4096, Duration::from_secs(10))
             .build()
             .expect("Configuration parameters are valid");

node/network/src/rpc/outbound.rs

@@ -34,6 +34,7 @@ pub enum OutboundRequest {
     Goodbye(GoodbyeReason),
     Ping(Ping),
     DataByHash(DataByHashRequest),
+    AnnounceFile(FileAnnouncement),
     GetChunks(GetChunksRequest),
 }
 
@@ -72,6 +73,11 @@ impl OutboundRequest {
                 Version::V1,
                 Encoding::SSZSnappy,
             )],
+            OutboundRequest::AnnounceFile(_) => vec![ProtocolId::new(
+                Protocol::AnnounceFile,
+                Version::V1,
+                Encoding::SSZSnappy,
+            )],
             OutboundRequest::GetChunks(_) => vec![ProtocolId::new(
                 Protocol::GetChunks,
                 Version::V1,
@@ -89,6 +95,7 @@ impl OutboundRequest {
             OutboundRequest::Goodbye(_) => 0,
             OutboundRequest::Ping(_) => 1,
             OutboundRequest::DataByHash(req) => req.hashes.len() as u64,
+            OutboundRequest::AnnounceFile(_) => 0,
             OutboundRequest::GetChunks(_) => 1,
         }
     }
@@ -100,6 +107,7 @@ impl OutboundRequest {
             OutboundRequest::Goodbye(_) => Protocol::Goodbye,
             OutboundRequest::Ping(_) => Protocol::Ping,
             OutboundRequest::DataByHash(_) => Protocol::DataByHash,
+            OutboundRequest::AnnounceFile(_) => Protocol::AnnounceFile,
             OutboundRequest::GetChunks(_) => Protocol::GetChunks,
         }
     }
@@ -114,6 +122,7 @@ impl OutboundRequest {
             OutboundRequest::Status(_) => unreachable!(),
             OutboundRequest::Goodbye(_) => unreachable!(),
             OutboundRequest::Ping(_) => unreachable!(),
+            OutboundRequest::AnnounceFile(_) => unreachable!(),
             OutboundRequest::GetChunks(_) => unreachable!(),
         }
     }
@@ -170,6 +179,9 @@ impl std::fmt::Display for OutboundRequest {
             OutboundRequest::DataByHash(req) => {
                 write!(f, "Data by hash: {:?}", req)
             }
+            OutboundRequest::AnnounceFile(req) => {
+                write!(f, "AnnounceFile: {:?}", req)
+            }
             OutboundRequest::GetChunks(req) => {
                 write!(f, "GetChunks: {:?}", req)
             }

node/network/src/rpc/protocol.rs

@@ -91,6 +91,8 @@ pub enum Protocol {
     /// TODO
     DataByHash,
 
+    /// The file announce protocol.
+    AnnounceFile,
     /// The Chunk sync protocol.
     GetChunks,
 }
@@ -115,6 +117,7 @@ impl std::fmt::Display for Protocol {
             Protocol::Goodbye => "goodbye",
             Protocol::Ping => "ping",
             Protocol::DataByHash => "data_by_hash",
+            Protocol::AnnounceFile => "announce_file",
             Protocol::GetChunks => "get_chunks",
         };
         f.write_str(repr)
@@ -155,6 +158,7 @@ impl UpgradeInfo for RPCProtocol {
             ProtocolId::new(Protocol::Goodbye, Version::V1, Encoding::SSZSnappy),
             ProtocolId::new(Protocol::Ping, Version::V1, Encoding::SSZSnappy),
             ProtocolId::new(Protocol::DataByHash, Version::V1, Encoding::SSZSnappy),
+            ProtocolId::new(Protocol::AnnounceFile, Version::V1, Encoding::SSZSnappy),
             ProtocolId::new(Protocol::GetChunks, Version::V1, Encoding::SSZSnappy),
         ]
     }
@@ -216,6 +220,10 @@ impl ProtocolId {
                 // TODO
                 RpcLimits::new(1, *DATA_BY_HASH_REQUEST_MAX)
             }
+            Protocol::AnnounceFile => RpcLimits::new(
+                <FileAnnouncement as Encode>::ssz_fixed_len(),
+                <FileAnnouncement as Encode>::ssz_fixed_len(),
+            ),
             Protocol::GetChunks => RpcLimits::new(
                 <GetChunksRequest as Encode>::ssz_fixed_len(),
                 <GetChunksRequest as Encode>::ssz_fixed_len(),
@@ -243,6 +251,7 @@ impl ProtocolId {
                 <ZgsData as Encode>::ssz_fixed_len(),
             ),
 
+            Protocol::AnnounceFile => RpcLimits::new(0, 0), // AnnounceFile request has no response
             Protocol::GetChunks => RpcLimits::new(*CHUNKS_RESPONSE_MIN, *CHUNKS_RESPONSE_MAX),
         }
     }
@@ -325,6 +334,7 @@ pub enum InboundRequest {
     Goodbye(GoodbyeReason),
     Ping(Ping),
     DataByHash(DataByHashRequest),
+    AnnounceFile(FileAnnouncement),
     GetChunks(GetChunksRequest),
 }
 
@@ -363,6 +373,11 @@ impl InboundRequest {
                 Version::V1,
                 Encoding::SSZSnappy,
             )],
+            InboundRequest::AnnounceFile(_) => vec![ProtocolId::new(
+                Protocol::AnnounceFile,
+                Version::V1,
+                Encoding::SSZSnappy,
+            )],
             InboundRequest::GetChunks(_) => vec![ProtocolId::new(
                 Protocol::GetChunks,
                 Version::V1,
@@ -380,6 +395,7 @@ impl InboundRequest {
             InboundRequest::Goodbye(_) => 0,
             InboundRequest::DataByHash(req) => req.hashes.len() as u64,
             InboundRequest::Ping(_) => 1,
+            InboundRequest::AnnounceFile(_) => 0,
             InboundRequest::GetChunks(_) => 1,
         }
     }
@@ -391,6 +407,7 @@ impl InboundRequest {
             InboundRequest::Goodbye(_) => Protocol::Goodbye,
             InboundRequest::Ping(_) => Protocol::Ping,
             InboundRequest::DataByHash(_) => Protocol::DataByHash,
+            InboundRequest::AnnounceFile(_) => Protocol::AnnounceFile,
             InboundRequest::GetChunks(_) => Protocol::GetChunks,
         }
     }
@@ -405,6 +422,7 @@ impl InboundRequest {
             InboundRequest::Status(_) => unreachable!(),
             InboundRequest::Goodbye(_) => unreachable!(),
             InboundRequest::Ping(_) => unreachable!(),
+            InboundRequest::AnnounceFile(_) => unreachable!(),
             InboundRequest::GetChunks(_) => unreachable!(),
         }
     }
@@ -523,6 +541,9 @@ impl std::fmt::Display for InboundRequest {
             InboundRequest::DataByHash(req) => {
                 write!(f, "Data by hash: {:?}", req)
             }
+            InboundRequest::AnnounceFile(req) => {
+                write!(f, "Announce File: {:?}", req)
+            }
             InboundRequest::GetChunks(req) => {
                 write!(f, "Get Chunks: {:?}", req)
             }

node/network/src/rpc/rate_limiter.rs

@@ -68,6 +68,8 @@ pub struct RPCRateLimiter {
     status_rl: Limiter<PeerId>,
     /// DataByHash rate limiter.
     data_by_hash_rl: Limiter<PeerId>,
+    /// AnnounceFile rate limiter.
+    announce_file_rl: Limiter<PeerId>,
     /// GetChunks rate limiter.
     get_chunks_rl: Limiter<PeerId>,
 }
@@ -91,6 +93,8 @@ pub struct RPCRateLimiterBuilder {
     status_quota: Option<Quota>,
     /// Quota for the DataByHash protocol.
     data_by_hash_quota: Option<Quota>,
+    /// Quota for the AnnounceFile protocol.
+    announce_file_quota: Option<Quota>,
     /// Quota for the GetChunks protocol.
     get_chunks_quota: Option<Quota>,
 }
@@ -109,6 +113,7 @@ impl RPCRateLimiterBuilder {
             Protocol::Status => self.status_quota = q,
             Protocol::Goodbye => self.goodbye_quota = q,
             Protocol::DataByHash => self.data_by_hash_quota = q,
+            Protocol::AnnounceFile => self.announce_file_quota = q,
             Protocol::GetChunks => self.get_chunks_quota = q,
         }
         self
@@ -145,6 +150,9 @@ impl RPCRateLimiterBuilder {
         let data_by_hash_quota = self
             .data_by_hash_quota
             .ok_or("DataByHash quota not specified")?;
+        let announce_file_quota = self
+            .announce_file_quota
+            .ok_or("AnnounceFile quota not specified")?;
         let get_chunks_quota = self
             .get_chunks_quota
             .ok_or("GetChunks quota not specified")?;
@@ -154,6 +162,7 @@ impl RPCRateLimiterBuilder {
         let status_rl = Limiter::from_quota(status_quota)?;
         let goodbye_rl = Limiter::from_quota(goodbye_quota)?;
         let data_by_hash_rl = Limiter::from_quota(data_by_hash_quota)?;
+        let announce_file_rl = Limiter::from_quota(announce_file_quota)?;
         let get_chunks_rl = Limiter::from_quota(get_chunks_quota)?;
 
         // check for peers to prune every 30 seconds, starting in 30 seconds
@@ -166,6 +175,7 @@ impl RPCRateLimiterBuilder {
             status_rl,
             goodbye_rl,
             data_by_hash_rl,
+            announce_file_rl,
             get_chunks_rl,
             init_time: Instant::now(),
         })
@@ -210,6 +220,7 @@ impl RPCRateLimiter {
             Protocol::Status => &mut self.status_rl,
             Protocol::Goodbye => &mut self.goodbye_rl,
             Protocol::DataByHash => &mut self.data_by_hash_rl,
+            Protocol::AnnounceFile => &mut self.announce_file_rl,
             Protocol::GetChunks => &mut self.get_chunks_rl,
         };
         check(limiter)

node/network/src/types/mod.rs

@@ -7,7 +7,7 @@ pub type Enr = discv5::enr::Enr<discv5::enr::CombinedKey>;
 
 pub use globals::NetworkGlobals;
 pub use pubsub::{
-    AnnounceChunks, AnnounceFile, AnnounceShardConfig, FindChunks, FindFile, HasSignature,
+    AnnounceChunks, AnnounceFile, AnnounceShardConfig, FindChunks, FindFile, HasSignature, NewFile,
     PubsubMessage, SignedAnnounceChunks, SignedAnnounceFile, SignedAnnounceShardConfig,
     SignedMessage, SnappyTransform,
 };

node/network/src/types/pubsub.rs

@@ -114,9 +114,22 @@ impl ssz::Decode for WrappedPeerId {
     }
 }
 
+/// Published when file uploaded or completed to sync from other peers.
+#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)]
+pub struct NewFile {
+    pub tx_id: TxID,
+    pub num_shard: usize,
+    pub shard_id: usize,
+    pub timestamp: u32,
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)]
 pub struct FindFile {
     pub tx_id: TxID,
+    pub num_shard: usize,
+    pub shard_id: usize,
+    /// Indicates whether publish to neighboar nodes only.
+    pub neighbors_only: bool,
     pub timestamp: u32,
 }
 
@@ -205,6 +218,7 @@ type SignedAnnounceFiles = Vec<SignedAnnounceFile>;
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum PubsubMessage {
     ExampleMessage(u64),
+    NewFile(NewFile),
     FindFile(FindFile),
     FindChunks(FindChunks),
     AnnounceFile(Vec<SignedAnnounceFile>),
@@ -283,6 +297,7 @@ impl PubsubMessage {
     pub fn kind(&self) -> GossipKind {
         match self {
             PubsubMessage::ExampleMessage(_) => GossipKind::Example,
+            PubsubMessage::NewFile(_) => GossipKind::NewFile,
             PubsubMessage::FindFile(_) => GossipKind::FindFile,
             PubsubMessage::FindChunks(_) => GossipKind::FindChunks,
             PubsubMessage::AnnounceFile(_) => GossipKind::AnnounceFile,
@@ -309,6 +324,9 @@ impl PubsubMessage {
                     GossipKind::Example => Ok(PubsubMessage::ExampleMessage(
                         u64::from_ssz_bytes(data).map_err(|e| format!("{:?}", e))?,
                     )),
+                    GossipKind::NewFile => Ok(PubsubMessage::NewFile(
+                        NewFile::from_ssz_bytes(data).map_err(|e| format!("{:?}", e))?,
+                    )),
                     GossipKind::FindFile => Ok(PubsubMessage::FindFile(
                         FindFile::from_ssz_bytes(data).map_err(|e| format!("{:?}", e))?,
                     )),
@@ -341,6 +359,7 @@ impl PubsubMessage {
         // messages for us.
         match &self {
             PubsubMessage::ExampleMessage(data) => data.as_ssz_bytes(),
+            PubsubMessage::NewFile(data) => data.as_ssz_bytes(),
             PubsubMessage::FindFile(data) => data.as_ssz_bytes(),
             PubsubMessage::FindChunks(data) => data.as_ssz_bytes(),
             PubsubMessage::AnnounceFile(data) => data.as_ssz_bytes(),
@@ -356,6 +375,9 @@ impl std::fmt::Display for PubsubMessage {
             PubsubMessage::ExampleMessage(msg) => {
                 write!(f, "Example message: {}", msg)
             }
+            PubsubMessage::NewFile(msg) => {
+                write!(f, "NewFile message: {:?}", msg)
+            }
             PubsubMessage::FindFile(msg) => {
                 write!(f, "FindFile message: {:?}", msg)
             }

node/network/src/types/topics.rs

@@ -8,13 +8,15 @@ use strum::AsRefStr;
 pub const TOPIC_PREFIX: &str = "eth2";
 pub const SSZ_SNAPPY_ENCODING_POSTFIX: &str = "ssz_snappy";
 pub const EXAMPLE_TOPIC: &str = "example";
+pub const NEW_FILE_TOPIC: &str = "new_file";
 pub const FIND_FILE_TOPIC: &str = "find_file";
 pub const FIND_CHUNKS_TOPIC: &str = "find_chunks";
 pub const ANNOUNCE_FILE_TOPIC: &str = "announce_file";
 pub const ANNOUNCE_CHUNKS_TOPIC: &str = "announce_chunks";
 pub const ANNOUNCE_SHARD_CONFIG_TOPIC: &str = "announce_shard_config";
 
-pub const CORE_TOPICS: [GossipKind; 4] = [
+pub const CORE_TOPICS: [GossipKind; 5] = [
+    GossipKind::NewFile,
     GossipKind::FindFile,
     GossipKind::FindChunks,
     GossipKind::AnnounceFile,
@@ -37,6 +39,7 @@ pub struct GossipTopic {
 #[strum(serialize_all = "snake_case")]
 pub enum GossipKind {
     Example,
+    NewFile,
     FindFile,
     FindChunks,
     AnnounceFile,
@@ -77,6 +80,7 @@ impl GossipTopic {
 
             let kind = match topic_parts[2] {
                 EXAMPLE_TOPIC => GossipKind::Example,
+                NEW_FILE_TOPIC => GossipKind::NewFile,
                 FIND_FILE_TOPIC => GossipKind::FindFile,
                 FIND_CHUNKS_TOPIC => GossipKind::FindChunks,
                 ANNOUNCE_FILE_TOPIC => GossipKind::AnnounceFile,
@@ -106,6 +110,7 @@ impl From<GossipTopic> for String {
 
         let kind = match topic.kind {
             GossipKind::Example => EXAMPLE_TOPIC,
+            GossipKind::NewFile => NEW_FILE_TOPIC,
             GossipKind::FindFile => FIND_FILE_TOPIC,
             GossipKind::FindChunks => FIND_CHUNKS_TOPIC,
             GossipKind::AnnounceFile => ANNOUNCE_FILE_TOPIC,
@@ -125,6 +130,7 @@ impl std::fmt::Display for GossipTopic {
 
         let kind = match self.kind {
             GossipKind::Example => EXAMPLE_TOPIC,
+            GossipKind::NewFile => NEW_FILE_TOPIC,
             GossipKind::FindFile => FIND_FILE_TOPIC,
             GossipKind::FindChunks => FIND_CHUNKS_TOPIC,
             GossipKind::AnnounceFile => ANNOUNCE_FILE_TOPIC,

node/router/src/libp2p_event_handler.rs

@@ -5,7 +5,8 @@ use std::{ops::Neg, sync::Arc};
 use chunk_pool::ChunkPoolMessage;
 use file_location_cache::FileLocationCache;
 use network::multiaddr::Protocol;
-use network::types::{AnnounceShardConfig, SignedAnnounceShardConfig};
+use network::rpc::methods::FileAnnouncement;
+use network::types::{AnnounceShardConfig, NewFile, SignedAnnounceShardConfig};
 use network::{
     rpc::StatusMessage,
     types::{
@@ -29,6 +30,11 @@ use crate::peer_manager::PeerManager;
 use crate::Config;
 
 lazy_static::lazy_static! {
+    /// Timeout to publish NewFile message to neighbor nodes.
+    pub static ref NEW_FILE_TIMEOUT: chrono::Duration = chrono::Duration::seconds(30);
+    /// Timeout to publish FindFile message to neighbor nodes.
+    pub static ref FIND_FILE_NEIGHBORS_TIMEOUT: chrono::Duration = chrono::Duration::seconds(30);
+    /// Timeout to publish FindFile message in the whole network.
     pub static ref FIND_FILE_TIMEOUT: chrono::Duration = chrono::Duration::minutes(5);
     pub static ref ANNOUNCE_FILE_TIMEOUT: chrono::Duration = chrono::Duration::minutes(5);
     pub static ref ANNOUNCE_SHARD_CONFIG_TIMEOUT: chrono::Duration = chrono::Duration::minutes(5);
@@ -219,6 +225,25 @@ impl Libp2pEventHandler {
                 });
                 metrics::LIBP2P_HANDLE_GET_CHUNKS_REQUEST.mark(1);
             }
+            Request::AnnounceFile(announcement) => {
+                match ShardConfig::new(announcement.shard_id, announcement.num_shard) {
+                    Ok(v) => {
+                        self.file_location_cache.insert_peer_config(peer_id, v);
+
+                        self.send_to_sync(SyncMessage::AnnounceFile {
+                            peer_id,
+                            request_id,
+                            announcement,
+                        });
+                    }
+                    Err(_) => self.send_to_network(NetworkMessage::ReportPeer {
+                        peer_id,
+                        action: PeerAction::Fatal,
+                        source: ReportSource::RPC,
+                        msg: "Invalid shard config in AnnounceFile RPC message",
+                    }),
+                }
+            }
             Request::DataByHash(_) => {
                 // ignore
             }
@@ -316,9 +341,13 @@ impl Libp2pEventHandler {
 
         match message {
             PubsubMessage::ExampleMessage(_) => MessageAcceptance::Ignore,
+            PubsubMessage::NewFile(msg) => {
+                metrics::LIBP2P_HANDLE_PUBSUB_NEW_FILE.mark(1);
+                self.on_new_file(propagation_source, msg).await
+            }
             PubsubMessage::FindFile(msg) => {
                 metrics::LIBP2P_HANDLE_PUBSUB_FIND_FILE.mark(1);
-                self.on_find_file(msg).await
+                self.on_find_file(propagation_source, msg).await
             }
             PubsubMessage::FindChunks(msg) => {
                 metrics::LIBP2P_HANDLE_PUBSUB_FIND_CHUNKS.mark(1);
@@ -348,6 +377,63 @@ impl Libp2pEventHandler {
         }
     }
 
+    /// Handle NewFile pubsub message `msg` that published by `from` peer.
+    async fn on_new_file(&self, from: PeerId, msg: NewFile) -> MessageAcceptance {
+        // verify timestamp
+        let d = duration_since(
+            msg.timestamp,
+            metrics::LIBP2P_HANDLE_PUBSUB_NEW_FILE_LATENCY.clone(),
+        );
+        if d < TOLERABLE_DRIFT.neg() || d > *NEW_FILE_TIMEOUT {
+            debug!(?d, ?msg, "Invalid timestamp, ignoring NewFile message");
+            metrics::LIBP2P_HANDLE_PUBSUB_NEW_FILE_TIMEOUT.mark(1);
+            self.send_to_network(NetworkMessage::ReportPeer {
+                peer_id: from,
+                action: PeerAction::LowToleranceError,
+                source: ReportSource::Gossipsub,
+                msg: "Received out of date NewFile message",
+            });
+            return MessageAcceptance::Ignore;
+        }
+
+        // verify announced shard config
+        let announced_shard_config = match ShardConfig::new(msg.shard_id, msg.num_shard) {
+            Ok(v) => v,
+            Err(_) => return MessageAcceptance::Reject,
+        };
+
+        // ignore if shard config mismatch
+        let my_shard_config = self.store.get_store().get_shard_config();
+        if !my_shard_config.intersect(&announced_shard_config) {
+            return MessageAcceptance::Ignore;
+        }
+
+        // ignore if already exists
+        match self.store.check_tx_completed(msg.tx_id.seq).await {
+            Ok(true) => return MessageAcceptance::Ignore,
+            Ok(false) => {}
+            Err(err) => {
+                warn!(?err, tx_seq = %msg.tx_id.seq, "Failed to check tx completed");
+                return MessageAcceptance::Ignore;
+            }
+        }
+
+        // ignore if already pruned
+        match self.store.check_tx_pruned(msg.tx_id.seq).await {
+            Ok(true) => return MessageAcceptance::Ignore,
+            Ok(false) => {}
+            Err(err) => {
+                warn!(?err, tx_seq = %msg.tx_id.seq, "Failed to check tx pruned");
+                return MessageAcceptance::Ignore;
+            }
+        }
+
+        // notify sync layer to handle in advance
+        self.send_to_sync(SyncMessage::NewFile { from, msg });
+
+        MessageAcceptance::Ignore
+    }
+
     async fn construct_announced_ip(&self) -> Option<Multiaddr> {
         // public address configured
         if let Some(ip) = self.config.public_address {
@@ -485,27 +571,69 @@ impl Libp2pEventHandler {
         Some(PubsubMessage::AnnounceShardConfig(signed))
     }
 
-    async fn on_find_file(&self, msg: FindFile) -> MessageAcceptance {
+    async fn on_find_file(&self, from: PeerId, msg: FindFile) -> MessageAcceptance {
-        let FindFile { tx_id, timestamp } = msg;
+        let FindFile {
+            tx_id, timestamp, ..
+        } = msg;
 
         // verify timestamp
         let d = duration_since(
             timestamp,
             metrics::LIBP2P_HANDLE_PUBSUB_FIND_FILE_LATENCY.clone(),
         );
-        if d < TOLERABLE_DRIFT.neg() || d > *FIND_FILE_TIMEOUT {
+        let timeout = if msg.neighbors_only {
+            *FIND_FILE_NEIGHBORS_TIMEOUT
+        } else {
+            *FIND_FILE_TIMEOUT
+        };
+        if d < TOLERABLE_DRIFT.neg() || d > timeout {
             debug!(%timestamp, ?d, "Invalid timestamp, ignoring FindFile message");
             metrics::LIBP2P_HANDLE_PUBSUB_FIND_FILE_TIMEOUT.mark(1);
+            if msg.neighbors_only {
+                self.send_to_network(NetworkMessage::ReportPeer {
+                    peer_id: from,
+                    action: PeerAction::LowToleranceError,
+                    source: ReportSource::Gossipsub,
+                    msg: "Received out of date FindFile message",
+                });
+            }
             return MessageAcceptance::Ignore;
         }
 
+        // verify announced shard config
+        let announced_shard_config = match ShardConfig::new(msg.shard_id, msg.num_shard) {
+            Ok(v) => v,
+            Err(_) => return MessageAcceptance::Reject,
+        };
+
+        // handle on shard config mismatch
+        let my_shard_config = self.store.get_store().get_shard_config();
+        if !my_shard_config.intersect(&announced_shard_config) {
+            return if msg.neighbors_only {
+                MessageAcceptance::Ignore
+            } else {
+                MessageAcceptance::Accept
+            };
+        }
+
         // check if we have it
         if matches!(self.store.check_tx_completed(tx_id.seq).await, Ok(true)) {
             if let Ok(Some(tx)) = self.store.get_tx_by_seq_number(tx_id.seq).await {
                 if tx.id() == tx_id {
                     trace!(?tx_id, "Found file locally, responding to FindFile query");
 
-                    if self.publish_file(tx_id).await.is_some() {
+                    if msg.neighbors_only {
+                        // announce file via RPC to avoid flooding pubsub message
+                        self.send_to_network(NetworkMessage::SendRequest {
+                            peer_id: from,
+                            request: Request::AnnounceFile(FileAnnouncement {
+                                tx_id,
+                                num_shard: my_shard_config.num_shard,
+                                shard_id: my_shard_config.shard_id,
+                            }),
+                            request_id: RequestId::Router(Instant::now()),
+                        });
+                    } else if self.publish_file(tx_id).await.is_some() {
                         metrics::LIBP2P_HANDLE_PUBSUB_FIND_FILE_STORE.mark(1);
                         return MessageAcceptance::Ignore;
                     }
@@ -513,6 +641,11 @@ impl Libp2pEventHandler {
             }
         }
 
+        // do not forward to whole network if only find file from neighbor nodes
+        if msg.neighbors_only {
+            return MessageAcceptance::Ignore;
+        }
+
         // try from cache
         if let Some(mut msg) = self.file_location_cache.get_one(tx_id) {
             trace!(?tx_id, "Found file in cache, responding to FindFile query");
@@ -834,7 +967,7 @@ impl Libp2pEventHandler {
         }
     }
 
-    pub async fn publish_file(&self, tx_id: TxID) -> Option<bool> {
+    async fn publish_file(&self, tx_id: TxID) -> Option<bool> {
         match self.file_batcher.write().await.add(tx_id) {
             Some(batch) => {
                 let announcement = self.construct_announce_file_message(batch).await?;
@@ -1203,7 +1336,13 @@ mod tests {
     ) -> MessageAcceptance {
         let (alice, bob) = (PeerId::random(), PeerId::random());
         let id = MessageId::new(b"dummy message");
-        let message = PubsubMessage::FindFile(FindFile { tx_id, timestamp });
+        let message = PubsubMessage::FindFile(FindFile {
+            tx_id,
+            num_shard: 1,
+            shard_id: 0,
+            neighbors_only: false,
+            timestamp,
+        });
         handler.on_pubsub_message(alice, bob, &id, message).await
     }
 

node/router/src/metrics.rs

@@ -44,6 +44,11 @@ lazy_static::lazy_static! {
     pub static ref LIBP2P_HANDLE_RESPONSE_ERROR: Arc<dyn Meter> = register_meter_with_group("router_libp2p_handle_response_error", "qps");
     pub static ref LIBP2P_HANDLE_RESPONSE_ERROR_LATENCY: Arc<dyn Histogram> = Sample::ExpDecay(0.015).register_with_group("router_libp2p_handle_response_error", "latency", 1024);
 
+    // libp2p_event_handler: new file
+    pub static ref LIBP2P_HANDLE_PUBSUB_NEW_FILE: Arc<dyn Meter> = register_meter_with_group("router_libp2p_handle_pubsub_new_file", "qps");
+    pub static ref LIBP2P_HANDLE_PUBSUB_NEW_FILE_LATENCY: Arc<dyn Histogram> = Sample::ExpDecay(0.015).register_with_group("router_libp2p_handle_pubsub_new_file", "latency", 1024);
+    pub static ref LIBP2P_HANDLE_PUBSUB_NEW_FILE_TIMEOUT: Arc<dyn Meter> = register_meter_with_group("router_libp2p_handle_pubsub_new_file", "timeout");
+
     // libp2p_event_handler: find & announce file
     pub static ref LIBP2P_HANDLE_PUBSUB_FIND_FILE: Arc<dyn Meter> = register_meter_with_group("router_libp2p_handle_pubsub_find_file", "qps");
     pub static ref LIBP2P_HANDLE_PUBSUB_FIND_FILE_LATENCY: Arc<dyn Histogram> = Sample::ExpDecay(0.015).register_with_group("router_libp2p_handle_pubsub_find_file", "latency", 1024);

node/router/src/service.rs

@@ -5,11 +5,14 @@ use chunk_pool::ChunkPoolMessage;
 use file_location_cache::FileLocationCache;
 use futures::{channel::mpsc::Sender, prelude::*};
 use miner::MinerMessage;
+use network::types::NewFile;
+use network::PubsubMessage;
 use network::{
     BehaviourEvent, Keypair, Libp2pEvent, NetworkGlobals, NetworkMessage, RequestId,
     Service as LibP2PService, Swarm,
 };
 use pruner::PrunerMessage;
+use shared_types::timestamp_now;
 use std::sync::Arc;
 use storage::log_store::Store as LogStore;
 use storage_async::Store;
@@ -44,6 +47,8 @@ pub struct RouterService {
     /// Stores potentially created UPnP mappings to be removed on shutdown. (TCP port and UDP
     /// port).
     upnp_mappings: (Option<u16>, Option<u16>),
+
+    store: Arc<dyn LogStore>,
 }
 
 impl RouterService {
@@ -63,7 +68,6 @@ impl RouterService {
         local_keypair: Keypair,
         config: Config,
     ) {
-        let store = Store::new(store, executor.clone());
         let peers = Arc::new(RwLock::new(PeerManager::new(config.clone())));
 
         // create the network service and spawn the task
@@ -81,11 +85,12 @@ impl RouterService {
                 sync_send,
                 chunk_pool_send,
                 local_keypair,
-                store,
+                Store::new(store.clone(), executor.clone()),
                 file_location_cache,
                 peers,
             ),
             upnp_mappings: (None, None),
+            store,
         };
 
         // spawn service
@@ -328,14 +333,15 @@ impl RouterService {
                 }
             }
             NetworkMessage::AnnounceLocalFile { tx_id } => {
-                if self
+                let shard_config = self.store.get_shard_config();
-                    .libp2p_event_handler
+                let msg = PubsubMessage::NewFile(NewFile {
-                    .publish_file(tx_id)
+                    tx_id,
-                    .await
+                    num_shard: shard_config.num_shard,
-                    .is_some()
+                    shard_id: shard_config.shard_id,
-                {
+                    timestamp: timestamp_now(),
-                    metrics::SERVICE_ROUTE_NETWORK_MESSAGE_ANNOUNCE_LOCAL_FILE.mark(1);
+                });
-                }
+                self.libp2p.swarm.behaviour_mut().publish(vec![msg]);
+                metrics::SERVICE_ROUTE_NETWORK_MESSAGE_ANNOUNCE_LOCAL_FILE.mark(1);
             }
             NetworkMessage::UPnPMappingEstablished {
                 tcp_socket,

node/src/client/builder.rs

@@ -112,7 +112,7 @@ impl ClientBuilder {
     pub fn with_rocksdb_store(mut self, config: &StorageConfig) -> Result<Self, String> {
         let executor = require!("sync", self, runtime_context).clone().executor;
         let store = Arc::new(
-            LogManager::rocksdb(LogConfig::default(), &config.db_dir, executor)
+            LogManager::rocksdb(config.log_config.clone(), &config.db_dir, executor)
                 .map_err(|e| format!("Unable to start RocksDB store: {:?}", e))?,
         );
 

node/src/config/convert.rs

@@ -11,6 +11,7 @@ use shared_types::{NetworkIdentity, ProtocolVersion};
 use std::net::IpAddr;
 use std::time::Duration;
 use storage::config::ShardConfig;
+use storage::log_store::log_manager::LogConfig;
 use storage::StorageConfig;
 
 impl ZgsConfig {
@@ -101,8 +102,11 @@ impl ZgsConfig {
     }
 
     pub fn storage_config(&self) -> Result<StorageConfig, String> {
+        let mut log_config = LogConfig::default();
+        log_config.flow.merkle_node_cache_capacity = self.merkle_node_cache_capacity;
         Ok(StorageConfig {
             db_dir: self.db_dir.clone().into(),
+            log_config,
         })
     }
 

node/src/config/mod.rs

@@ -60,6 +60,7 @@ build_config! {
     (prune_check_time_s, (u64), 60)
     (prune_batch_size, (usize), 16 * 1024)
     (prune_batch_wait_time_ms, (u64), 1000)
+    (merkle_node_cache_capacity, (usize), 32 * 1024 * 1024)
 
     // misc
     (log_config_file, (String), "log_config".to_string())

node/storage/Cargo.toml

@@ -31,6 +31,9 @@ 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]
 rand = "0.8.5"

node/storage/src/config.rs

@@ -1,3 +1,4 @@
+use crate::log_store::log_manager::LogConfig;
 use serde::{Deserialize, Serialize};
 use ssz_derive::{Decode, Encode};
 use std::{cell::RefCell, path::PathBuf, rc::Rc, str::FromStr};
@@ -7,6 +8,7 @@ pub const SHARD_CONFIG_KEY: &str = "shard_config";
 #[derive(Clone)]
 pub struct Config {
     pub db_dir: PathBuf,
+    pub log_config: LogConfig,
 }
 
 #[derive(Clone, Copy, Debug, Decode, Encode, Serialize, Deserialize, Eq, PartialEq)]

node/storage/src/log_store/flow_store.rs

@@ -1,17 +1,20 @@
-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, data_to_merkle_leaves, COL_ENTRY_BATCH, COL_ENTRY_BATCH_ROOT,
     COL_FLOW_MPT_NODES, ENTRY_SIZE, 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};
-use append_merkle::{MerkleTreeInitialData, MerkleTreeRead};
+use append_merkle::{MerkleTreeInitialData, MerkleTreeRead, NodeDatabase, NodeTransaction};
 use itertools::Itertools;
+use kvdb::DBTransaction;
 use parking_lot::RwLock;
 use shared_types::{ChunkArray, DataRoot, FlowProof, Merkle};
 use ssz::{Decode, Encode};
@@ -20,27 +23,32 @@ use std::cmp::Ordering;
 use std::collections::BTreeMap;
 use std::fmt::Debug;
 use std::sync::Arc;
-use std::{cmp, mem};
+use std::time::Instant;
+use std::{any, cmp, mem};
 use tracing::{debug, error, trace};
 use zgs_spec::{BYTES_PER_SECTOR, SEALS_PER_LOAD, SECTORS_PER_LOAD, SECTORS_PER_SEAL};
 
 pub struct FlowStore {
-    db: FlowDBStore,
+    db: Arc<FlowDBStore>,
     seal_manager: SealTaskManager,
     config: FlowConfig,
 }
 
 impl FlowStore {
-    pub fn new(db: Arc<dyn ZgsKeyValueDB>, config: FlowConfig) -> Self {
+    pub fn new(db: Arc<FlowDBStore>, config: FlowConfig) -> Self {
         Self {
-            db: FlowDBStore::new(db),
+            db,
             seal_manager: Default::default(),
             config,
         }
     }
 
     pub fn put_batch_root_list(&self, root_map: BTreeMap<usize, (DataRoot, usize)>) -> Result<()> {
-        self.db.put_batch_root_list(root_map)
+        let start_time = Instant::now();
+        let res = self.db.put_batch_root_list(root_map);
+
+        metrics::PUT_BATCH_ROOT_LIST.update_since(start_time);
+        res
     }
 
     pub fn insert_subtree_list_for_batch(
@@ -48,6 +56,7 @@ impl FlowStore {
         batch_index: usize,
         subtree_list: Vec<(usize, usize, DataRoot)>,
     ) -> Result<()> {
+        let start_time = Instant::now();
         let mut batch = self
             .db
             .get_entry_batch(batch_index as u64)?
@@ -55,6 +64,8 @@ impl FlowStore {
         batch.set_subtree_list(subtree_list);
         self.db.put_entry_raw(vec![(batch_index as u64, batch)])?;
 
+        metrics::INSERT_SUBTREE_LIST.update_since(start_time);
+
         Ok(())
     }
 
@@ -73,7 +84,10 @@ impl FlowStore {
     }
 
     pub fn put_mpt_node_list(&self, node_list: Vec<(usize, usize, DataRoot)>) -> Result<()> {
-        self.db.put_mpt_node_list(node_list)
+        let start_time = Instant::now();
+        let res = self.db.put_mpt_node_list(node_list);
+        metrics::PUT_MPT_NODE.update_since(start_time);
+        res
     }
 
     pub fn delete_batch_list(&self, batch_list: &[u64]) -> Result<()> {
@@ -93,6 +107,7 @@ impl FlowStore {
 #[derive(Clone, Debug)]
 pub struct FlowConfig {
     pub batch_size: usize,
+    pub merkle_node_cache_capacity: usize,
     pub shard_config: Arc<RwLock<ShardConfig>>,
 }
 
@@ -100,6 +115,8 @@ impl Default for FlowConfig {
     fn default() -> Self {
         Self {
             batch_size: SECTORS_PER_LOAD,
+            // Each node takes (8+8+32=)48 Bytes, so the default value is 1.5 GB memory size.
+            merkle_node_cache_capacity: 32 * 1024 * 1024,
             shard_config: Default::default(),
         }
     }
@@ -222,6 +239,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 {
@@ -264,6 +282,8 @@ impl FlowWrite for FlowStore {
 
             batch_list.push((chunk_index, batch));
         }
+
+        metrics::APPEND_ENTRIES.update_since(start_time);
         self.db.put_entry_batch_list(batch_list)
     }
 
@@ -373,6 +393,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 {
@@ -393,6 +414,7 @@ impl FlowDBStore {
             }
         }
         self.kvdb.write(tx)?;
+        metrics::PUT_ENTRY_BATCH_LIST.update_since(start_time);
         Ok(completed_batches)
     }
 
@@ -436,7 +458,7 @@ impl FlowDBStore {
         let mut expected_index = 0;
 
         let empty_data = vec![0; PORA_CHUNK_SIZE * ENTRY_SIZE];
-        let empty_root = *Merkle::new(data_to_merkle_leaves(&empty_data)?, 0, None).root();
+        let empty_root = Merkle::new(data_to_merkle_leaves(&empty_data)?, 0, None).root();
 
         for r in self.kvdb.iter(COL_ENTRY_BATCH_ROOT) {
             let (index_bytes, root_bytes) = r?;
@@ -666,3 +688,84 @@ fn decode_mpt_node_key(data: &[u8]) -> Result<(usize, usize)> {
     let position = try_decode_usize(&data[mem::size_of::<u64>()..])?;
     Ok((layer_index, position))
 }
+
+fn layer_size_key(layer: usize) -> Vec<u8> {
+    let mut key = "layer_size".as_bytes().to_vec();
+    key.extend_from_slice(&layer.to_be_bytes());
+    key
+}
+
+pub struct NodeDBTransaction(DBTransaction);
+
+impl NodeDatabase<DataRoot> for FlowDBStore {
+    fn get_node(&self, layer: usize, pos: usize) -> Result<Option<DataRoot>> {
+        Ok(self
+            .kvdb
+            .get(COL_FLOW_MPT_NODES, &encode_mpt_node_key(layer, pos))?
+            .map(|v| DataRoot::from_slice(&v)))
+    }
+
+    fn get_layer_size(&self, layer: usize) -> Result<Option<usize>> {
+        match self.kvdb.get(COL_FLOW_MPT_NODES, &layer_size_key(layer))? {
+            Some(v) => Ok(Some(try_decode_usize(&v)?)),
+            None => Ok(None),
+        }
+    }
+
+    fn start_transaction(&self) -> Box<dyn NodeTransaction<DataRoot>> {
+        Box::new(NodeDBTransaction(self.kvdb.transaction()))
+    }
+
+    fn commit(&self, tx: Box<dyn NodeTransaction<DataRoot>>) -> Result<()> {
+        let db_tx: Box<NodeDBTransaction> = tx
+            .into_any()
+            .downcast()
+            .map_err(|e| anyhow!("downcast failed, e={:?}", e))?;
+        self.kvdb.write(db_tx.0).map_err(Into::into)
+    }
+}
+
+impl NodeTransaction<DataRoot> for NodeDBTransaction {
+    fn save_node(&mut self, layer: usize, pos: usize, node: &DataRoot) {
+        self.0.put(
+            COL_FLOW_MPT_NODES,
+            &encode_mpt_node_key(layer, pos),
+            node.as_bytes(),
+        );
+    }
+
+    fn save_node_list(&mut self, nodes: &[(usize, usize, &DataRoot)]) {
+        for (layer_index, position, data) in nodes {
+            self.0.put(
+                COL_FLOW_MPT_NODES,
+                &encode_mpt_node_key(*layer_index, *position),
+                data.as_bytes(),
+            );
+        }
+    }
+
+    fn remove_node_list(&mut self, nodes: &[(usize, usize)]) {
+        for (layer_index, position) in nodes {
+            self.0.delete(
+                COL_FLOW_MPT_NODES,
+                &encode_mpt_node_key(*layer_index, *position),
+            );
+        }
+    }
+
+    fn save_layer_size(&mut self, layer: usize, size: usize) {
+        self.0.put(
+            COL_FLOW_MPT_NODES,
+            &layer_size_key(layer),
+            &size.to_be_bytes(),
+        );
+    }
+
+    fn remove_layer_size(&mut self, layer: usize) {
+        self.0.delete(COL_FLOW_MPT_NODES, &layer_size_key(layer));
+    }
+
+    fn into_any(self: Box<Self>) -> Box<dyn Any> {
+        self
+    }
+}

node/storage/src/log_store/load_chunk/mod.rs

@@ -4,11 +4,10 @@ mod seal;
 mod serde;
 
 use ::serde::{Deserialize, Serialize};
-use std::cmp::min;
-
 use anyhow::Result;
 use ethereum_types::H256;
 use ssz_derive::{Decode, Encode};
+use std::cmp::min;
 
 use crate::log_store::log_manager::data_to_merkle_leaves;
 use crate::try_option;
@@ -206,7 +205,7 @@ impl EntryBatch {
             }
         }
         Ok(Some(
-            *try_option!(self.to_merkle_tree(is_first_chunk)?).root(),
+            try_option!(self.to_merkle_tree(is_first_chunk)?).root(),
         ))
     }
 

node/storage/src/log_store/log_manager.rs

@@ -1,5 +1,5 @@
 use crate::config::ShardConfig;
-use crate::log_store::flow_store::{batch_iter_sharded, FlowConfig, FlowStore};
+use crate::log_store::flow_store::{batch_iter_sharded, FlowConfig, FlowDBStore, FlowStore};
 use crate::log_store::tx_store::TransactionStore;
 use crate::log_store::{
     FlowRead, FlowWrite, LogStoreChunkRead, LogStoreChunkWrite, LogStoreRead, LogStoreWrite,
@@ -11,6 +11,7 @@ use ethereum_types::H256;
 use kvdb_rocksdb::{Database, DatabaseConfig};
 use merkle_light::merkle::{log2_pow2, MerkleTree};
 use merkle_tree::RawLeafSha3Algorithm;
+use once_cell::sync::Lazy;
 use parking_lot::RwLock;
 use rayon::iter::ParallelIterator;
 use rayon::prelude::ParallelSlice;
@@ -23,10 +24,12 @@ use std::collections::BTreeMap;
 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 super::tx_store::BlockHashAndSubmissionIndex;
+use crate::log_store::metrics;
-use super::{FlowSeal, MineLoadChunk, SealAnswer, SealTask};
+use crate::log_store::tx_store::BlockHashAndSubmissionIndex;
+use crate::log_store::{FlowSeal, MineLoadChunk, SealAnswer, SealTask};
 
 /// 256 Bytes
 pub const ENTRY_SIZE: usize = 256;
@@ -47,6 +50,14 @@ pub const COL_NUM: u32 = 9;
 // Process at most 1M entries (256MB) pad data at a time.
 const PAD_MAX_SIZE: usize = 1 << 20;
 
+static PAD_SEGMENT_ROOT: Lazy<H256> = Lazy::new(|| {
+    Merkle::new(
+        data_to_merkle_leaves(&[0; ENTRY_SIZE * PORA_CHUNK_SIZE]).unwrap(),
+        0,
+        None,
+    )
+    .root()
+});
 pub struct UpdateFlowMessage {
     pub root_map: BTreeMap<usize, (H256, usize)>,
     pub pad_data: usize,
@@ -94,6 +105,7 @@ impl MerkleManager {
     }
 
     fn revert_merkle_tree(&mut self, tx_seq: u64, tx_store: &TransactionStore) -> Result<()> {
+        debug!("revert merkle tree {}", tx_seq);
         // Special case for reverting tx_seq == 0
         if tx_seq == u64::MAX {
             self.pora_chunks_merkle.reset();
@@ -116,7 +128,7 @@ impl MerkleManager {
         if self.pora_chunks_merkle.leaves() == 0 && self.last_chunk_merkle.leaves() == 0 {
             self.last_chunk_merkle.append(H256::zero());
             self.pora_chunks_merkle
-                .update_last(*self.last_chunk_merkle.root());
+                .update_last(self.last_chunk_merkle.root());
         } else if self.last_chunk_merkle.leaves() != 0 {
             let last_chunk_start_index = self.last_chunk_start_index();
             let last_chunk_data = flow_store.get_available_entries(
@@ -355,7 +367,7 @@ impl LogStoreWrite for LogManager {
         merkle.revert_merkle_tree(tx_seq, &self.tx_store)?;
         merkle.try_initialize(&self.flow_store)?;
         assert_eq!(
-            Some(*merkle.last_chunk_merkle.root()),
+            Some(merkle.last_chunk_merkle.root()),
             merkle
                 .pora_chunks_merkle
                 .leaf_at(merkle.pora_chunks_merkle.leaves() - 1)?
@@ -577,7 +589,7 @@ impl LogStoreRead for LogManager {
     fn get_context(&self) -> crate::error::Result<(DataRoot, u64)> {
         let merkle = self.merkle.read_recursive();
         Ok((
-            *merkle.pora_chunks_merkle.root(),
+            merkle.pora_chunks_merkle.root(),
             merkle.last_chunk_start_index() + merkle.last_chunk_merkle.leaves() as u64,
         ))
     }
@@ -626,13 +638,10 @@ impl LogManager {
         executor: task_executor::TaskExecutor,
     ) -> Result<Self> {
         let tx_store = TransactionStore::new(db.clone())?;
-        let flow_store = Arc::new(FlowStore::new(db.clone(), config.flow));
+        let flow_db = Arc::new(FlowDBStore::new(db.clone()));
-        let mut initial_data = flow_store.get_chunk_root_list()?;
+        let flow_store = Arc::new(FlowStore::new(flow_db.clone(), config.flow.clone()));
-        // If the last tx `put_tx` does not complete, we will revert it in `initial_data.subtree_list`
+        // If the last tx `put_tx` does not complete, we will revert it in `pora_chunks_merkle`
-        // first and call `put_tx` later. The known leaves in its data will be saved in `extra_leaves`
+        // first and call `put_tx` later.
-        // and inserted later.
-        let mut extra_leaves = Vec::new();
-
         let next_tx_seq = tx_store.next_tx_seq();
         let mut start_tx_seq = if next_tx_seq > 0 {
             Some(next_tx_seq - 1)
@@ -640,15 +649,25 @@ impl LogManager {
             None
         };
         let mut last_tx_to_insert = None;
+
+        let mut pora_chunks_merkle = Merkle::new_with_subtrees(
+            flow_db,
+            config.flow.merkle_node_cache_capacity,
+            log2_pow2(PORA_CHUNK_SIZE),
+        )?;
         if let Some(last_tx_seq) = start_tx_seq {
             if !tx_store.check_tx_completed(last_tx_seq)? {
                 // Last tx not finalized, we need to check if its `put_tx` is completed.
                 let last_tx = tx_store
                     .get_tx_by_seq_number(last_tx_seq)?
                     .expect("tx missing");
-                let mut current_len = initial_data.leaves();
+                let current_len = pora_chunks_merkle.leaves();
-                let expected_len =
+                let expected_len = sector_to_segment(
-                    sector_to_segment(last_tx.start_entry_index + last_tx.num_entries() as u64);
+                    last_tx.start_entry_index
+                        + last_tx.num_entries() as u64
+                        + PORA_CHUNK_SIZE as u64
+                        - 1,
+                );
                 match expected_len.cmp(&(current_len)) {
                     Ordering::Less => {
                         bail!(
@@ -676,43 +695,33 @@ impl LogManager {
                                 previous_tx.start_entry_index + previous_tx.num_entries() as u64,
                             );
                             if current_len > expected_len {
-                                while let Some((subtree_depth, _)) = initial_data.subtree_list.pop()
+                                pora_chunks_merkle.revert_to_leaves(expected_len)?;
-                                {
-                                    current_len -= 1 << (subtree_depth - 1);
-                                    if current_len == expected_len {
-                                        break;
-                                    }
-                                }
                             } else {
-                                warn!(
+                                assert_eq!(current_len, expected_len);
-                                    "revert last tx with no-op: {} {}",
-                                    current_len, expected_len
-                                );
                             }
-                            assert_eq!(current_len, expected_len);
+                            start_tx_seq = Some(previous_tx.seq);
-                            while let Some((index, h)) = initial_data.known_leaves.pop() {
-                                if index < current_len {
-                                    initial_data.known_leaves.push((index, h));
-                                    break;
-                                } else {
-                                    extra_leaves.push((index, h));
-                                }
-                            }
-                            start_tx_seq = Some(last_tx_seq - 1);
                         };
                     }
                 }
             }
         }
 
-        let mut pora_chunks_merkle =
-            Merkle::new_with_subtrees(initial_data, log2_pow2(PORA_CHUNK_SIZE), start_tx_seq)?;
         let last_chunk_merkle = match start_tx_seq {
             Some(tx_seq) => {
-                tx_store.rebuild_last_chunk_merkle(pora_chunks_merkle.leaves(), tx_seq)?
+                let tx = tx_store.get_tx_by_seq_number(tx_seq)?.expect("tx missing");
+                if (tx.start_entry_index() + tx.num_entries() as u64) % PORA_CHUNK_SIZE as u64 == 0
+                {
+                    // The last chunk should be aligned, so it's empty.
+                    Merkle::new_with_depth(vec![], log2_pow2(PORA_CHUNK_SIZE) + 1, None)
+                } else {
+                    tx_store.rebuild_last_chunk_merkle(pora_chunks_merkle.leaves() - 1, tx_seq)?
+                }
             }
             // Initialize
-            None => Merkle::new_with_depth(vec![], 1, None),
+            None => {
+                pora_chunks_merkle.reset();
+                Merkle::new_with_depth(vec![], 1, None)
+            }
         };
 
         debug!(
@@ -722,10 +731,10 @@ impl LogManager {
             last_chunk_merkle.leaves(),
         );
         if last_chunk_merkle.leaves() != 0 {
-            pora_chunks_merkle.append(*last_chunk_merkle.root());
+            pora_chunks_merkle.update_last(last_chunk_merkle.root());
-            // update the merkle root
-            pora_chunks_merkle.commit(start_tx_seq);
         }
+        // update the merkle root
+        pora_chunks_merkle.commit(start_tx_seq);
         let merkle = RwLock::new(MerkleManager {
             pora_chunks_merkle,
             last_chunk_merkle,
@@ -744,18 +753,7 @@ impl LogManager {
         log_manager.start_receiver(receiver, executor);
 
         if let Some(tx) = last_tx_to_insert {
-            log_manager.revert_to(tx.seq - 1)?;
             log_manager.put_tx(tx)?;
-            let mut merkle = log_manager.merkle.write();
-            for (index, h) in extra_leaves {
-                if index < merkle.pora_chunks_merkle.leaves() {
-                    merkle.pora_chunks_merkle.fill_leaf(index, h);
-                } else {
-                    error!("out of range extra leaf: index={} hash={:?}", index, h);
-                }
-            }
-        } else {
-            assert!(extra_leaves.is_empty());
         }
         log_manager
             .merkle
@@ -880,6 +878,7 @@ impl LogManager {
         if merkle_list.is_empty() {
             return Ok(());
         }
+        let start_time = Instant::now();
 
         self.pad_tx(tx_start_index, &mut *merkle)?;
 
@@ -894,16 +893,16 @@ impl LogManager {
                     // `last_chunk_merkle` was empty, so this is a new leaf in the top_tree.
                     merkle
                         .pora_chunks_merkle
-                        .append_subtree(1, *merkle.last_chunk_merkle.root())?;
+                        .append_subtree(1, merkle.last_chunk_merkle.root())?;
                 } else {
                     merkle
                         .pora_chunks_merkle
-                        .update_last(*merkle.last_chunk_merkle.root());
+                        .update_last(merkle.last_chunk_merkle.root());
                 }
                 if merkle.last_chunk_merkle.leaves() == PORA_CHUNK_SIZE {
                     batch_root_map.insert(
                         merkle.pora_chunks_merkle.leaves() - 1,
-                        (*merkle.last_chunk_merkle.root(), 1),
+                        (merkle.last_chunk_merkle.root(), 1),
                     );
                     self.complete_last_chunk_merkle(
                         merkle.pora_chunks_merkle.leaves() - 1,
@@ -925,12 +924,15 @@ impl LogManager {
             }
         }
         self.flow_store.put_batch_root_list(batch_root_map)?;
+
+        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;
@@ -959,7 +961,7 @@ impl LogManager {
                         .append_list(data_to_merkle_leaves(&pad_data)?);
                     merkle
                         .pora_chunks_merkle
-                        .update_last(*merkle.last_chunk_merkle.root());
+                        .update_last(merkle.last_chunk_merkle.root());
                 } else {
                     if last_chunk_pad != 0 {
                         is_full_empty = false;
@@ -969,10 +971,10 @@ impl LogManager {
                             .append_list(data_to_merkle_leaves(&pad_data[..last_chunk_pad])?);
                         merkle
                             .pora_chunks_merkle
-                            .update_last(*merkle.last_chunk_merkle.root());
+                            .update_last(merkle.last_chunk_merkle.root());
                         root_map.insert(
                             merkle.pora_chunks_merkle.leaves() - 1,
-                            (*merkle.last_chunk_merkle.root(), 1),
+                            (merkle.last_chunk_merkle.root(), 1),
                         );
                         completed_chunk_index = Some(merkle.pora_chunks_merkle.leaves() - 1);
                     }
@@ -980,12 +982,11 @@ impl LogManager {
                     // Pad with more complete chunks.
                     let mut start_index = last_chunk_pad / ENTRY_SIZE;
                     while pad_data.len() >= (start_index + PORA_CHUNK_SIZE) * ENTRY_SIZE {
-                        let data = pad_data[start_index * ENTRY_SIZE
+                        merkle.pora_chunks_merkle.append(*PAD_SEGMENT_ROOT);
-                            ..(start_index + PORA_CHUNK_SIZE) * ENTRY_SIZE]
+                        root_map.insert(
-                            .to_vec();
+                            merkle.pora_chunks_merkle.leaves() - 1,
-                        let root = *Merkle::new(data_to_merkle_leaves(&data)?, 0, None).root();
+                            (*PAD_SEGMENT_ROOT, 1),
-                        merkle.pora_chunks_merkle.append(root);
+                        );
-                        root_map.insert(merkle.pora_chunks_merkle.leaves() - 1, (root, 1));
                         start_index += PORA_CHUNK_SIZE;
                     }
                     assert_eq!(pad_data.len(), start_index * ENTRY_SIZE);
@@ -1020,6 +1021,8 @@ impl LogManager {
             merkle.pora_chunks_merkle.leaves(),
             merkle.last_chunk_merkle.leaves()
         );
+
+        metrics::PAD_TX.update_since(start_time);
         Ok(())
     }
 
@@ -1061,7 +1064,7 @@ impl LogManager {
             }
             merkle
                 .pora_chunks_merkle
-                .update_last(*merkle.last_chunk_merkle.root());
+                .update_last(merkle.last_chunk_merkle.root());
         }
         let chunk_roots = self.flow_store.append_entries(flow_entry_array)?;
         for (chunk_index, chunk_root) in chunk_roots {
@@ -1148,6 +1151,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.
@@ -1196,6 +1201,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(())
     }
 
@@ -1268,6 +1275,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");
     }
@@ -1283,6 +1291,8 @@ pub fn data_to_merkle_leaves(leaf_data: &[u8]) -> Result<Vec<H256>> {
             .map(Sha3Algorithm::leaf)
             .collect()
     };
+
+    metrics::DATA_TO_MERKLE_LEAVES.update_since(start_time);
     Ok(r)
 }
 

node/storage/src/log_store/metrics.rs

@@ -0,0 +1,33 @@
+use std::sync::Arc;
+
+use metrics::{register_timer, Timer};
+
+lazy_static::lazy_static! {
+    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");
+}

node/storage/src/log_store/mod.rs

@@ -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;

node/storage/src/log_store/tests.rs

@@ -8,6 +8,7 @@ use ethereum_types::H256;
 use rand::random;
 use shared_types::{compute_padded_chunk_size, ChunkArray, Transaction, CHUNK_SIZE};
 use std::cmp;
+
 use task_executor::test_utils::TestRuntime;
 
 #[test]

node/storage/src/log_store/tx_store.rs

@@ -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())?
+        let start_time = Instant::now();
-            == Some(vec![TX_STATUS_FINALIZED]))
+        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> {
@@ -292,6 +301,9 @@ impl TransactionStore {
             match tx.start_entry_index.cmp(&last_chunk_start_index) {
                 cmp::Ordering::Greater => {
                     tx_list.push((tx_seq, tx.merkle_nodes));
+                    if tx.start_entry_index >= last_chunk_start_index + PORA_CHUNK_SIZE as u64 {
+                        break;
+                    }
                 }
                 cmp::Ordering::Equal => {
                     tx_list.push((tx_seq, tx.merkle_nodes));

node/sync/src/auto_sync/batcher_random.rs

@@ -59,14 +59,13 @@ impl RandomBatcher {
     pub async fn start(mut self, catched_up: Arc<AtomicBool>) {
         info!("Start to sync files");
 
-        loop {
+        // wait for log entry sync catched up
-            // disable file sync until catched up
+        while !catched_up.load(Ordering::Relaxed) {
-            if !catched_up.load(Ordering::Relaxed) {
+            trace!("Cannot sync file in catch-up phase");
-                trace!("Cannot sync file in catch-up phase");
+            sleep(self.config.auto_sync_idle_interval).await;
-                sleep(self.config.auto_sync_idle_interval).await;
+        }
-                continue;
-            }
 
+        loop {
             if let Ok(state) = self.get_state().await {
                 metrics::RANDOM_STATE_TXS_SYNCING.update(state.tasks.len() as u64);
                 metrics::RANDOM_STATE_TXS_READY.update(state.ready_txs as u64);

node/sync/src/auto_sync/manager.rs

@@ -9,18 +9,23 @@ use storage_async::Store;
 use task_executor::TaskExecutor;
 use tokio::sync::{
     broadcast,
-    mpsc::{unbounded_channel, UnboundedSender},
+    mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender},
     oneshot,
 };
 
 use crate::{Config, SyncSender};
 
-use super::{batcher_random::RandomBatcher, batcher_serial::SerialBatcher, sync_store::SyncStore};
+use super::{
+    batcher_random::RandomBatcher,
+    batcher_serial::SerialBatcher,
+    sync_store::{Queue, SyncStore},
+};
 
 pub struct AutoSyncManager {
-    pub serial: SerialBatcher,
+    pub serial: Option<SerialBatcher>,
     pub random: RandomBatcher,
     pub file_announcement_send: UnboundedSender<u64>,
+    pub new_file_send: UnboundedSender<u64>,
     pub catched_up: Arc<AtomicBool>,
 }
 
@@ -33,42 +38,80 @@ impl AutoSyncManager {
         log_sync_recv: broadcast::Receiver<LogSyncEvent>,
         catch_up_end_recv: oneshot::Receiver<()>,
     ) -> Result<Self> {
-        let (send, recv) = unbounded_channel();
+        let (file_announcement_send, file_announcement_recv) = unbounded_channel();
-        let sync_store = Arc::new(SyncStore::new(store.clone()));
+        let (new_file_send, new_file_recv) = unbounded_channel();
+        let sync_store = if config.neighbors_only {
+            // use v2 db to avoid reading v1 files that announced from the whole network instead of neighbors
+            Arc::new(SyncStore::new_with_name(
+                store.clone(),
+                "pendingv2",
+                "readyv2",
+            ))
+        } else {
+            Arc::new(SyncStore::new(store.clone()))
+        };
         let catched_up = Arc::new(AtomicBool::new(false));
 
-        // sync in sequence
+        // handle new file
-        let serial =
-            SerialBatcher::new(config, store.clone(), sync_send.clone(), sync_store.clone())
-                .await?;
         executor.spawn(
-            serial
+            Self::handle_new_file(new_file_recv, sync_store.clone()),
-                .clone()
+            "auto_sync_handle_new_file",
-                .start(recv, log_sync_recv, catched_up.clone()),
-            "auto_sync_serial",
         );
 
+        // sync in sequence
+        let serial = if config.neighbors_only {
+            None
+        } else {
+            let serial =
+                SerialBatcher::new(config, store.clone(), sync_send.clone(), sync_store.clone())
+                    .await?;
+            executor.spawn(
+                serial
+                    .clone()
+                    .start(file_announcement_recv, log_sync_recv, catched_up.clone()),
+                "auto_sync_serial",
+            );
+
+            Some(serial)
+        };
+
         // sync randomly
         let random = RandomBatcher::new(config, store, sync_send, sync_store);
         executor.spawn(random.clone().start(catched_up.clone()), "auto_sync_random");
 
         // handle on catched up notification
-        let catched_up_cloned = catched_up.clone();
         executor.spawn(
-            async move {
+            Self::listen_catch_up(catch_up_end_recv, catched_up.clone()),
-                if catch_up_end_recv.await.is_ok() {
-                    info!("log entry catched up");
-                    catched_up_cloned.store(true, Ordering::Relaxed);
-                }
-            },
             "auto_sync_wait_for_catchup",
         );
 
         Ok(Self {
             serial,
             random,
-            file_announcement_send: send,
+            file_announcement_send,
+            new_file_send,
             catched_up,
         })
     }
+
+    async fn handle_new_file(
+        mut new_file_recv: UnboundedReceiver<u64>,
+        sync_store: Arc<SyncStore>,
+    ) {
+        while let Some(tx_seq) = new_file_recv.recv().await {
+            if let Err(err) = sync_store.insert(tx_seq, Queue::Ready).await {
+                warn!(?err, %tx_seq, "Failed to insert new file to ready queue");
+            }
+        }
+    }
+
+    async fn listen_catch_up(
+        catch_up_end_recv: oneshot::Receiver<()>,
+        catched_up: Arc<AtomicBool>,
+    ) {
+        if catch_up_end_recv.await.is_ok() {
+            info!("log entry catched up");
+            catched_up.store(true, Ordering::Relaxed);
+        }
+    }
 }

node/sync/src/auto_sync/sync_store.rs

@@ -42,6 +42,14 @@ impl SyncStore {
         }
     }
 
+    pub fn new_with_name(store: Store, pending: &'static str, ready: &'static str) -> Self {
+        Self {
+            store: Arc::new(RwLock::new(store)),
+            pending_txs: TxStore::new(pending),
+            ready_txs: TxStore::new(ready),
+        }
+    }
+
     /// Returns the number of pending txs and ready txs.
     pub async fn stat(&self) -> Result<(usize, usize)> {
         let async_store = self.store.read().await;

node/sync/src/controllers/serial.rs

@@ -14,12 +14,13 @@ use shared_types::{timestamp_now, ChunkArrayWithProof, TxID, CHUNK_SIZE};
 use ssz::Encode;
 use std::{sync::Arc, time::Instant};
 use storage::log_store::log_manager::{sector_to_segment, segment_to_sector, PORA_CHUNK_SIZE};
-use storage_async::Store;
+use storage_async::{ShardConfig, Store};
 
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub enum FailureReason {
     DBError(String),
     TxReverted(TxID),
+    TimeoutFindFile,
 }
 
 #[derive(Clone, Debug, PartialEq, Eq)]
@@ -159,11 +160,14 @@ impl SerialSyncController {
 
     /// Find more peers to sync chunks. Return whether `FindFile` pubsub message published,
     fn try_find_peers(&mut self) {
-        let (published, num_new_peers) = if self.goal.is_all_chunks() {
+        let (published, num_new_peers) = if !self.goal.is_all_chunks() {
-            self.publish_find_file()
-        } else {
             self.publish_find_chunks();
             (true, 0)
+        } else if self.config.neighbors_only {
+            self.do_publish_find_file();
+            (true, 0)
+        } else {
+            self.publish_find_file()
         };
 
         info!(%self.tx_seq, %published, %num_new_peers, "Finding peers");
@@ -199,14 +203,23 @@ impl SerialSyncController {
             return (false, num_new_peers);
         }
 
-        self.ctx.publish(PubsubMessage::FindFile(FindFile {
+        self.do_publish_find_file();
-            tx_id: self.tx_id,
-            timestamp: timestamp_now(),
-        }));
 
         (true, num_new_peers)
     }
 
+    fn do_publish_find_file(&self) {
+        let shard_config = self.store.get_store().get_shard_config();
+
+        self.ctx.publish(PubsubMessage::FindFile(FindFile {
+            tx_id: self.tx_id,
+            num_shard: shard_config.num_shard,
+            shard_id: shard_config.shard_id,
+            neighbors_only: self.config.neighbors_only,
+            timestamp: timestamp_now(),
+        }));
+    }
+
     fn publish_find_chunks(&self) {
         self.ctx.publish(PubsubMessage::FindChunks(FindChunks {
             tx_id: self.tx_id,
@@ -337,6 +350,14 @@ impl SerialSyncController {
         }
     }
 
+    /// Triggered when any peer (TCP connected) announced file via RPC message.
+    pub fn on_peer_announced(&mut self, peer_id: PeerId, shard_config: ShardConfig) {
+        self.peers
+            .add_new_peer_with_config(peer_id, Multiaddr::empty(), shard_config);
+        self.peers
+            .update_state_force(&peer_id, PeerState::Connected);
+    }
+
     pub fn on_dail_failed(&mut self, peer_id: PeerId, err: &DialError) {
         match err {
             DialError::ConnectionLimit(_) => {
@@ -545,6 +566,9 @@ impl SerialSyncController {
                 info!(%self.tx_seq, "Succeeded to finalize file");
                 self.state = SyncState::Completed;
                 metrics::SERIAL_SYNC_FILE_COMPLETED.update_since(self.since.0);
+                // notify neighbor nodes about new file completed to sync
+                self.ctx
+                    .send(NetworkMessage::AnnounceLocalFile { tx_id: self.tx_id });
             }
             Ok(false) => {
                 warn!(?self.tx_id, %self.tx_seq, "Transaction reverted during finalize_tx");
@@ -637,12 +661,19 @@ impl SerialSyncController {
                     {
                         self.state = SyncState::FoundPeers;
                     } else {
-                        // storage node may not have the specific file when `FindFile`
+                        // FindFile timeout
-                        // gossip message received. In this case, just broadcast the
-                        // `FindFile` message again.
                         if since.elapsed() >= self.config.peer_find_timeout {
-                            debug!(%self.tx_seq, "Finding peer timeout and try to find peers again");
+                            if self.config.neighbors_only {
-                            self.try_find_peers();
+                                self.state = SyncState::Failed {
+                                    reason: FailureReason::TimeoutFindFile,
+                                };
+                            } else {
+                                // storage node may not have the specific file when `FindFile`
+                                // gossip message received. In this case, just broadcast the
+                                // `FindFile` message again.
+                                debug!(%self.tx_seq, "Finding peer timeout and try to find peers again");
+                                self.try_find_peers();
+                            }
                         }
 
                         completed = true;
@@ -1513,6 +1544,10 @@ mod tests {
 
         controller.on_response(peer_id, chunks).await;
         assert_eq!(*controller.get_status(), SyncState::Completed);
+        assert!(matches!(
+            network_recv.try_recv().unwrap(),
+            NetworkMessage::AnnounceLocalFile { .. }
+        ));
         assert!(network_recv.try_recv().is_err());
     }
 

node/sync/src/lib.rs

@@ -21,6 +21,10 @@ use std::{
 #[serde(default)]
 pub struct Config {
     // sync service config
+    /// Indicates whether to sync file from neighbor nodes only.
+    /// This is to avoid flooding file announcements in the whole network,
+    /// which leads to high latency or even timeout to sync files.
+    pub neighbors_only: bool,
     #[serde(deserialize_with = "deserialize_duration")]
     pub heartbeat_interval: Duration,
     pub auto_sync_enabled: bool,
@@ -64,6 +68,7 @@ impl Default for Config {
     fn default() -> Self {
         Self {
             // sync service config
+            neighbors_only: false,
             heartbeat_interval: Duration::from_secs(5),
             auto_sync_enabled: false,
             max_sync_files: 8,

node/sync/src/service.rs

@@ -8,7 +8,8 @@ use anyhow::{anyhow, bail, Result};
 use file_location_cache::FileLocationCache;
 use libp2p::swarm::DialError;
 use log_entry_sync::LogSyncEvent;
-use network::types::{AnnounceChunks, FindFile};
+use network::rpc::methods::FileAnnouncement;
+use network::types::{AnnounceChunks, FindFile, NewFile};
 use network::PubsubMessage;
 use network::{
     rpc::GetChunksRequest, rpc::RPCResponseErrorCode, Multiaddr, NetworkMessage, PeerId,
@@ -70,6 +71,15 @@ pub enum SyncMessage {
     AnnounceChunksGossip {
         msg: AnnounceChunks,
     },
+    NewFile {
+        from: PeerId,
+        msg: NewFile,
+    },
+    AnnounceFile {
+        peer_id: PeerId,
+        request_id: PeerRequestId,
+        announcement: FileAnnouncement,
+    },
 }
 
 #[derive(Debug)]
@@ -265,6 +275,12 @@ impl SyncService {
             SyncMessage::AnnounceShardConfig { .. } => {
                 // FIXME: Check if controllers need to be reset?
             }
+            SyncMessage::NewFile { from, msg } => self.on_new_file_gossip(from, msg).await,
+            SyncMessage::AnnounceFile {
+                peer_id,
+                announcement,
+                ..
+            } => self.on_announce_file(peer_id, announcement).await,
         }
     }
 
@@ -279,7 +295,10 @@ impl SyncService {
                     Some(manager) => SyncServiceState {
                         num_syncing: self.controllers.len(),
                         catched_up: Some(manager.catched_up.load(Ordering::Relaxed)),
-                        auto_sync_serial: Some(manager.serial.get_state().await),
+                        auto_sync_serial: match &manager.serial {
+                            Some(v) => Some(v.get_state().await),
+                            None => None,
+                        },
                         auto_sync_random: manager.random.get_state().await.ok(),
                     },
                     None => SyncServiceState {
@@ -577,8 +596,12 @@ impl SyncService {
             Some(tx) => tx,
             None => bail!("Transaction not found"),
         };
+        let shard_config = self.store.get_store().get_shard_config();
         self.ctx.publish(PubsubMessage::FindFile(FindFile {
             tx_id: tx.id(),
+            num_shard: shard_config.num_shard,
+            shard_id: shard_config.shard_id,
+            neighbors_only: false,
             timestamp: timestamp_now(),
         }));
         Ok(())
@@ -642,7 +665,10 @@ impl SyncService {
                             Some(s) => s,
                             None => {
                                 debug!(%tx.seq, "No more data needed");
-                                self.store.finalize_tx_with_hash(tx.seq, tx.hash()).await?;
+                                if self.store.finalize_tx_with_hash(tx.seq, tx.hash()).await? {
+                                    self.ctx
+                                        .send(NetworkMessage::AnnounceLocalFile { tx_id: tx.id() });
+                                }
                                 return Ok(());
                             }
                         };
@@ -748,6 +774,34 @@ impl SyncService {
         }
     }
 
+    /// Handle on `NewFile` gossip message received.
+    async fn on_new_file_gossip(&mut self, from: PeerId, msg: NewFile) {
+        debug!(%from, ?msg, "Received NewFile gossip");
+
+        if let Some(controller) = self.controllers.get_mut(&msg.tx_id.seq) {
+            // Notify new peer announced if file already in sync
+            if let Ok(shard_config) = ShardConfig::new(msg.shard_id, msg.num_shard) {
+                controller.on_peer_announced(from, shard_config);
+                controller.transition();
+            }
+        } else if let Some(manager) = &self.auto_sync_manager {
+            let _ = manager.new_file_send.send(msg.tx_id.seq);
+        }
+    }
+
+    /// Handle on `AnnounceFile` RPC message received.
+    async fn on_announce_file(&mut self, from: PeerId, announcement: FileAnnouncement) {
+        // Notify new peer announced if file already in sync
+        if let Some(controller) = self.controllers.get_mut(&announcement.tx_id.seq) {
+            if let Ok(shard_config) =
+                ShardConfig::new(announcement.shard_id, announcement.num_shard)
+            {
+                controller.on_peer_announced(from, shard_config);
+                controller.transition();
+            }
+        }
+    }
+
     /// Terminate file sync of `min_tx_seq`.
     /// If `is_reverted` is `true` (means confirmed transactions reverted),
     /// also terminate `tx_seq` greater than `min_tx_seq`
@@ -1504,6 +1558,10 @@ mod tests {
         .await;
 
         wait_for_tx_finalized(runtime.store.clone(), tx_seq).await;
+        assert!(matches!(
+            runtime.network_recv.try_recv().unwrap(),
+            NetworkMessage::AnnounceLocalFile { .. }
+        ));
 
         assert!(!runtime.store.check_tx_completed(0).unwrap());
 
@@ -1528,6 +1586,10 @@ mod tests {
         .await;
 
         wait_for_tx_finalized(runtime.store, tx_seq).await;
+        assert!(matches!(
+            runtime.network_recv.try_recv().unwrap(),
+            NetworkMessage::AnnounceLocalFile { .. }
+        ));
 
         sync_send
             .notify(SyncMessage::PeerDisconnected {

run/config-testnet-standard.toml

@@ -232,6 +232,10 @@ batcher_announcement_capacity = 100
 # all files, and sufficient disk space is required.
 auto_sync_enabled = true
 
+# Indicates whether to sync file from neighbor nodes only. This is to avoid flooding file
+# announcements in the whole network, which leads to high latency or even timeout to sync files.
+neighbors_only = true
+
 #  Maximum number of files in sync from other peers simultaneously.
 # max_sync_files = 8
 

run/config-testnet-turbo.toml

@@ -244,6 +244,10 @@ batcher_announcement_capacity = 100
 # all files, and sufficient disk space is required.
 auto_sync_enabled = true
 
+# Indicates whether to sync file from neighbor nodes only. This is to avoid flooding file
+# announcements in the whole network, which leads to high latency or even timeout to sync files.
+neighbors_only = true
+
 #  Maximum number of files in sync from other peers simultaneously.
 # max_sync_files = 8
 

run/config.toml

@@ -246,6 +246,10 @@
 # all files, and sufficient disk space is required.
 # auto_sync_enabled = false
 
+# Indicates whether to sync file from neighbor nodes only. This is to avoid flooding file
+# announcements in the whole network, which leads to high latency or even timeout to sync files.
+neighbors_only = true
+
 #  Maximum number of files in sync from other peers simultaneously.
 # max_sync_files = 8
 

tests/node_cache_test.py

@@ -0,0 +1,43 @@
+#!/usr/bin/env python3
+
+from test_framework.test_framework import TestFramework
+from utility.submission import create_submission, submit_data
+from utility.utils import wait_until
+
+
+class NodeCacheTest(TestFramework):
+    def setup_params(self):
+        self.zgs_node_configs[0] = {
+            "merkle_node_cache_capacity": 1024,
+        }
+
+    def run_test(self):
+        client = self.nodes[0]
+
+        chunk_data = b"\x02" * 256 * 1024 * 1024 * 3
+        submissions, data_root = create_submission(chunk_data)
+        self.contract.submit(submissions)
+        wait_until(lambda: self.contract.num_submissions() == 1)
+        wait_until(lambda: client.zgs_get_file_info(data_root) is not None)
+
+        segment = submit_data(client, chunk_data)
+        self.log.info("segment: %s", len(segment))
+        wait_until(lambda: client.zgs_get_file_info(data_root)["finalized"])
+
+        self.stop_storage_node(0)
+        self.start_storage_node(0)
+        self.nodes[0].wait_for_rpc_connection()
+
+        chunk_data = b"\x03" * 256 * (1024 * 765 + 5)
+        submissions, data_root = create_submission(chunk_data)
+        self.contract.submit(submissions)
+        wait_until(lambda: self.contract.num_submissions() == 2)
+        wait_until(lambda: client.zgs_get_file_info(data_root) is not None)
+
+        segment = submit_data(client, chunk_data)
+        self.log.info("segment: %s", len(segment))
+        wait_until(lambda: client.zgs_get_file_info(data_root)["finalized"])
+
+
+if __name__ == "__main__":
+    NodeCacheTest().main()

tests/sync_auto_random_v2_test.py

@@ -0,0 +1,36 @@
+#!/usr/bin/env python3
+
+from test_framework.test_framework import TestFramework
+from utility.utils import wait_until
+
+
+class AutoRandomSyncV2Test(TestFramework):
+    def setup_params(self):
+        self.num_nodes = 4
+
+        # Enable random auto sync v2
+        for i in range(self.num_nodes):
+            self.zgs_node_configs[i] = {
+                "sync": {
+                    "auto_sync_enabled": True,
+                    "max_sequential_workers": 0,
+                    "max_random_workers": 3,
+                    "neighbors_only": True,
+                }
+            }
+
+    def run_test(self):
+        # Submit and upload files on node 0
+        data_root_1 = self.__upload_file__(0, 256 * 1024)
+        data_root_2 = self.__upload_file__(0, 256 * 1024)
+
+        # Files should be available on other nodes via auto sync
+        for i in range(1, self.num_nodes):
+            wait_until(lambda: self.nodes[i].zgs_get_file_info(data_root_1) is not None)
+            wait_until(lambda: self.nodes[i].zgs_get_file_info(data_root_1)["finalized"])
+            wait_until(lambda: self.nodes[i].zgs_get_file_info(data_root_2) is not None)
+            wait_until(lambda: self.nodes[i].zgs_get_file_info(data_root_2)["finalized"])
+
+
+if __name__ == "__main__":
+    AutoRandomSyncV2Test().main()

tests/test_framework/test_framework.py

@@ -209,13 +209,11 @@ class TestFramework:
             if i > 0:
                 time.sleep(1)
             node.start()
+            node.wait_for_rpc_connection()
 
         self.log.info("Wait the zgs_node launch for %d seconds", self.launch_wait_seconds)
         time.sleep(self.launch_wait_seconds)
         
-        for node in self.nodes:
-            node.wait_for_rpc_connection()
-
     def add_arguments(self, parser: argparse.ArgumentParser):
         parser.add_argument(
             "--conflux-binary",