Merge branch 'develop' into develop

.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/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;
                 }
 

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

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/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),

test/src/node.rs

@@ -38,6 +38,16 @@ pub(crate) struct ProcessGuard {
     pub killed: bool,
 }
 
+impl ProcessGuard {
+    pub(crate) fn is_alive(&mut self) -> bool {
+        let try_wait = self.child.try_wait();
+        match try_wait {
+            Ok(status_op) => status_op.is_none(),
+            Err(_err) => false,
+        }
+    }
+}
+
 impl Drop for ProcessGuard {
     fn drop(&mut self) {
         if !self.killed {
@@ -738,6 +748,15 @@ impl Node {
         g.take()
     }
 
+    pub(crate) fn is_alive(&mut self) -> bool {
+        let mut g = self.inner.guard.write().unwrap();
+        if let Some(guard) = g.as_mut() {
+            guard.is_alive()
+        } else {
+            false
+        }
+    }
+
     pub fn stop(&mut self) {
         drop(self.take_guard());
     }

test/src/rpc.rs

@@ -174,13 +174,24 @@ impl RpcClient {
     }
 
     pub fn wait_rpc_ready(&self) {
+        self.wait_rpc_ready_internal(|| {
+            panic!("wait rpc ready timeout");
+        });
+    }
+
+    pub fn wait_rpc_ready_internal<F>(&self, fail: F) -> bool
+    where
+        F: Fn(),
+    {
         let now = std::time::Instant::now();
         while self.inner.get_tip_block_number().is_err() {
             std::thread::sleep(std::time::Duration::from_millis(100));
             if now.elapsed() > std::time::Duration::from_secs(60) {
-                panic!("wait rpc ready timeout");
+                fail();
+                return false;
             }
         }
+        true
     }
 
     pub fn get_block_template(

test/src/specs/fault_injection/randomly_kill.rs

@@ -11,8 +11,23 @@ impl Spec for RandomlyKill {
     fn run(&self, nodes: &mut Vec<Node>) {
         let mut rng = thread_rng();
         let node = &mut nodes[0];
-        for _ in 0..rng.gen_range(10..20) {
-            node.rpc_client().wait_rpc_ready();
+        let max_restart_times = rng.gen_range(10..20);
+
+        let mut node_crash_times = 0;
+
+        let mut randomly_kill_times = 0;
+        while randomly_kill_times < max_restart_times {
+            node.rpc_client().wait_rpc_ready_internal(|| {});
+
+            if !node.is_alive() {
+                node.start();
+                node_crash_times += 1;
+
+                if node_crash_times > 3 {
+                    panic!("Node crash too many times");
+                }
+            }
+
             let n = rng.gen_range(0..10);
             // TODO: the kill of child process and mining are actually sequential here
             // We need to find some way to so these two things in parallel.
@@ -25,6 +40,7 @@ impl Spec for RandomlyKill {
             }
             info!("Stop the node");
             node.stop_gracefully();
+            randomly_kill_times += 1;
             info!("Start the node");
             node.start();
         }