add api of getting available file info by root (#357)

* add api of getting available file info by root

Author: 0g-peterzhb

.github/workflows/tests.yml

@@ -45,6 +45,11 @@ jobs:
         python-version: '3.9'
         cache: 'pip'
 
+    - name: Set up Go
+      uses: actions/setup-go@v4
+      with:
+        go-version: '1.22'
+
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip

node/rpc/src/zgs/api.rs

@@ -63,7 +63,11 @@ pub trait Rpc {
     async fn check_file_finalized(&self, tx_seq_or_root: TxSeqOrRoot) -> RpcResult<Option<bool>>;
 
     #[method(name = "getFileInfo")]
-    async fn get_file_info(&self, data_root: DataRoot) -> RpcResult<Option<FileInfo>>;
+    async fn get_file_info(
+        &self,
+        data_root: DataRoot,
+        need_available: bool,
+    ) -> RpcResult<Option<FileInfo>>;
 
     #[method(name = "getFileInfoByTxSeq")]
     async fn get_file_info_by_tx_seq(&self, tx_seq: u64) -> RpcResult<Option<FileInfo>>;

node/rpc/src/zgs/impl.rs

@@ -95,7 +95,7 @@ impl RpcServer for RpcServerImpl {
         let tx_seq = try_option!(
             self.ctx
                 .log_store
-                .get_tx_seq_by_data_root(&data_root)
+                .get_tx_seq_by_data_root(&data_root, true)
                 .await?
         );
 
@@ -121,7 +121,12 @@ impl RpcServer for RpcServerImpl {
     ) -> RpcResult<Option<SegmentWithProof>> {
         info!(%data_root, %index, "zgs_downloadSegmentWithProof");
 
-        let tx = try_option!(self.ctx.log_store.get_tx_by_data_root(&data_root).await?);
+        let tx = try_option!(
+            self.ctx
+                .log_store
+                .get_tx_by_data_root(&data_root, true)
+                .await?
+        );
 
         self.get_segment_with_proof_by_tx(tx, index).await
     }
@@ -144,7 +149,12 @@ impl RpcServer for RpcServerImpl {
         let seq = match tx_seq_or_root {
             TxSeqOrRoot::TxSeq(v) => v,
             TxSeqOrRoot::Root(v) => {
-                try_option!(self.ctx.log_store.get_tx_seq_by_data_root(&v).await?)
+                try_option!(
+                    self.ctx
+                        .log_store
+                        .get_tx_seq_by_data_root(&v, false)
+                        .await?
+                )
             }
         };
 
@@ -163,10 +173,19 @@ impl RpcServer for RpcServerImpl {
         }
     }
 
-    async fn get_file_info(&self, data_root: DataRoot) -> RpcResult<Option<FileInfo>> {
+    async fn get_file_info(
+        &self,
+        data_root: DataRoot,
+        need_available: bool,
+    ) -> RpcResult<Option<FileInfo>> {
         debug!(%data_root, "zgs_getFileInfo");
 
-        let tx = try_option!(self.ctx.log_store.get_tx_by_data_root(&data_root).await?);
+        let tx = try_option!(
+            self.ctx
+                .log_store
+                .get_tx_by_data_root(&data_root, need_available)
+                .await?
+        );
 
         Ok(Some(self.get_file_info_by_tx(tx).await?))
     }
@@ -288,7 +307,7 @@ impl RpcServerImpl {
         let maybe_tx = self
             .ctx
             .log_store
-            .get_tx_by_data_root(&segment.root)
+            .get_tx_by_data_root(&segment.root, false)
             .await?;
 
         self.put_segment_with_maybe_tx(segment, maybe_tx).await

node/storage-async/src/lib.rs

@@ -59,15 +59,23 @@ impl Store {
     delegate!(fn get_proof_at_root(root: Option<DataRoot>, index: u64, length: u64) -> Result<FlowRangeProof>);
     delegate!(fn get_context() -> Result<(DataRoot, u64)>);
 
-    pub async fn get_tx_seq_by_data_root(&self, data_root: &DataRoot) -> Result<Option<u64>> {
+    pub async fn get_tx_seq_by_data_root(
+        &self,
+        data_root: &DataRoot,
+        need_available: bool,
+    ) -> Result<Option<u64>> {
         let root = *data_root;
-        self.spawn(move |store| store.get_tx_seq_by_data_root(&root))
+        self.spawn(move |store| store.get_tx_seq_by_data_root(&root, need_available))
             .await
     }
 
-    pub async fn get_tx_by_data_root(&self, data_root: &DataRoot) -> Result<Option<Transaction>> {
+    pub async fn get_tx_by_data_root(
+        &self,
+        data_root: &DataRoot,
+        need_available: bool,
+    ) -> Result<Option<Transaction>> {
         let root = *data_root;
-        self.spawn(move |store| store.get_tx_by_data_root(&root))
+        self.spawn(move |store| store.get_tx_by_data_root(&root, need_available))
             .await
     }
 

node/storage/src/log_store/log_manager.rs

@@ -511,7 +511,7 @@ impl LogStoreChunkRead for LogManager {
         index_start: usize,
         index_end: usize,
     ) -> crate::error::Result<Option<ChunkArray>> {
-        let tx_seq = try_option!(self.get_tx_seq_by_data_root(data_root)?);
+        let tx_seq = try_option!(self.get_tx_seq_by_data_root(data_root, true)?);
         self.get_chunks_by_tx_and_index_range(tx_seq, index_start, index_end)
     }
 
@@ -536,13 +536,27 @@ impl LogStoreRead for LogManager {
         self.tx_store.get_tx_by_seq_number(seq)
     }
 
-    fn get_tx_seq_by_data_root(&self, data_root: &DataRoot) -> crate::error::Result<Option<u64>> {
+    fn get_tx_seq_by_data_root(
+        &self,
+        data_root: &DataRoot,
+        need_available: bool,
+    ) -> crate::error::Result<Option<u64>> {
         let seq_list = self.tx_store.get_tx_seq_list_by_data_root(data_root)?;
+        let mut available_seq = None;
         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));
             }
+            if need_available
+                && available_seq.is_none()
+                && !self.tx_store.check_tx_pruned(*tx_seq)?
+            {
+                available_seq = Some(*tx_seq);
+            }
+        }
+        if need_available {
+            return Ok(available_seq);
         }
         // No tx is finalized, return the first one.
         Ok(seq_list.first().cloned())

node/storage/src/log_store/mod.rs

@@ -31,14 +31,22 @@ 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.
+    /// If all txs are not finalized, return the first one if need available is false.
     /// Otherwise, return the first finalized tx.
-    fn get_tx_seq_by_data_root(&self, data_root: &DataRoot) -> Result<Option<u64>>;
+    fn get_tx_seq_by_data_root(
+        &self,
+        data_root: &DataRoot,
+        need_available: bool,
+    ) -> Result<Option<u64>>;
 
-    /// If all txs are not finalized, return the first one.
+    /// If all txs are not finalized, return the first one if need available is false.
     /// 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)? {
+    fn get_tx_by_data_root(
+        &self,
+        data_root: &DataRoot,
+        need_available: bool,
+    ) -> Result<Option<Transaction>> {
+        match self.get_tx_seq_by_data_root(data_root, need_available)? {
             Some(seq) => self.get_tx_by_seq_number(seq),
             None => Ok(None),
         }

tests/cache_test.py

@@ -17,7 +17,10 @@ def run_test(self):
         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)
-        wait_until(lambda: not client.zgs_get_file_info(data_root)["isCached"] and client.zgs_get_file_info(data_root)["uploadedSegNum"] == 1)
+        wait_until(
+            lambda: not client.zgs_get_file_info(data_root)["isCached"]
+            and client.zgs_get_file_info(data_root)["uploadedSegNum"] == 1
+        )
         client.zgs_upload_segment(segments[1])
         wait_until(lambda: client.zgs_get_file_info(data_root)["finalized"])
 

tests/config/node_config.py

@@ -7,9 +7,7 @@
     "log_config_file": "log_config",
     "confirmation_block_count": 1,
     "discv5_disable_ip_limit": True,
-    "network_peer_manager": {
-        "heartbeat_interval": "1s"
-    },
+    "network_peer_manager": {"heartbeat_interval": "1s"},
     "router": {
         "private_ip_enabled": True,
     },
@@ -22,7 +20,7 @@
         "auto_sync_idle_interval": "1s",
         "sequential_find_peer_timeout": "10s",
         "random_find_peer_timeout": "10s",
-    }
+    },
 }
 
 CONFIG_DIR = os.path.dirname(__file__)
@@ -75,11 +73,12 @@
 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():
+    for key, value in custom.items():
         if default.get(key) is None or type(value) != dict:
             default[key] = value
         else:

tests/crash_test.py

@@ -20,17 +20,15 @@ def run_test(self):
 
         segment = submit_data(self.nodes[0], chunk_data)
         self.log.info("segment: %s", segment)
-        wait_until(lambda: self.nodes[0].zgs_get_file_info(data_root)["finalized"] is True)
+        wait_until(
+            lambda: self.nodes[0].zgs_get_file_info(data_root)["finalized"] is True
+        )
 
         for i in range(1, self.num_nodes):
-            wait_until(
-                lambda: self.nodes[i].zgs_get_file_info(data_root) is not None
-            )
+            wait_until(lambda: self.nodes[i].zgs_get_file_info(data_root) is not None)
             self.nodes[i].admin_start_sync_file(0)
             self.log.info("wait for node: %s", i)
-            wait_until(
-                lambda: self.nodes[i].zgs_get_file_info(data_root)["finalized"]
-            )
+            wait_until(lambda: self.nodes[i].zgs_get_file_info(data_root)["finalized"])
 
         # 2: first node runnging, other nodes killed
         self.log.info("kill node")
@@ -56,22 +54,16 @@ def run_test(self):
         for i in range(2, self.num_nodes):
             self.start_storage_node(i)
             self.nodes[i].wait_for_rpc_connection()
-            wait_until(
-                lambda: self.nodes[i].zgs_get_file_info(data_root) is not None
-            )
+            wait_until(lambda: self.nodes[i].zgs_get_file_info(data_root) is not None)
             self.nodes[i].admin_start_sync_file(1)
 
             self.nodes[i].stop(kill=True)
             self.start_storage_node(i)
             self.nodes[i].wait_for_rpc_connection()
-            wait_until(
-                lambda: self.nodes[i].zgs_get_file_info(data_root) is not None
-            )
+            wait_until(lambda: self.nodes[i].zgs_get_file_info(data_root) is not None)
             self.nodes[i].admin_start_sync_file(1)
-            
-            wait_until(
-                lambda: self.nodes[i].zgs_get_file_info(data_root)["finalized"]
-            )
+
+            wait_until(lambda: self.nodes[i].zgs_get_file_info(data_root)["finalized"])
 
         # 4: node[1..] synced contract entries and killed
         self.log.info("kill node 0")
@@ -96,13 +88,9 @@ def run_test(self):
             self.log.info("wait for node: %s", i)
             self.start_storage_node(i)
             self.nodes[i].wait_for_rpc_connection()
-            wait_until(
-                lambda: self.nodes[i].zgs_get_file_info(data_root) is not None
-            )
+            wait_until(lambda: self.nodes[i].zgs_get_file_info(data_root) is not None)
             self.nodes[i].admin_start_sync_file(2)
-            wait_until(
-                lambda: self.nodes[i].zgs_get_file_info(data_root)["finalized"]
-            )
+            wait_until(lambda: self.nodes[i].zgs_get_file_info(data_root)["finalized"])
 
         # 5: node[1..] synced contract entries and killed, sync disorder
         self.nodes[0].stop(kill=True)
@@ -137,21 +125,13 @@ def run_test(self):
             self.log.info("wait for node: %s", i)
             self.start_storage_node(i)
             self.nodes[i].wait_for_rpc_connection()
-            wait_until(
-                lambda: self.nodes[i].zgs_get_file_info(data_root1) is not None
-            )
+            wait_until(lambda: self.nodes[i].zgs_get_file_info(data_root1) is not None)
             self.nodes[i].admin_start_sync_file(4)
-            wait_until(
-                lambda: self.nodes[i].zgs_get_file_info(data_root1)["finalized"]
-            )
+            wait_until(lambda: self.nodes[i].zgs_get_file_info(data_root1)["finalized"])
 
-            wait_until(
-                lambda: self.nodes[i].zgs_get_file_info(data_root) is not None
-            )
+            wait_until(lambda: self.nodes[i].zgs_get_file_info(data_root) is not None)
             self.nodes[i].admin_start_sync_file(3)
-            wait_until(
-                lambda: self.nodes[i].zgs_get_file_info(data_root)["finalized"]
-            )
+            wait_until(lambda: self.nodes[i].zgs_get_file_info(data_root)["finalized"])
 
 
 if __name__ == "__main__":

tests/fuzz_test.py

@@ -45,13 +45,9 @@ def run_chunk_size(sizes, nodes, contract, log):
                 lock.release()
 
                 log.info("submit data via client %s", idx)
-                wait_until(
-                    lambda: nodes[idx].zgs_get_file_info(data_root) is not None
-                )
+                wait_until(lambda: nodes[idx].zgs_get_file_info(data_root) is not None)
                 segment = submit_data(nodes[idx], chunk_data)
-                wait_until(
-                    lambda: nodes[idx].zgs_get_file_info(data_root)["finalized"]
-                )
+                wait_until(lambda: nodes[idx].zgs_get_file_info(data_root)["finalized"])
 
                 lock.acquire()
                 nodes_index.append(idx)
@@ -65,17 +61,15 @@ def run_chunk_size(sizes, nodes, contract, log):
                         lambda: nodes[idx].zgs_get_file_info(data_root) is not None
                     )
 
-                    def wait_finalized():    
+                    def wait_finalized():
                         ret = nodes[idx].zgs_get_file_info(data_root)
                         if ret["finalized"]:
                             return True
                         else:
-                            nodes[idx].admin_start_sync_file(ret['tx']['seq'])
+                            nodes[idx].admin_start_sync_file(ret["tx"]["seq"])
                             return False
 
-                    wait_until(
-                        lambda: wait_finalized(), timeout = 180
-                    )
+                    wait_until(lambda: wait_finalized(), timeout=180)
 
         def run_small_chunk_size(nodes, contract, log):
             sizes = [i for i in range(1, SAMLL_SIZE + 1)]
@@ -84,7 +78,7 @@ def run_small_chunk_size(nodes, contract, log):
             run_chunk_size(sizes, nodes, contract, log)
 
         def run_large_chunk_size(nodes, contract, log):
-            sizes = [i for i in range(256 * 1024 * 256 - LARGE_SIZE, 256 * 1024 * 256 )]
+            sizes = [i for i in range(256 * 1024 * 256 - LARGE_SIZE, 256 * 1024 * 256)]
             random.shuffle(sizes)
 
             run_chunk_size(sizes, nodes, contract, log)