Return the first finalized tx by data root if possible. (#278)

* Only use tx seq for tx status.

* Return the first finalized tx by data root if possible.

This index is used for upload/download segments and file status check.
In all the cases, if there is a finalized transaction, we should use it.

node/network/src/lib.rs

@@ -96,7 +96,8 @@ pub use service::{load_private_key, Context, Libp2pEvent, Service, NETWORK_KEY_F
 /// 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];
+pub const PROTOCOL_VERSION_V1: [u8; 3] = [0, 1, 0];
+pub const PROTOCOL_VERSION_V2: [u8; 3] = [0, 2, 0];
 
 /// Application level requests sent to the network.
 #[derive(Debug, Clone, Copy)]

node/rpc/src/zgs/impl.rs

@@ -246,12 +246,7 @@ impl RpcServerImpl {
     }
 
     async fn get_file_info_by_tx(&self, tx: Transaction) -> RpcResult<FileInfo> {
-        let (finalized, pruned) = match self
+        let (finalized, pruned) = match self.ctx.log_store.get_store().get_tx_status(tx.seq)? {
-            .ctx
-            .log_store
-            .get_store()
-            .get_tx_status(TxSeqOrRoot::TxSeq(tx.seq))?
-        {
             Some(TxStatus::Finalized) => (true, false),
             Some(TxStatus::Pruned) => (false, true),
             None => (false, false),

node/src/config/convert.rs

@@ -39,13 +39,18 @@ impl ZgsConfig {
             .await
             .map_err(|e| format!("Unable to get chain id: {:?}", e))?
             .as_u64();
+        let network_protocol_version = if self.sync.neighbors_only {
+            network::PROTOCOL_VERSION_V2
+        } else {
+            network::PROTOCOL_VERSION_V1
+        };
         let local_network_id = NetworkIdentity {
             chain_id,
             flow_address,
             p2p_protocol_version: ProtocolVersion {
-                major: network::PROTOCOL_VERSION[0],
+                major: network_protocol_version[0],
-                minor: network::PROTOCOL_VERSION[1],
+                minor: network_protocol_version[1],
-                build: network::PROTOCOL_VERSION[2],
+                build: network_protocol_version[2],
             },
         };
         network_config.network_id = local_network_id.clone();

node/storage/src/log_store/log_manager.rs

@@ -21,7 +21,6 @@ use rayon::prelude::ParallelSlice;
 use shared_types::{
     bytes_to_chunks, compute_padded_chunk_size, compute_segment_size, Chunk, ChunkArray,
     ChunkArrayWithProof, ChunkWithProof, DataRoot, FlowProof, FlowRangeProof, Merkle, Transaction,
-    TxSeqOrRoot,
 };
 use std::cmp::Ordering;
 
@@ -538,7 +537,15 @@ impl LogStoreRead for LogManager {
     }
 
     fn get_tx_seq_by_data_root(&self, data_root: &DataRoot) -> crate::error::Result<Option<u64>> {
-        self.tx_store.get_first_tx_seq_by_data_root(data_root)
+        let seq_list = self.tx_store.get_tx_seq_list_by_data_root(data_root)?;
+        for tx_seq in &seq_list {
+            if self.tx_store.check_tx_completed(*tx_seq)? {
+                // Return the first finalized tx if possible.
+                return Ok(Some(*tx_seq));
+            }
+        }
+        // No tx is finalized, return the first one.
+        Ok(seq_list.first().cloned())
     }
 
     fn get_chunk_with_proof_by_tx_and_index(
@@ -582,14 +589,7 @@ impl LogStoreRead for LogManager {
         }))
     }
 
-    fn get_tx_status(&self, tx_seq_or_data_root: TxSeqOrRoot) -> Result<Option<TxStatus>> {
+    fn get_tx_status(&self, tx_seq: u64) -> Result<Option<TxStatus>> {
-        let tx_seq = match tx_seq_or_data_root {
-            TxSeqOrRoot::TxSeq(v) => v,
-            TxSeqOrRoot::Root(root) => {
-                try_option!(self.tx_store.get_first_tx_seq_by_data_root(&root)?)
-            }
-        };
-
         self.tx_store.get_tx_status(tx_seq)
     }
 

node/storage/src/log_store/mod.rs

@@ -4,7 +4,7 @@ use ethereum_types::H256;
 use flow_store::PadPair;
 use shared_types::{
     Chunk, ChunkArray, ChunkArrayWithProof, ChunkWithProof, DataRoot, FlowProof, FlowRangeProof,
-    Transaction, TxSeqOrRoot,
+    Transaction,
 };
 use zgs_spec::{BYTES_PER_SEAL, SEALS_PER_LOAD};
 
@@ -31,8 +31,12 @@ pub trait LogStoreRead: LogStoreChunkRead {
     fn get_tx_by_seq_number(&self, seq: u64) -> Result<Option<Transaction>>;
 
     /// Get a transaction by the data root of its data.
+    /// If all txs are not finalized, return the first one.
+    /// Otherwise, return the first finalized tx.
     fn get_tx_seq_by_data_root(&self, data_root: &DataRoot) -> Result<Option<u64>>;
 
+    /// If all txs are not finalized, return the first one.
+    /// Otherwise, return the first finalized tx.
     fn get_tx_by_data_root(&self, data_root: &DataRoot) -> Result<Option<Transaction>> {
         match self.get_tx_seq_by_data_root(data_root)? {
             Some(seq) => self.get_tx_by_seq_number(seq),
@@ -58,7 +62,7 @@ pub trait LogStoreRead: LogStoreChunkRead {
 
     fn check_tx_pruned(&self, tx_seq: u64) -> Result<bool>;
 
-    fn get_tx_status(&self, tx_seq_or_data_root: TxSeqOrRoot) -> Result<Option<TxStatus>>;
+    fn get_tx_status(&self, tx_seq: u64) -> Result<Option<TxStatus>>;
 
     fn next_tx_seq(&self) -> u64;
 

node/storage/src/log_store/tx_store.rs

@@ -181,14 +181,6 @@ impl TransactionStore {
         Ok(Vec::<u64>::from_ssz_bytes(&value).map_err(Error::from)?)
     }
 
-    pub fn get_first_tx_seq_by_data_root(&self, data_root: &DataRoot) -> Result<Option<u64>> {
-        let value = try_option!(self
-            .kvdb
-            .get(COL_TX_DATA_ROOT_INDEX, data_root.as_bytes())?);
-        let seq_list = Vec::<u64>::from_ssz_bytes(&value).map_err(Error::from)?;
-        Ok(seq_list.first().cloned())
-    }
-
     #[instrument(skip(self))]
     pub fn finalize_tx(&self, tx_seq: u64) -> Result<()> {
         Ok(self.kvdb.put(

node/sync/src/auto_sync/batcher.rs

@@ -1,7 +1,6 @@
 use crate::{controllers::SyncState, SyncRequest, SyncResponse, SyncSender};
 use anyhow::{bail, Result};
 use serde::{Deserialize, Serialize};
-use shared_types::TxSeqOrRoot;
 use std::{collections::HashSet, fmt::Debug, sync::Arc, time::Duration};
 use storage_async::Store;
 use tokio::sync::RwLock;
@@ -86,11 +85,7 @@ impl Batcher {
 
     async fn poll_tx(&self, tx_seq: u64) -> Result<Option<SyncResult>> {
         // file already finalized or even pruned
-        if let Some(tx_status) = self
+        if let Some(tx_status) = self.store.get_store().get_tx_status(tx_seq)? {
-            .store
-            .get_store()
-            .get_tx_status(TxSeqOrRoot::TxSeq(tx_seq))?
-        {
             let num_terminated: usize = self.terminate_file_sync(tx_seq, false).await;
             if num_terminated > 0 {
                 info!(%tx_seq, %num_terminated, ?tx_status, "Terminate file sync due to file already completed in db");

node/sync/src/auto_sync/historical_tx_writer.rs

@@ -87,12 +87,7 @@ impl HistoricalTxWriter {
         }
 
         // write tx in sync store if not finalized or pruned
-        if self
+        if self.store.get_store().get_tx_status(next_tx_seq)?.is_none() {
-            .store
-            .get_store()
-            .get_tx_status(shared_types::TxSeqOrRoot::TxSeq(next_tx_seq))?
-            .is_none()
-        {
             self.sync_store.insert(next_tx_seq, Queue::Ready).await?;
         }
 

requirements.txt

@@ -1,10 +1,10 @@
 jsonrpcclient==4.0.3
 pyyaml==6.0.1
-pysha3==1.0.2
+safe-pysha3==1.0.4
-coincurve==18.0.0
+coincurve==20.0.0
-eth-utils==3.0.0
+eth-utils==5.1.0
 py-ecc==7.0.0
-web3==6.14.0
+web3==7.5.0
 eth_tester
 cffi==1.16.0
-rtoml==0.10.0
+rtoml==0.11.0

tests/snapshot_test.py

@@ -0,0 +1,41 @@
+#!/usr/bin/env python3
+
+import os
+import shutil
+from test_framework.test_framework import TestFramework
+from utility.utils import wait_until
+
+class SnapshotTask(TestFramework):
+    def setup_params(self):
+        self.num_nodes = 2
+
+        # Enable random auto sync only
+        for i in range(self.num_nodes):
+            self.zgs_node_configs[i] = {
+                "sync": {
+                    "auto_sync_enabled": True,
+                    "max_sequential_workers": 3,
+                    "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)
+        wait_until(lambda: self.nodes[1].zgs_get_file_info(data_root_1) is not None)
+        wait_until(lambda: self.nodes[1].zgs_get_file_info(data_root_1)["finalized"])
+
+        # Start the last node to verify historical file sync
+        self.nodes[1].shutdown()
+        shutil.rmtree(os.path.join(self.nodes[1].data_dir, 'db/data_db'))
+        
+        self.start_storage_node(1)
+        self.nodes[1].wait_for_rpc_connection()
+        
+        wait_until(lambda: self.nodes[1].zgs_get_file_info(data_root_1) is not None)
+        wait_until(lambda: self.nodes[1].zgs_get_file_info(data_root_1)["finalized"])
+
+
+if __name__ == "__main__":
+    SnapshotTask().main()

tests/test_framework/blockchain_node.py

@@ -4,7 +4,7 @@ import tempfile
 import time
 
 from web3 import Web3, HTTPProvider
-from web3.middleware import construct_sign_and_send_raw_middleware
+from web3.middleware import SignAndSendRawMiddlewareBuilder
 from enum import Enum, unique
 from config.node_config import (
     GENESIS_PRIV_KEY,
@@ -262,7 +262,7 @@ class BlockchainNode(TestNode):
         account1 = w3.eth.account.from_key(GENESIS_PRIV_KEY)
         account2 = w3.eth.account.from_key(GENESIS_PRIV_KEY1)
         w3.middleware_onion.add(
-            construct_sign_and_send_raw_middleware([account1, account2])
+            SignAndSendRawMiddlewareBuilder.build([account1, account2])
         )
         # account = w3.eth.account.from_key(GENESIS_PRIV_KEY1)
         # w3.middleware_onion.add(construct_sign_and_send_raw_middleware(account))
@@ -365,7 +365,7 @@ class BlockchainNode(TestNode):
         account1 = w3.eth.account.from_key(GENESIS_PRIV_KEY)
         account2 = w3.eth.account.from_key(GENESIS_PRIV_KEY1)
         w3.middleware_onion.add(
-            construct_sign_and_send_raw_middleware([account1, account2])
+            SignAndSendRawMiddlewareBuilder.build([account1, account2])
         )
 
         contract_interface = load_contract_metadata(self.contract_path, "Flow")

tests/test_framework/test_framework.py

@@ -44,7 +44,7 @@ class TestFramework:
         self.num_blockchain_nodes = 1
         self.num_nodes = 1
         self.blockchain_nodes = []
-        self.nodes = []
+        self.nodes: list[ZgsNode] = []
         self.contract = None
         self.blockchain_node_configs = {}
         self.zgs_node_configs = {}