node/src/config/mod.rs

@@ -33,7 +33,7 @@ build_config! {
     (blockchain_rpc_endpoint, (String), "http://127.0.0.1:8545".to_string())
     (log_contract_address, (String), "".to_string())
     (log_sync_start_block_number, (u64), 0)
-    (confirmation_block_count, (u64), 3)
+    (confirmation_block_count, (u64), 12)
     (log_page_size, (u64), 999)
     (max_cache_data_size, (usize), 100 * 1024 * 1024) // 100 MB
     (cache_tx_seq_ttl, (usize), 500)

node/sync/src/auto_sync/batcher_random.rs

@@ -1,6 +1,6 @@
 use super::{batcher::Batcher, sync_store::SyncStore};
 use crate::{
-    auto_sync::{batcher::SyncResult, metrics, sync_store::Queue},
+    auto_sync::{batcher::SyncResult, metrics},
     Config, SyncSender,
 };
 use anyhow::Result;
@@ -108,17 +108,16 @@ impl RandomBatcher {
             SyncResult::Timeout => metrics::RANDOM_SYNC_RESULT_TIMEOUT.inc(1),
         }
 
-        if matches!(sync_result, SyncResult::Completed) {
+        match sync_result {
-            self.sync_store.remove(tx_seq).await?;
+            SyncResult::Completed => self.sync_store.remove_tx(tx_seq).await?,
-        } else {
+            _ => self.sync_store.downgrade_tx_to_pending(tx_seq).await?,
-            self.sync_store.insert(tx_seq, Queue::Pending).await?;
+        };
-        }
 
         Ok(true)
     }
 
     async fn schedule(&mut self) -> Result<bool> {
-        let tx_seq = match self.sync_store.random().await? {
+        let tx_seq = match self.sync_store.random_tx().await? {
             Some(v) => v,
             None => return Ok(false),
         };

node/sync/src/auto_sync/batcher_serial.rs

@@ -2,10 +2,7 @@ use super::{
     batcher::{Batcher, SyncResult},
     sync_store::SyncStore,
 };
-use crate::{
+use crate::{auto_sync::metrics, Config, SyncSender};
-    auto_sync::{metrics, sync_store::Queue},
-    Config, SyncSender,
-};
 use anyhow::Result;
 use log_entry_sync::LogSyncEvent;
 use serde::{Deserialize, Serialize};
@@ -203,7 +200,7 @@ impl SerialBatcher {
         }
 
         // otherwise, mark tx as ready for sync
-        if let Err(err) = self.sync_store.upgrade(announced_tx_seq).await {
+        if let Err(err) = self.sync_store.upgrade_tx_to_ready(announced_tx_seq).await {
             error!(%err, %announced_tx_seq, "Failed to promote announced tx to ready, state = {:?}", self.get_state().await);
         }
 
@@ -283,28 +280,11 @@ impl SerialBatcher {
 
     /// Schedule file sync in sequence.
     async fn schedule_next(&mut self) -> Result<bool> {
-        let max_tx_seq = self.max_tx_seq.load(Ordering::Relaxed);
+        let next_tx_seq = self.next_tx_seq.load(Ordering::Relaxed);
-        if max_tx_seq == u64::MAX {
+        if next_tx_seq > self.max_tx_seq.load(Ordering::Relaxed) {
             return Ok(false);
         }
 
-        let mut next_tx_seq = self.next_tx_seq.load(Ordering::Relaxed);
-        if next_tx_seq > max_tx_seq {
-            return Ok(false);
-        }
-
-        // If sequential sync disabled, delegate to random sync.
-        if self.config.max_sequential_workers == 0 {
-            self.sync_store.insert(next_tx_seq, Queue::Ready).await?;
-
-            next_tx_seq += 1;
-            self.sync_store.set_next_tx_seq(next_tx_seq).await?;
-            self.next_tx_seq.store(next_tx_seq, Ordering::Relaxed);
-            self.next_tx_seq_in_db.store(next_tx_seq, Ordering::Relaxed);
-
-            return Ok(true);
-        }
-
         if !self.batcher.add(next_tx_seq).await? {
             return Ok(false);
         }
@@ -329,7 +309,7 @@ impl SerialBatcher {
 
             // downgrade to random sync if file sync failed or timeout
             if matches!(sync_result, SyncResult::Failed | SyncResult::Timeout) {
-                self.sync_store.insert(current, Queue::Pending).await?;
+                self.sync_store.add_pending_tx(current).await?;
             }
 
             // always move forward in db

node/sync/src/auto_sync/sync_store.rs

@@ -8,20 +8,6 @@ use tokio::sync::RwLock;
 const KEY_NEXT_TX_SEQ: &str = "sync.manager.next_tx_seq";
 const KEY_MAX_TX_SEQ: &str = "sync.manager.max_tx_seq";
 
-#[derive(Debug, PartialEq, Eq)]
-pub enum Queue {
-    Ready,
-    Pending,
-}
-
-#[derive(Debug, PartialEq, Eq)]
-pub enum InsertResult {
-    NewAdded,      // new added in target queue
-    AlreadyExists, // already exists in target queue
-    Upgraded,      // upgraded from pending queue to ready queue
-    Downgraded,    // downgraged from ready queue to pending queue
-}
-
 pub struct SyncStore {
     store: Arc<RwLock<Store>>,
 
@@ -78,69 +64,31 @@ impl SyncStore {
         store.set_config_encoded(&KEY_MAX_TX_SEQ, &tx_seq)
     }
 
-    pub async fn contains(&self, tx_seq: u64) -> Result<Option<Queue>> {
+    pub async fn add_pending_tx(&self, tx_seq: u64) -> Result<bool> {
-        let async_store = self.store.read().await;
+        let async_store = self.store.write().await;
         let store = async_store.get_store();
 
+        // already in ready queue
         if self.ready_txs.has(store, tx_seq)? {
-            return Ok(Some(Queue::Ready));
+            return Ok(false);
         }
 
-        if self.pending_txs.has(store, tx_seq)? {
+        // always add in pending queue
-            return Ok(Some(Queue::Pending));
+        self.pending_txs.add(store, None, tx_seq)
-        }
-
-        Ok(None)
     }
 
-    pub async fn insert(&self, tx_seq: u64, queue: Queue) -> Result<InsertResult> {
+    pub async fn upgrade_tx_to_ready(&self, tx_seq: u64) -> Result<bool> {
-        let async_store = self.store.write().await;
-        let store = async_store.get_store();
-
-        let mut tx = ConfigTx::default();
-
-        match queue {
-            Queue::Ready => {
-                if !self.ready_txs.add(store, Some(&mut tx), tx_seq)? {
-                    return Ok(InsertResult::AlreadyExists);
-                }
-
-                let removed = self.pending_txs.remove(store, Some(&mut tx), tx_seq)?;
-                store.exec_configs(tx)?;
-
-                if removed {
-                    Ok(InsertResult::Upgraded)
-                } else {
-                    Ok(InsertResult::NewAdded)
-                }
-            }
-            Queue::Pending => {
-                if !self.pending_txs.add(store, Some(&mut tx), tx_seq)? {
-                    return Ok(InsertResult::AlreadyExists);
-                }
-
-                let removed = self.ready_txs.remove(store, Some(&mut tx), tx_seq)?;
-                store.exec_configs(tx)?;
-
-                if removed {
-                    Ok(InsertResult::Downgraded)
-                } else {
-                    Ok(InsertResult::NewAdded)
-                }
-            }
-        }
-    }
-
-    pub async fn upgrade(&self, tx_seq: u64) -> Result<bool> {
         let async_store = self.store.write().await;
         let store = async_store.get_store();
 
         let mut tx = ConfigTx::default();
 
+        // not in pending queue
         if !self.pending_txs.remove(store, Some(&mut tx), tx_seq)? {
             return Ok(false);
         }
 
+        // move from pending to ready queue
         let added = self.ready_txs.add(store, Some(&mut tx), tx_seq)?;
 
         store.exec_configs(tx)?;
@@ -148,7 +96,26 @@ impl SyncStore {
         Ok(added)
     }
 
-    pub async fn random(&self) -> Result<Option<u64>> {
+    pub async fn downgrade_tx_to_pending(&self, tx_seq: u64) -> Result<bool> {
+        let async_store = self.store.write().await;
+        let store = async_store.get_store();
+
+        let mut tx = ConfigTx::default();
+
+        // not in ready queue
+        if !self.ready_txs.remove(store, Some(&mut tx), tx_seq)? {
+            return Ok(false);
+        }
+
+        // move from ready to pending queue
+        let added = self.pending_txs.add(store, Some(&mut tx), tx_seq)?;
+
+        store.exec_configs(tx)?;
+
+        Ok(added)
+    }
+
+    pub async fn random_tx(&self) -> Result<Option<u64>> {
         let async_store = self.store.read().await;
         let store = async_store.get_store();
 
@@ -161,21 +128,17 @@ impl SyncStore {
         self.pending_txs.random(store)
     }
 
-    pub async fn remove(&self, tx_seq: u64) -> Result<Option<Queue>> {
+    pub async fn remove_tx(&self, tx_seq: u64) -> Result<bool> {
         let async_store = self.store.write().await;
         let store = async_store.get_store();
 
         // removed in ready queue
         if self.ready_txs.remove(store, None, tx_seq)? {
-            return Ok(Some(Queue::Ready));
+            return Ok(true);
         }
 
-        // removed in pending queue
+        // otherwise, try to remove in pending queue
-        if self.pending_txs.remove(store, None, tx_seq)? {
+        self.pending_txs.remove(store, None, tx_seq)
-            return Ok(Some(Queue::Pending));
-        }
-
-        Ok(None)
     }
 }
 
@@ -183,7 +146,7 @@ impl SyncStore {
 mod tests {
     use crate::test_util::tests::TestStoreRuntime;
 
-    use super::{InsertResult::*, Queue::*, SyncStore};
+    use super::SyncStore;
 
     #[tokio::test]
     async fn test_tx_seq_range() {
@@ -202,79 +165,106 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn test_insert() {
+    async fn test_add_pending_tx() {
         let runtime = TestStoreRuntime::default();
         let store = SyncStore::new(runtime.store.clone());
 
-        assert_eq!(store.contains(1).await.unwrap(), None);
+        // add pending tx 3
-        assert_eq!(store.insert(1, Pending).await.unwrap(), NewAdded);
+        assert!(store.add_pending_tx(3).await.unwrap());
-        assert_eq!(store.contains(1).await.unwrap(), Some(Pending));
-        assert_eq!(store.insert(1, Pending).await.unwrap(), AlreadyExists);
-        assert_eq!(store.insert(1, Ready).await.unwrap(), Upgraded);
-        assert_eq!(store.contains(1).await.unwrap(), Some(Ready));
 
-        assert_eq!(store.insert(2, Ready).await.unwrap(), NewAdded);
+        // cannot add pending tx 3 again
-        assert_eq!(store.contains(2).await.unwrap(), Some(Ready));
+        assert!(!store.add_pending_tx(3).await.unwrap());
-        assert_eq!(store.insert(2, Ready).await.unwrap(), AlreadyExists);
-        assert_eq!(store.insert(2, Pending).await.unwrap(), Downgraded);
-        assert_eq!(store.contains(2).await.unwrap(), Some(Pending));
     }
 
     #[tokio::test]
-    async fn test_upgrade() {
+    async fn test_upgrade_tx() {
         let runtime = TestStoreRuntime::default();
         let store = SyncStore::new(runtime.store.clone());
 
         // cannot upgrade by default
-        assert!(!store.upgrade(3).await.unwrap());
+        assert!(!store.upgrade_tx_to_ready(3).await.unwrap());
 
         // add pending tx 3
-        assert_eq!(store.insert(3, Pending).await.unwrap(), NewAdded);
+        assert!(store.add_pending_tx(3).await.unwrap());
 
         // can upgrade to ready
-        assert!(store.upgrade(3).await.unwrap());
+        assert!(store.upgrade_tx_to_ready(3).await.unwrap());
-        assert_eq!(store.contains(3).await.unwrap(), Some(Ready));
+
+        // cannot add pending tx 3 again event upgraded to ready
+        assert!(!store.add_pending_tx(3).await.unwrap());
 
         // cannot upgrade again
-        assert!(!store.upgrade(3).await.unwrap());
+        assert!(!store.upgrade_tx_to_ready(3).await.unwrap());
     }
 
     #[tokio::test]
-    async fn test_random() {
+    async fn test_downgrade_tx() {
+        let runtime = TestStoreRuntime::default();
+        let store = SyncStore::new(runtime.store.clone());
+
+        // cannot downgrade by default
+        assert!(!store.downgrade_tx_to_pending(3).await.unwrap());
+
+        // add pending tx 3
+        assert!(store.add_pending_tx(3).await.unwrap());
+
+        // cannot downgrade for non-ready tx
+        assert!(!store.downgrade_tx_to_pending(3).await.unwrap());
+
+        // upgrade tx 3 to ready
+        assert!(store.upgrade_tx_to_ready(3).await.unwrap());
+
+        // can downgrade now
+        assert!(store.downgrade_tx_to_pending(3).await.unwrap());
+
+        // cannot downgrade now
+        assert!(!store.downgrade_tx_to_pending(3).await.unwrap());
+    }
+
+    #[tokio::test]
+    async fn test_random_tx() {
         let runtime = TestStoreRuntime::default();
         let store = SyncStore::new(runtime.store.clone());
 
         // no tx by default
-        assert_eq!(store.random().await.unwrap(), None);
+        assert_eq!(store.random_tx().await.unwrap(), None);
 
         // add pending txs 1, 2, 3
-        assert_eq!(store.insert(1, Pending).await.unwrap(), NewAdded);
+        assert!(store.add_pending_tx(1).await.unwrap());
-        assert_eq!(store.insert(2, Pending).await.unwrap(), NewAdded);
+        assert!(store.add_pending_tx(2).await.unwrap());
-        assert_eq!(store.insert(3, Pending).await.unwrap(), NewAdded);
+        assert!(store.add_pending_tx(3).await.unwrap());
-        let tx = store.random().await.unwrap().unwrap();
+        let tx = store.random_tx().await.unwrap().unwrap();
         assert!((1..=3).contains(&tx));
 
-        // upgrade tx 2 to ready
+        // upgrade tx 1 to ready
-        assert_eq!(store.insert(2, Ready).await.unwrap(), Upgraded);
+        assert!(store.upgrade_tx_to_ready(2).await.unwrap());
-        assert_eq!(store.random().await.unwrap(), Some(2));
+        assert_eq!(store.random_tx().await.unwrap(), Some(2));
     }
 
     #[tokio::test]
-    async fn test_remove() {
+    async fn test_remove_tx() {
         let runtime = TestStoreRuntime::default();
         let store = SyncStore::new(runtime.store.clone());
 
         // cannot remove by default
-        assert_eq!(store.remove(1).await.unwrap(), None);
+        assert!(!store.remove_tx(1).await.unwrap());
 
-        // add tx 1, 2
+        // add pending tx 1, 2
-        assert_eq!(store.insert(1, Pending).await.unwrap(), NewAdded);
+        assert!(store.add_pending_tx(1).await.unwrap());
-        assert_eq!(store.insert(2, Ready).await.unwrap(), NewAdded);
+        assert!(store.add_pending_tx(2).await.unwrap());
 
-        // remove txs
+        // upgrade tx 1 to ready
-        assert_eq!(store.remove(1).await.unwrap(), Some(Pending));
+        assert!(store.upgrade_tx_to_ready(1).await.unwrap());
-        assert_eq!(store.remove(1).await.unwrap(), None);
+        assert_eq!(store.random_tx().await.unwrap(), Some(1));
-        assert_eq!(store.remove(2).await.unwrap(), Some(Ready));
+
-        assert_eq!(store.remove(2).await.unwrap(), None);
+        // remove tx 1
+        assert!(store.remove_tx(1).await.unwrap());
+        assert_eq!(store.random_tx().await.unwrap(), Some(2));
+        assert!(!store.remove_tx(1).await.unwrap());
+
+        // remove tx 2
+        assert!(store.remove_tx(2).await.unwrap());
+        assert_eq!(store.random_tx().await.unwrap(), None);
+        assert!(!store.remove_tx(2).await.unwrap());
     }
 }

run/config-testnet-standard.toml

@@ -87,7 +87,7 @@ log_contract_address = "0x0460aA47b41a66694c0a73f667a1b795A5ED3556"
 log_sync_start_block_number = 595059
 
 # Number of blocks to confirm a transaction.
-# confirmation_block_count = 3
+confirmation_block_count = 6
 
 # Maximum number of event logs to poll at a time.
 # log_page_size = 999

run/config-testnet-turbo.toml

@@ -87,7 +87,7 @@ log_contract_address = "0xbD2C3F0E65eDF5582141C35969d66e34629cC768"
 log_sync_start_block_number = 595059
 
 # Number of blocks to confirm a transaction.
-# confirmation_block_count = 3
+confirmation_block_count = 6
 
 # Maximum number of event logs to poll at a time.
 # log_page_size = 999

run/config.toml

@@ -87,7 +87,7 @@
 # log_sync_start_block_number = 0
 
 # Number of blocks to confirm a transaction.
-# confirmation_block_count = 3
+# confirmation_block_count = 12
 
 # Maximum number of event logs to poll at a time.
 # log_page_size = 999

tests/config/node_config.py

@@ -63,13 +63,3 @@ TX_PARAMS1 = {
 
 NO_SEAL_FLAG = 0x1
 NO_MERKLE_PROOF_FLAG = 0x2
-
-def update_config(default: dict, custom: dict):
-    """
-    Supports to update configurations with dict value.
-    """
-    for (key, value) in custom.items():
-        if default.get(key) is None or type(value) != dict:
-            default[key] = value
-        else:
-            update_config(default[key], value)

tests/random_test.py

@@ -5,14 +5,18 @@ import random
 from test_framework.test_framework import TestFramework
 from utility.submission import create_submission
 from utility.submission import submit_data
-from utility.utils import wait_until
+from utility.utils import (
+    assert_equal,
+    wait_until,
+)
+
 
 class RandomTest(TestFramework):
     def setup_params(self):
         self.num_blockchain_nodes = 1
         self.num_nodes = 4
         for i in range(self.num_nodes):
-            self.zgs_node_configs[i] = {"sync": {"auto_sync_enabled": True}}
+            self.zgs_node_configs[i] = {"find_peer_timeout_secs": 1, "confirmation_block_count": 1, "sync": {"auto_sync_enabled": True}}
 
     def run_test(self):
         max_size = 256 * 1024 * 64

tests/sync_auto_random_test.py

@@ -1,32 +0,0 @@
-#!/usr/bin/env python3
-
-from test_framework.test_framework import TestFramework
-from utility.utils import wait_until
-
-class AutoRandomSyncTest(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": 0,
-                    "max_random_workers": 3,
-                }
-            }
-
-    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 node 1 via auto sync
-        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"])
-        wait_until(lambda: self.nodes[1].zgs_get_file_info(data_root_2) is not None)
-        wait_until(lambda: self.nodes[1].zgs_get_file_info(data_root_2)["finalized"])
-
-if __name__ == "__main__":
-    AutoRandomSyncTest().main()

tests/sync_auto_sequential_test.py

@@ -1,32 +0,0 @@
-#!/usr/bin/env python3
-
-from test_framework.test_framework import TestFramework
-from utility.utils import wait_until
-
-class AutoSequentialSyncTest(TestFramework):
-    def setup_params(self):
-        self.num_nodes = 2
-
-        # Enable sequential 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": 0,
-                }
-            }
-
-    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 node 1 via auto sync
-        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"])
-        wait_until(lambda: self.nodes[1].zgs_get_file_info(data_root_2) is not None)
-        wait_until(lambda: self.nodes[1].zgs_get_file_info(data_root_2)["finalized"])
-
-if __name__ == "__main__":
-    AutoSequentialSyncTest().main()

tests/sync_auto_test.py

@@ -1,32 +0,0 @@
-#!/usr/bin/env python3
-
-from test_framework.test_framework import TestFramework
-from utility.utils import wait_until
-
-class AutoSyncTest(TestFramework):
-    def setup_params(self):
-        self.num_nodes = 2
-
-        # Enable auto sync
-        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,
-                }
-            }
-
-    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 node 1 via auto sync
-        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"])
-        wait_until(lambda: self.nodes[1].zgs_get_file_info(data_root_2) is not None)
-        wait_until(lambda: self.nodes[1].zgs_get_file_info(data_root_2)["finalized"])
-
-if __name__ == "__main__":
-    AutoSyncTest().main()

tests/sync_test.py

@@ -4,7 +4,8 @@ import random
 import time
 
 from test_framework.test_framework import TestFramework
-from utility.submission import data_to_segments
+from utility.submission import create_submission
+from utility.submission import submit_data, data_to_segments
 from utility.utils import (
     assert_equal,
     wait_until,
@@ -12,7 +13,9 @@ from utility.utils import (
 
 class SyncTest(TestFramework):
     def setup_params(self):
+        self.num_blockchain_nodes = 2
         self.num_nodes = 2
+        self.__deployed_contracts = 0
 
     def run_test(self):
         # By default, auto_sync_enabled and sync_file_on_announcement_enabled are both false,
@@ -29,8 +32,19 @@ class SyncTest(TestFramework):
         # stop client2, preventing it from receiving AnnounceFile
         client2.shutdown()
 
-        data_root = self.__upload_file__(0, 256 * 1024)
+        # Create submission
+        chunk_data = random.randbytes(256 * 1024)
+        data_root = self.__create_submission(chunk_data)
 
+        # Ensure log entry sync from blockchain node
+        wait_until(lambda: client1.zgs_get_file_info(data_root) is not None)
+        assert_equal(client1.zgs_get_file_info(data_root)["finalized"], False)
+
+        # Upload file to storage node
+        segments = submit_data(client1, chunk_data)
+        self.log.info("segments: %s", [(s["root"], s["index"], s["proof"]) for s in segments])
+        wait_until(lambda: client1.zgs_get_file_info(data_root)["finalized"])
+        
         # restart client2
         client2.start()
         client2.wait_for_rpc_connection()
@@ -61,7 +75,7 @@ class SyncTest(TestFramework):
 
         # Prepare 3 segments to upload
         chunk_data = random.randbytes(256 * 1024 * 3)
-        data_root = self.__submit_file__(chunk_data)
+        data_root = self.__create_submission(chunk_data)
 
         # Ensure log entry sync from blockchain node
         wait_until(lambda: client1.zgs_get_file_info(data_root) is not None)
@@ -97,5 +111,13 @@ class SyncTest(TestFramework):
         # Validate data
         assert_equal(client2.zgs_download_segment_decoded(data_root, 1024, 2048), chunk_data[1024*256:2048*256])
 
+    def __create_submission(self, chunk_data: bytes) -> str:
+        submissions, data_root = create_submission(chunk_data)
+        self.contract.submit(submissions)
+        self.__deployed_contracts += 1
+        wait_until(lambda: self.contract.num_submissions() == self.__deployed_contracts)
+        self.log.info("Submission created, data root: %s, submissions(%s) = %s", data_root, len(submissions), submissions)
+        return data_root
+
 if __name__ == "__main__":
     SyncTest().main()

tests/test_framework/blockchain_node.py

@@ -2,7 +2,9 @@ import os
 import subprocess
 import tempfile
 import time
+import rlp
 
+from eth_utils import decode_hex, keccak
 from web3 import Web3, HTTPProvider
 from web3.middleware import construct_sign_and_send_raw_middleware
 from enum import Enum, unique
@@ -10,6 +12,7 @@ from config.node_config import (
     GENESIS_PRIV_KEY,
     GENESIS_PRIV_KEY1,
     TX_PARAMS,
+    MINER_ID,
 )
 from utility.simple_rpc_proxy import SimpleRpcProxy
 from utility.utils import (

tests/test_framework/contracts.py

@@ -1,5 +1,7 @@
 from pathlib import Path
 import json
+from web3 import Web3
+
 
 def load_contract_metadata(path: str, name: str):
     path = Path(path)

tests/test_framework/test_framework.py

@@ -20,9 +20,8 @@ from test_framework.zgs_node import ZgsNode
 from test_framework.blockchain_node import BlockChainNodeType
 from test_framework.conflux_node import ConfluxNode, connect_sample_nodes
 from test_framework.zg_node import ZGNode, zg_node_init_genesis
-from utility.utils import PortMin, is_windows_platform, wait_until, assert_equal
+from utility.utils import PortMin, is_windows_platform, wait_until
 from utility.build_binary import build_cli
-from utility.submission import create_submission, submit_data
 
 __file_path__ = os.path.dirname(os.path.realpath(__file__))
 
@@ -41,8 +40,8 @@ class TestFramework:
         if "http_proxy" in os.environ:
             del os.environ["http_proxy"]
 
-        self.num_blockchain_nodes = 1
+        self.num_blockchain_nodes = None
-        self.num_nodes = 1
+        self.num_nodes = None
         self.blockchain_nodes = []
         self.nodes = []
         self.contract = None
@@ -54,7 +53,6 @@ class TestFramework:
         self.mine_period = 100
         self.lifetime_seconds = 3600
         self.launch_wait_seconds = 1
-        self.num_deployed_contracts = 0
 
         # Set default binary path
         binary_ext = ".exe" if is_windows_platform() else ""
@@ -400,31 +398,6 @@ class TestFramework:
 
         return root
 
-    def __submit_file__(self, chunk_data: bytes) -> str:
-        submissions, data_root = create_submission(chunk_data)
-        self.contract.submit(submissions)
-        self.num_deployed_contracts += 1
-        wait_until(lambda: self.contract.num_submissions() == self.num_deployed_contracts)
-        self.log.info("Submission completed, data root: %s, submissions(%s) = %s", data_root, len(submissions), submissions)
-        return data_root
-
-    def __upload_file__(self, node_index: int, random_data_size: int) -> str:
-        # Create submission
-        chunk_data = random.randbytes(random_data_size)
-        data_root = self.__submit_file__(chunk_data)
-
-        # Ensure log entry sync from blockchain node
-        client = self.nodes[node_index]
-        wait_until(lambda: client.zgs_get_file_info(data_root) is not None)
-        assert_equal(client.zgs_get_file_info(data_root)["finalized"], False)
-
-        # Upload file to storage node
-        segments = submit_data(client, chunk_data)
-        self.log.info("segments: %s", [(s["root"], s["index"], s["proof"]) for s in segments])
-        wait_until(lambda: client.zgs_get_file_info(data_root)["finalized"])
-
-        return data_root
-
     def setup_params(self):
         self.num_blockchain_nodes = 1
         self.num_nodes = 1

tests/test_framework/zgs_node.py

@@ -2,8 +2,9 @@ import os
 import shutil
 import base64
 
-from config.node_config import ZGS_CONFIG, update_config
+from config.node_config import ZGS_CONFIG
 from test_framework.blockchain_node import NodeType, TestNode
+from config.node_config import MINER_ID
 from utility.utils import (
     initialize_toml_config,
     p2p_port,
@@ -11,6 +12,7 @@ from utility.utils import (
     blockchain_rpc_port,
 )
 
+
 class ZgsNode(TestNode):
     def __init__(
         self,
@@ -46,9 +48,9 @@ class ZgsNode(TestNode):
             "blockchain_rpc_endpoint": f"http://127.0.0.1:{blockchain_rpc_port(0)}",
         }
         # Set configs for this specific node.
-        update_config(local_conf, indexed_config)
+        local_conf.update(indexed_config)
         # Overwrite with personalized configs.
-        update_config(local_conf, updated_config)
+        local_conf.update(updated_config)
         data_dir = os.path.join(root_dir, "zgs_node" + str(index))
         rpc_url = "http://" + local_conf["rpc_listen_address"]
         super().__init__(

tests/utility/submission.py

@@ -1,6 +1,6 @@
 import base64
 
-from eth_utils import decode_hex
+from eth_utils import encode_hex, decode_hex
 from math import log2
 from utility.merkle_tree import add_0x_prefix, Leaf, MerkleTree
 from utility.spec import ENTRY_SIZE, PORA_CHUNK_SIZE

tests/utility/utils.py

@@ -1,10 +1,15 @@
 import base64
 import inspect
+import os
 import platform
 import rtoml
 import time
 import sha3
 
+from config.node_config import ZGS_CONFIG
+from eth_utils import encode_hex
+
+
 class PortMin:
     # Must be initialized with a unique integer for each process
     n = 11000