Merge pull request #254 from eval-exec/exec/wasm-refactor

quake · web-flow · commit 182f1cad3d8d · 2025-12-09T15:14:08.000+09:00
Refactor: Reduce wasm32 and non-wasm32 platform duplication code
diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
@@ -48,9 +48,11 @@ jobs:
         uses: actions/setup-node@v4
         with:
           node-version: 20
+      - name: Install cargo-binstall
+        uses: cargo-bins/cargo-binstall@v1.16.3
       - name: Install dependencies
         run: |
-          cargo install wasm-pack
+          cargo binstall wasm-pack
       - name: Build
         run: |
           make build-wasm
diff --git a/light-client-lib/src/protocols/filter/block_filter.rs b/light-client-lib/src/protocols/filter/block_filter.rs
@@ -1,8 +1,9 @@
 use super::{components, BAD_MESSAGE_BAN_TIME};
 use crate::protocols::{Peers, Status, StatusCode};
 use crate::storage::Storage;
-use crate::types::RwLock;
+use crate::types::{Duration, Instant, RwLock};
 use crate::utils::network::prove_or_download_matched_blocks;
+use crate::{read_lock, write_lock};
 use ckb_constant::sync::INIT_BLOCKS_IN_TRANSIT_PER_PEER;
 use ckb_network::{
     async_trait, bytes::Bytes, BoxedCKBProtocolContext, CKBProtocolHandler, PeerIndex,
@@ -12,11 +13,7 @@ use golomb_coded_set::{GCSFilterReader, SipHasher24Builder, M, P};
 use log::{debug, info, log_enabled, trace, warn, Level};
 use rand::seq::SliceRandom as _;
 use std::io::Cursor;
-#[cfg(not(target_arch = "wasm32"))]
-use std::time::Instant;
-use std::{sync::Arc, time::Duration};
-#[cfg(target_arch = "wasm32")]
-use web_time::Instant;
+use std::sync::Arc;
 
 pub(crate) const GET_BLOCK_FILTERS_TOKEN: u64 = 0;
 pub(crate) const GET_BLOCK_FILTER_HASHES_TOKEN: u64 = 1;
@@ -82,33 +79,18 @@ impl FilterProtocol {
     }
 
     async fn should_ask(&self, immediately: bool) -> bool {
-        #[cfg(target_arch = "wasm32")]
-        let result = !self.storage.is_filter_scripts_empty()
+        let last_ask = read_lock!(self.last_ask_time);
+        !self.storage.is_filter_scripts_empty()
             && (immediately
-                || self.last_ask_time.read().await.is_none()
-                || self.last_ask_time.read().await.unwrap().elapsed() > GET_BLOCK_FILTERS_TIMEOUT);
-        #[cfg(not(target_arch = "wasm32"))]
-        let result = !self.storage.is_filter_scripts_empty()
-            && (immediately
-                || self.last_ask_time.read().unwrap().is_none()
-                || self.last_ask_time.read().unwrap().unwrap().elapsed()
-                    > GET_BLOCK_FILTERS_TIMEOUT);
-
-        result
+                || last_ask.is_none()
+                || last_ask.unwrap().elapsed() > GET_BLOCK_FILTERS_TIMEOUT)
     }
-    #[cfg(target_arch = "wasm32")]
     pub async fn update_min_filtered_block_number(&self, block_number: BlockNumber) {
         self.storage.update_min_filtered_block_number(block_number);
         self.peers
             .update_min_filtered_block_number(block_number)
             .await;
-        self.last_ask_time.write().await.replace(Instant::now());
-    }
-    #[cfg(not(target_arch = "wasm32"))]
-    pub fn update_min_filtered_block_number(&self, block_number: BlockNumber) {
-        self.storage.update_min_filtered_block_number(block_number);
-        self.peers.update_min_filtered_block_number(block_number);
-        self.last_ask_time.write().unwrap().replace(Instant::now());
+        write_lock!(self.last_ask_time).replace(Instant::now());
     }
     pub(crate) async fn try_send_get_block_filters(
         &self,
@@ -129,12 +111,8 @@ impl FilterProtocol {
             let finalized_check_point_number = self
                 .peers
                 .calc_check_point_number(finalized_check_point_index);
-            #[cfg(target_arch = "wasm32")]
             let (cached_check_point_index, cached_hashes) =
                 self.peers.get_cached_block_filter_hashes().await;
-            #[cfg(not(target_arch = "wasm32"))]
-            let (cached_check_point_index, cached_hashes) =
-                self.peers.get_cached_block_filter_hashes();
 
             let cached_check_point_number =
                 self.peers.calc_check_point_number(cached_check_point_index);
@@ -213,19 +191,12 @@ impl FilterProtocol {
 
     pub(crate) async fn try_send_get_block_filter_hashes(&self, nc: BoxedCKBProtocolContext) {
         let min_filtered_block_number = self.storage.get_min_filtered_block_number();
-        #[cfg(target_arch = "wasm32")]
         self.peers
             .update_min_filtered_block_number(min_filtered_block_number)
             .await;
-        #[cfg(not(target_arch = "wasm32"))]
-        self.peers
-            .update_min_filtered_block_number(min_filtered_block_number);
 
         let finalized_check_point_index = self.storage.get_max_check_point_index();
-        #[cfg(target_arch = "wasm32")]
         let cached_check_point_index = self.peers.get_cached_block_filter_hashes().await.0;
-        #[cfg(not(target_arch = "wasm32"))]
-        let cached_check_point_index = self.peers.get_cached_block_filter_hashes().0;
 
         if let Some(start_number) = self
             .peers
diff --git a/light-client-lib/src/protocols/filter/components/block_filter_hashes_process.rs b/light-client-lib/src/protocols/filter/components/block_filter_hashes_process.rs
@@ -67,12 +67,8 @@ impl<'a> BlockFilterHashesProcess<'a> {
             .protocol
             .peers
             .calc_check_point_number(finalized_check_point_index);
-        #[cfg(target_arch = "wasm32")]
         let (cached_check_point_index, cached_hashes) =
             self.protocol.peers.get_cached_block_filter_hashes().await;
-        #[cfg(not(target_arch = "wasm32"))]
-        let (cached_check_point_index, cached_hashes) =
-            self.protocol.peers.get_cached_block_filter_hashes();
 
         let cached_check_point_number = self
             .protocol
diff --git a/light-client-lib/src/protocols/filter/components/block_filters_process.rs b/light-client-lib/src/protocols/filter/components/block_filters_process.rs
@@ -105,12 +105,8 @@ impl<'a> BlockFiltersProcess<'a> {
         let (mut parent_block_filter_hash, expected_block_filter_hashes) =
             if start_number <= finalized_check_point_number {
                 // Use cached block filter hashes to check the block filters.
-                #[cfg(target_arch = "wasm32")]
                 let (cached_check_point_index, mut cached_block_filter_hashes) =
                     self.filter.peers.get_cached_block_filter_hashes().await;
-                #[cfg(not(target_arch = "wasm32"))]
-                let (cached_check_point_index, mut cached_block_filter_hashes) =
-                    self.filter.peers.get_cached_block_filter_hashes();
 
                 let cached_check_point_number = self
                     .filter
@@ -250,13 +246,9 @@ impl<'a> BlockFiltersProcess<'a> {
                 .storage
                 .update_block_number(filtered_block_number)
         }
-        #[cfg(target_arch = "wasm32")]
         self.filter
             .update_min_filtered_block_number(filtered_block_number)
             .await;
-        #[cfg(not(target_arch = "wasm32"))]
-        self.filter
-            .update_min_filtered_block_number(filtered_block_number);
 
         let could_request_more_block_filters = self
             .filter
diff --git a/light-client-lib/src/protocols/light_client/peers.rs b/light-client-lib/src/protocols/light_client/peers.rs
@@ -19,8 +19,11 @@ use governor::{clock::DefaultClock, state::keyed::DefaultKeyedStateStore, Quota,
 
 use super::prelude::*;
 use crate::{
+    mutex_lock,
     protocols::{Status, StatusCode, BAD_MESSAGE_ALLOWED_EACH_HOUR, MESSAGE_TIMEOUT},
+    read_lock,
     types::{Mutex, RwLock},
+    write_lock,
 };
 
 pub type BadMessageRateLimiter<T> = RateLimiter<T, DefaultKeyedStateStore<T>, DefaultClock>;
@@ -1299,10 +1302,7 @@ impl Peers {
         self.mark_fetching_headers_timeout(index);
         self.mark_fetching_txs_timeout(index);
         self.inner.remove(&index);
-        #[cfg(target_arch = "wasm32")]
-        self.rate_limiter.lock().await.retain_recent();
-        #[cfg(not(target_arch = "wasm32"))]
-        let _ignore_error = self.rate_limiter.lock().map(|inner| inner.retain_recent());
+        mutex_lock!(self.rate_limiter).retain_recent();
     }
 
     pub(crate) fn get_peers_index(&self) -> Vec<PeerIndex> {
@@ -1644,64 +1644,33 @@ impl Peers {
             Err(StatusCode::PeerIsNotFound.into())
         }
     }
-    #[cfg(target_arch = "wasm32")]
     pub(crate) async fn update_min_filtered_block_number(
         &self,
         min_filtered_block_number: BlockNumber,
     ) {
         let should_cached_check_point_index =
             self.calc_cached_check_point_index_when_sync_at(min_filtered_block_number + 1);
-        let current_cached_check_point_index = self.cached_block_filter_hashes.read().await.0;
+        let current_cached_check_point_index = read_lock!(self.cached_block_filter_hashes).0;
         if current_cached_check_point_index != should_cached_check_point_index {
-            let mut tmp = self.cached_block_filter_hashes.write().await;
-            tmp.0 = should_cached_check_point_index;
-            tmp.1.clear();
-        }
-    }
-    #[cfg(not(target_arch = "wasm32"))]
-    pub(crate) fn update_min_filtered_block_number(&self, min_filtered_block_number: BlockNumber) {
-        let should_cached_check_point_index =
-            self.calc_cached_check_point_index_when_sync_at(min_filtered_block_number + 1);
-        let current_cached_check_point_index =
-            self.cached_block_filter_hashes.read().expect("poisoned").0;
-        if current_cached_check_point_index != should_cached_check_point_index {
-            let mut tmp = self.cached_block_filter_hashes.write().expect("poisoned");
+            let mut tmp = write_lock!(self.cached_block_filter_hashes);
             tmp.0 = should_cached_check_point_index;
             tmp.1.clear();
         }
     }
 
-    #[cfg(target_arch = "wasm32")]
     pub(crate) async fn get_cached_block_filter_hashes(&self) -> (u32, Vec<packed::Byte32>) {
-        self.cached_block_filter_hashes.read().await.clone()
-    }
-    #[cfg(not(target_arch = "wasm32"))]
-    pub(crate) fn get_cached_block_filter_hashes(&self) -> (u32, Vec<packed::Byte32>) {
-        self.cached_block_filter_hashes
-            .read()
-            .expect("poisoned")
-            .clone()
+        read_lock!(self.cached_block_filter_hashes).clone()
     }
     pub(crate) async fn update_cached_block_filter_hashes(&self, hashes: Vec<packed::Byte32>) {
-        #[cfg(target_arch = "wasm32")]
-        {
-            self.cached_block_filter_hashes.write().await.1 = hashes;
-        }
-        #[cfg(not(target_arch = "wasm32"))]
-        {
-            self.cached_block_filter_hashes.write().expect("poisoned").1 = hashes;
-        }
+        write_lock!(self.cached_block_filter_hashes).1 = hashes;
     }
 
     pub(crate) async fn if_cached_block_filter_hashes_require_update(
         &self,
         finalized_check_point_index: u32,
     ) -> Option<BlockNumber> {
         let (cached_index, cached_length) = {
-            #[cfg(target_arch = "wasm32")]
-            let tmp = self.cached_block_filter_hashes.read().await;
-            #[cfg(not(target_arch = "wasm32"))]
-            let tmp = self.cached_block_filter_hashes.read().expect("poisoned");
+            let tmp = read_lock!(self.cached_block_filter_hashes);
             (tmp.0, tmp.1.len())
         };
         if cached_index >= finalized_check_point_index {
@@ -1884,10 +1853,7 @@ impl Peers {
             // Check:
             // - If cached block filter hashes is same check point as the required,
             // - If all block filter hashes in that check point are downloaded.
-            #[cfg(target_arch = "wasm32")]
             let cached_data = self.get_cached_block_filter_hashes().await;
-            #[cfg(not(target_arch = "wasm32"))]
-            let cached_data = self.get_cached_block_filter_hashes();
 
             let current_cached_check_point_index = cached_data.0;
             should_cached_check_point_index == current_cached_check_point_index
@@ -2026,22 +1992,10 @@ impl Peers {
         if self.bad_message_allowed_each_hour == 0 {
             return true;
         }
-        #[cfg(target_arch = "wasm32")]
-        let result = self
-            .rate_limiter
-            .lock()
-            .await
+        mutex_lock!(self.rate_limiter)
             .check_key(&peer_index)
             .map_err(|_| ())
-            .is_err();
-        #[cfg(not(target_arch = "wasm32"))]
-        let result = self
-            .rate_limiter
-            .lock()
-            .map_err(|_| ())
-            .and_then(|inner| inner.check_key(&peer_index).map_err(|_| ()))
-            .is_err();
-        result
+            .is_err()
     }
 }
 
diff --git a/light-client-lib/src/protocols/relayer.rs b/light-client-lib/src/protocols/relayer.rs
@@ -9,14 +9,11 @@ use linked_hash_map::LinkedHashMap;
 use log::{debug, trace, warn};
 use std::collections::{HashMap, HashSet};
 use std::sync::Arc;
-#[cfg(not(target_arch = "wasm32"))]
-use std::time::{Duration, Instant};
-#[cfg(target_arch = "wasm32")]
-use web_time::{Duration, Instant};
 
 use crate::protocols::{Peers, BAD_MESSAGE_BAN_TIME};
 use crate::storage::Storage;
-use crate::types::RwLock;
+use crate::types::{Duration, Instant, RwLock};
+use crate::{read_lock, write_lock};
 
 const CHECK_PENDING_TXS_TOKEN: u64 = 0;
 
@@ -167,37 +164,15 @@ impl CKBProtocolHandler for RelayProtocol {
             debug!("peer={} is ckb2023 enabled, ignore", peer);
             return;
         }
-        #[cfg(target_arch = "wasm32")]
-        let flag = self
-            .pending_txs
-            .read()
-            .await
-            .is_not_empty_and_updated_at(60);
-
-        #[cfg(not(target_arch = "wasm32"))]
-        let flag = self
-            .pending_txs
-            .read()
-            .unwrap()
-            .is_not_empty_and_updated_at(60);
+        let flag = read_lock!(self.pending_txs).is_not_empty_and_updated_at(60);
 
         if flag {
             let peer_id = nc
                 .get_peer(peer)
                 .and_then(|p| extract_peer_id(&p.connected_addr))
                 .unwrap();
-            #[cfg(target_arch = "wasm32")]
-            let tx_hashes = self
-                .pending_txs
-                .write()
-                .await
-                .fetch_transaction_hashes_for_broadcast(peer_id);
-            #[cfg(not(target_arch = "wasm32"))]
-            let tx_hashes = self
-                .pending_txs
-                .write()
-                .unwrap()
-                .fetch_transaction_hashes_for_broadcast(peer_id);
+            let tx_hashes =
+                write_lock!(self.pending_txs).fetch_transaction_hashes_for_broadcast(peer_id);
             if !tx_hashes.is_empty() {
                 let content = packed::RelayTransactionHashes::new_builder()
                     .tx_hashes(tx_hashes.pack())
@@ -246,10 +221,7 @@ impl CKBProtocolHandler for RelayProtocol {
             message.item_name()
         );
         if let packed::RelayMessageUnionReader::GetRelayTransactions(reader) = message {
-            #[cfg(target_arch = "wasm32")]
-            let pending_txs = self.pending_txs.read().await;
-            #[cfg(not(target_arch = "wasm32"))]
-            let pending_txs = self.pending_txs.read().expect("read access should be OK");
+            let pending_txs = read_lock!(self.pending_txs);
             let relay_txs: Vec<_> = reader
                 .tx_hashes()
                 .iter()
@@ -285,19 +257,7 @@ impl CKBProtocolHandler for RelayProtocol {
             CHECK_PENDING_TXS_TOKEN => {
                 // we check pending txs every 2 seconds, if the timestamp of the pending txs is updated in the last minute
                 // and connected relay protocol peers is empty, we try to open the protocol and broadcast the pending txs
-                #[cfg(target_arch = "wasm32")]
-                let flag = self
-                    .pending_txs
-                    .read()
-                    .await
-                    .is_not_empty_and_updated_at(60);
-
-                #[cfg(not(target_arch = "wasm32"))]
-                let flag = self
-                    .pending_txs
-                    .read()
-                    .unwrap()
-                    .is_not_empty_and_updated_at(60);
+                let flag = read_lock!(self.pending_txs).is_not_empty_and_updated_at(60);
 
                 if flag && self.opened_peers.is_empty() {
                     let p2p_control = nc.p2p_control().expect("p2p_control should be exist");
@@ -310,10 +270,7 @@ impl CKBProtocolHandler for RelayProtocol {
                         }
                     }
                 } else {
-                    #[cfg(target_arch = "wasm32")]
-                    let mut pending_txs = self.pending_txs.write().await;
-                    #[cfg(not(target_arch = "wasm32"))]
-                    let mut pending_txs = self.pending_txs.write().unwrap();
+                    let mut pending_txs = write_lock!(self.pending_txs);
                     for (&peer, instant) in self.opened_peers.iter_mut() {
                         if let Some(peer_id) = nc
                             .get_peer(peer)
diff --git a/light-client-lib/src/storage/mod.rs b/light-client-lib/src/storage/mod.rs
diff --git a/light-client-lib/src/types.rs b/light-client-lib/src/types.rs
