Merge pull request #4632 from chenyukang/yukang-backport-118.x

[Backport] merge several fixes into 118 for next rc release

.github/workflows/ci_integration_tests_windows.yaml

@@ -51,7 +51,7 @@ jobs:
   ci_integration_tests_windows:
     name: ci_integration_tests_windows
     needs: prologue
-    runs-on: ${{ needs.prologue.outputs.windows_runner_label }}
+    runs-on: windows-latest
     timeout-minutes: 140
     steps:
     - uses: actions/checkout@v3

chain/src/init.rs

@@ -2,11 +2,11 @@
 
 //! Bootstrap InitLoadUnverified, PreloadUnverifiedBlock, ChainService and ConsumeUnverified threads.
 use crate::chain_service::ChainService;
-use crate::consume_unverified::ConsumeUnverifiedBlocks;
 use crate::init_load_unverified::InitLoadUnverified;
 use crate::orphan_broker::OrphanBroker;
 use crate::preload_unverified_blocks_channel::PreloadUnverifiedBlocksChannel;
 use crate::utils::orphan_block_pool::OrphanBlockPool;
+use crate::verify::ConsumeUnverifiedBlocks;
 use crate::{chain_controller::ChainController, LonelyBlockHash, UnverifiedBlock};
 use ckb_channel::{self as channel, SendError};
 use ckb_constant::sync::BLOCK_DOWNLOAD_WINDOW;
@@ -37,7 +37,7 @@ pub fn start_chain_services(builder: ChainServicesBuilder) -> ChainController {
     let is_pending_verify: Arc<DashSet<Byte32>> = Arc::new(DashSet::new());
 
     let consumer_unverified_thread = thread::Builder::new()
-        .name("consume_unverified_blocks".into())
+        .name("verify_blocks".into())
         .spawn({
             let shared = builder.shared.clone();
             let is_pending_verify = Arc::clone(&is_pending_verify);

chain/src/init_load_unverified.rs

@@ -93,6 +93,15 @@ impl InitLoadUnverified {
             let unverified_hashes: Vec<packed::Byte32> =
                 self.find_unverified_block_hashes(check_unverified_number);
 
+            if check_unverified_number > tip_number && unverified_hashes.is_empty() {
+                info!(
+                    "no unverified blocks found after tip, current tip: {}-{}",
+                    tip_number,
+                    self.shared.snapshot().tip_hash()
+                );
+                return;
+            }
+
             for unverified_hash in unverified_hashes {
                 f(&unverified_hash);
             }

chain/src/lib.rs

@@ -16,14 +16,14 @@ use std::sync::Arc;
 
 mod chain_controller;
 mod chain_service;
-pub mod consume_unverified;
 mod init;
 mod init_load_unverified;
 mod orphan_broker;
 mod preload_unverified_blocks_channel;
 #[cfg(test)]
 mod tests;
 mod utils;
+pub mod verify;
 
 pub use chain_controller::ChainController;
 use ckb_logger::{error, info};

chain/src/tests/find_fork.rs

@@ -1,5 +1,5 @@
-use crate::consume_unverified::ConsumeUnverifiedBlockProcessor;
 use crate::utils::forkchanges::ForkChanges;
+use crate::verify::ConsumeUnverifiedBlockProcessor;
 use crate::{start_chain_services, UnverifiedBlock};
 use ckb_chain_spec::consensus::{Consensus, ProposalWindow};
 use ckb_proposal_table::ProposalTable;

chain/src/consume_unverified.rs → chain/src/verify.rs

@@ -103,7 +103,7 @@ impl ConsumeUnverifiedBlocks {
                     },
                 },
                 recv(self.stop_rx) -> _ => {
-                    info!("consume_unverified_blocks thread received exit signal, exit now");
+                    info!("verify_blocks thread received exit signal, exit now");
                     break;
                 }
 

deny.toml

@@ -70,6 +70,10 @@ feature-depth = 1
 # A list of advisory IDs to ignore. Note that ignored advisories will still
 # output a note when they are encountered.
 ignore = [
+# https://rustsec.org/advisories/RUSTSEC-2024-0363
+# https://github.com/launchbadge/sqlx/issues/3440
+# The queries for the rich indexer receive input parameters via RPC, and the data size is far less than 4GB, so this issue can be temporarily ignored while waiting for sqlx to be fixed.
+  "RUSTSEC-2024-0363",
 # https://rustsec.org/advisories/RUSTSEC-2022-0090
 # It was sometimes possible for SQLite versions >= 1.0.12, < 3.39.2 to allow an array-bounds overflow when large string were input into SQLite's `printf` function.
   "RUSTSEC-2022-0090",

devtools/ci/check-relaxed.sh

@@ -0,0 +1,40 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+case "$OSTYPE" in
+  darwin*)
+    if ! type gsed &>/dev/null || ! type ggrep &>/dev/null; then
+      echo "GNU sed and grep not found! You can install via Homebrew" >&2
+      echo >&2
+      echo "    brew install grep gnu-sed" >&2
+      exit 1
+    fi
+
+    SED=gsed
+    GREP=ggrep
+    ;;
+  *)
+    SED=sed
+    GREP=grep
+    ;;
+esac
+
+function main() {
+  local res=$(find ./ -not -path '*/target/*' -type f -name "*.rs" | xargs grep -H "Relaxed")
+
+  if [ -z "${res}" ]; then
+    echo "ok"
+    exit 0
+  else
+    echo "find use Relaxed on code, please check"
+
+    for file in ${res}; do
+        printf ${file}
+    done
+
+    exit 1
+  fi
+}
+
+main "$@"

devtools/ci/ci_main.sh

@@ -60,6 +60,7 @@ case $GITHUB_WORKFLOW in
     make check-dirty-rpc-doc
     make check-dirty-hashes-toml
     devtools/ci/check-cyclic-dependencies.py
+    devtools/ci/check-relaxed.sh
     ;;
   ci_aarch64_build*)
     echo "ci_aarch64_build"

network/src/network.rs

@@ -485,7 +485,7 @@ impl NetworkState {
 
     /// Network message processing controller, default is true, if false, discard any received messages
     pub fn is_active(&self) -> bool {
-        self.active.load(Ordering::Relaxed)
+        self.active.load(Ordering::Acquire)
     }
 }
 
@@ -1368,7 +1368,7 @@ impl NetworkController {
 
     /// Change active status, if set false discard any received messages
     pub fn set_active(&self, active: bool) {
-        self.network_state.active.store(active, Ordering::Relaxed);
+        self.network_state.active.store(active, Ordering::Release);
     }
 
     /// Return all connected peers' protocols info

rpc/README.md

@@ -109,6 +109,7 @@ The crate `ckb-rpc`'s minimum supported rustc version is 1.71.1.
         * [Method `remove_transaction`](#pool-remove_transaction)
         * [Method `tx_pool_info`](#pool-tx_pool_info)
         * [Method `clear_tx_pool`](#pool-clear_tx_pool)
+        * [Method `clear_tx_verify_queue`](#pool-clear_tx_verify_queue)
         * [Method `get_raw_tx_pool`](#pool-get_raw_tx_pool)
         * [Method `get_pool_tx_detail_info`](#pool-get_pool_tx_detail_info)
         * [Method `tx_pool_ready`](#pool-tx_pool_ready)
@@ -4748,6 +4749,37 @@ Response
 }
 ```
 
+<a id="pool-clear_tx_verify_queue"></a>
+#### Method `clear_tx_verify_queue`
+* `clear_tx_verify_queue()`
+
+* result: `null`
+
+Removes all transactions from the verification queue.
+
+###### Examples
+
+Request
+
+```json
+{
+  "id": 42,
+  "jsonrpc": "2.0",
+  "method": "clear_tx_verify_queue",
+  "params": []
+}
+```
+
+Response
+
+```json
+{
+  "id": 42,
+  "jsonrpc": "2.0",
+  "result": null
+}
+```
+
 <a id="pool-get_raw_tx_pool"></a>
 #### Method `get_raw_tx_pool`
 * `get_raw_tx_pool(verbose)`

rpc/src/module/pool.rs

@@ -322,6 +322,33 @@ pub trait PoolRpc {
     #[rpc(name = "clear_tx_pool")]
     fn clear_tx_pool(&self) -> Result<()>;
 
+    /// Removes all transactions from the verification queue.
+    ///
+    /// ## Examples
+    ///
+    /// Request
+    ///
+    /// ```json
+    /// {
+    ///   "id": 42,
+    ///   "jsonrpc": "2.0",
+    ///   "method": "clear_tx_verify_queue",
+    ///   "params": []
+    /// }
+    /// ```
+    ///
+    /// Response
+    ///
+    /// ```json
+    /// {
+    ///   "id": 42,
+    ///   "jsonrpc": "2.0",
+    ///   "result": null
+    /// }
+    /// ```
+    #[rpc(name = "clear_tx_verify_queue")]
+    fn clear_tx_verify_queue(&self) -> Result<()>;
+
     /// Returns all transaction ids in tx pool as a json array of string transaction ids.
     /// ## Params
     ///
@@ -662,6 +689,15 @@ impl PoolRpc for PoolRpcImpl {
         Ok(())
     }
 
+    fn clear_tx_verify_queue(&self) -> Result<()> {
+        let tx_pool = self.shared.tx_pool_controller();
+        tx_pool
+            .clear_verify_queue()
+            .map_err(|err| RPCError::custom(RPCError::Invalid, err.to_string()))?;
+
+        Ok(())
+    }
+
     fn get_raw_tx_pool(&self, verbose: Option<bool>) -> Result<RawTxPool> {
         let tx_pool = self.shared.tx_pool_controller();
 

script/src/verify/tests/ckb_latest/features_since_v2021.rs

@@ -835,10 +835,12 @@ fn _check_typical_secp256k1_blake160_2_in_2_out_tx_with_state(step_cycles: Cycle
     let mut cycles = 0;
     let verifier = TransactionScriptsVerifierWithEnv::new();
     let result = verifier.verify_map(script_version, &rtx, |verifier| {
+        #[allow(unused_assignments)]
         let mut init_state: Option<TransactionState> = None;
 
-        if let VerifyResult::Suspended(state) = verifier.resumable_verify(step_cycles).unwrap() {
-            init_state = Some(state);
+        match verifier.resumable_verify(step_cycles).unwrap() {
+            VerifyResult::Suspended(state) => init_state = Some(state),
+            VerifyResult::Completed(cycle) => return Ok(cycle),
         }
 
         loop {
@@ -948,12 +950,12 @@ fn _check_typical_secp256k1_blake160_2_in_2_out_tx_with_snap(step_cycles: Cycle)
     if script_version == crate::ScriptVersion::V2 {
         assert!(
             cycles >= TWO_IN_TWO_OUT_CYCLES - V2_CYCLE_BOUND,
-            "step_cycles {step_cycles}"
+            "cycles {cycles} step_cycles {step_cycles}"
         );
     } else {
         assert!(
             cycles >= TWO_IN_TWO_OUT_CYCLES - CYCLE_BOUND,
-            "step_cycles {step_cycles}"
+            "cycles {cycles} step_cycles {step_cycles}"
         );
     }
     assert_eq!(cycles, cycles_once, "step_cycles {step_cycles}");

shared/src/shared.rs

@@ -378,14 +378,14 @@ impl Shared {
     /// Return whether chain is in initial block download
     pub fn is_initial_block_download(&self) -> bool {
         // Once this function has returned false, it must remain false.
-        if self.ibd_finished.load(Ordering::Relaxed) {
+        if self.ibd_finished.load(Ordering::Acquire) {
             false
         } else if unix_time_as_millis().saturating_sub(self.snapshot().tip_header().timestamp())
             > MAX_TIP_AGE
         {
             true
         } else {
-            self.ibd_finished.store(true, Ordering::Relaxed);
+            self.ibd_finished.store(true, Ordering::Release);
             false
         }
     }

shared/src/types/header_map/kernel_lru.rs

@@ -157,7 +157,7 @@ where
             self.stats().tick_primary_delete();
         }
         // If IBD is not finished, don't shrink memory map
-        let allow_shrink_to_fit = self.ibd_finished.load(Ordering::Relaxed);
+        let allow_shrink_to_fit = self.ibd_finished.load(Ordering::Acquire);
         self.memory.remove(hash, allow_shrink_to_fit);
         if self.backend.is_empty() {
             return;
@@ -175,7 +175,7 @@ where
             });
 
             // If IBD is not finished, don't shrink memory map
-            let allow_shrink_to_fit = self.ibd_finished.load(Ordering::Relaxed);
+            let allow_shrink_to_fit = self.ibd_finished.load(Ordering::Acquire);
             self.memory
                 .remove_batch(values.iter().map(|value| value.hash()), allow_shrink_to_fit);
         }

sync/src/synchronizer/mod.rs

@@ -680,6 +680,11 @@ impl Synchronizer {
             return;
         }
 
+        if ckb_stop_handler::has_received_stop_signal() {
+            info!("received stop signal, stop find_blocks_to_fetch");
+            return;
+        }
+
         let unverified_tip = self.shared.active_chain().unverified_tip_number();
 
         let disconnect_list = {

sync/src/tests/types.rs

@@ -8,7 +8,10 @@ use ckb_types::{
 use rand::{thread_rng, Rng};
 use std::{
     collections::{BTreeMap, HashMap},
-    sync::atomic::{AtomicUsize, Ordering::Relaxed},
+    sync::atomic::{
+        AtomicUsize,
+        Ordering::{Acquire, SeqCst},
+    },
 };
 
 use crate::types::{TtlFilter, FILTER_TTL};
@@ -64,15 +67,15 @@ fn test_get_ancestor_use_skip_list() {
                 0,
                 b,
                 |hash, _| {
-                    count.fetch_add(1, Relaxed);
+                    count.fetch_add(1, SeqCst);
                     header_map.get(hash).cloned()
                 },
                 |_, _| None,
             )
             .unwrap();
 
         // Search must finished in <limit> steps
-        assert!(count.load(Relaxed) <= limit);
+        assert!(count.load(Acquire) <= limit);
 
         header
     };

test/src/main.rs

@@ -491,6 +491,7 @@ fn all_specs() -> Vec<Box<dyn Spec>> {
         Box::new(RbfReplaceProposedSuccess),
         Box::new(RbfConcurrency),
         Box::new(RbfCellDepsCheck),
+        Box::new(RbfCyclingAttack),
         Box::new(CompactBlockEmpty),
         Box::new(CompactBlockEmptyParentUnknown),
         Box::new(CompactBlockPrefilled),