Merge branch 'develop' into rc/v0.203.x

Author: doitian

Cargo.lock

@@ -308,7 +308,7 @@ url = "2.2.2"
 walkdir = "2.1.4"
 yansi = "0.5"
 seq-macro = "0.3"
-strum = { version = "0.27", features = ["derive"] }
+strum = { version = "0.27", default-features = false, features = ["derive"] }
 phf = { version = "0.12", features = ["macros"] }
 
 [profile.release]

Cargo.toml

@@ -49,7 +49,7 @@ ckb-stop-handler.workspace = true
 tokio = { workspace = true, features = ["sync"] }
 
 [target.'cfg(not(target_os="windows"))'.dependencies]
-daemonize = { version = "0.5.0" }
+daemonize-me = { version = "2" }
 nix = { version = "0.29.0", default-features = false, features = ["signal"] }
 colored = "2.0"
 

ckb-bin/Cargo.toml

@@ -22,7 +22,7 @@ use setup_guard::SetupGuard;
 #[cfg(not(target_os = "windows"))]
 use colored::Colorize;
 #[cfg(not(target_os = "windows"))]
-use daemonize::Daemonize;
+use daemonize_me::Daemon;
 #[cfg(not(target_os = "windows"))]
 use subcommand::check_process;
 #[cfg(feature = "with_sentry")]
@@ -105,10 +105,7 @@ fn run_app_in_daemon(
     eprintln!("no ckb process, starting ...");
 
     let pwd = std::env::current_dir()?;
-    let daemon = Daemonize::new()
-        .pid_file(pid_file)
-        .chown_pid_file(true)
-        .working_directory(pwd);
+    let daemon = Daemon::new().pid_file(pid_file, Some(false)).work_dir(pwd);
 
     match daemon.start() {
         Ok(_) => {

ckb-bin/src/lib.rs

@@ -14,7 +14,6 @@ retry_cargo_publish() {
     -e '/^\[dev-dependencies\]/, /^\[/ { /^[^\[]/d }' \
     -e '/# dev-feature$/d' \
     -e '/^ckb-test\.workspace/d' \
-    -e '/^ckb-test\.workspace/d' \
     -e '/^ckb-test = .*version/d' \
     Cargo.toml
 

devtools/release/publish-crates.sh

@@ -20,7 +20,7 @@ use self::{
     protocol::{decode, encode},
     state::RemoteAddress,
 };
-use crate::{Flags, NetworkState, ProtocolId, SupportProtocols};
+use crate::{Flags, NetworkState, ProtocolId};
 
 mod addr;
 pub(crate) mod protocol;
@@ -79,17 +79,6 @@ impl<M: AddressManager + Send + Sync> ServiceProtocol for DiscoveryProtocol<M> {
         {
             self.sessions
                 .insert(session.id, SessionState::new(context, &self.addr_mgr).await);
-        } else if let Err(e) = context
-            .close_protocol(
-                context.session.id,
-                SupportProtocols::Discovery.protocol_id(),
-            )
-            .await
-        {
-            error!(
-                "Block-Relay-Only Session close DiscoveryProtocol failed, session: {:?} {}",
-                context.session, e
-            );
         }
     }
 

network/src/protocols/discovery/mod.rs

@@ -206,13 +206,12 @@ impl<T: Callback> ServiceProtocol for IdentifyProtocol<T> {
     async fn connected(&mut self, context: ProtocolContextMutRef<'_>, version: &str) {
         let session = context.session;
         debug!("IdentifyProtocol connected, session: {:?}", session);
-        if self.callback.register(&context, version) {
-            let remote_info =
-                RemoteInfo::new(session.clone(), Duration::from_secs(DEFAULT_TIMEOUT));
-            self.remote_infos.insert(session.id, remote_info);
-
-            let listen_addrs: Vec<Multiaddr> = self
-                .callback
+        let remote_info = RemoteInfo::new(session.clone(), Duration::from_secs(DEFAULT_TIMEOUT));
+        self.remote_infos.insert(session.id, remote_info);
+        let listen_addrs = if self.callback.register(&context, version) {
+            Vec::new()
+        } else {
+            self.callback
                 .local_listen_addrs()
                 .iter()
                 .filter(|addr| {
@@ -222,24 +221,15 @@ impl<T: Callback> ServiceProtocol for IdentifyProtocol<T> {
                 })
                 .take(MAX_ADDRS)
                 .cloned()
-                .collect();
-
-            let identify = self.callback.identify();
-            let data =
-                IdentifyMessage::new(listen_addrs, session.address.clone(), identify).encode();
-            let _ = context
-                .quick_send_message(data)
-                .await
-                .map_err(|err| error!("IdentifyProtocol quick_send_message, error: {:?}", err));
-        } else if let Err(e) = context
-            .close_protocol(context.session.id, SupportProtocols::Identify.protocol_id())
+                .collect()
+        };
+
+        let identify = self.callback.identify();
+        let data = IdentifyMessage::new(listen_addrs, session.address.clone(), identify).encode();
+        let _ = context
+            .quick_send_message(data)
             .await
-        {
-            error!(
-                "Block-Relay-Only Session close IdentifyProtocol failed, session: {:?} {}",
-                context.session, e
-            );
-        }
+            .map_err(|err| error!("IdentifyProtocol quick_send_message, error: {:?}", err));
     }
 
     async fn disconnected(&mut self, context: ProtocolContextMutRef<'_>) {
@@ -363,17 +353,12 @@ impl IdentifyCallback {
 impl Callback for IdentifyCallback {
     fn register(&self, context: &ProtocolContextMutRef, version: &str) -> bool {
         let session_id = context.session.id;
-        let is_anchor = self
-            .network_state
-            .with_peer_registry(|reg| reg.is_anchor(session_id));
-        if !is_anchor {
-            self.network_state.with_peer_registry_mut(|reg| {
-                reg.get_peer_mut(session_id).map(|peer| {
-                    peer.protocols.insert(context.proto_id, version.to_owned());
-                })
-            });
-        }
-        !is_anchor
+        self.network_state.with_peer_registry_mut(|reg| {
+            if let Some(peer) = reg.get_peer_mut(session_id) {
+                peer.protocols.insert(context.proto_id, version.to_owned());
+            }
+            reg.is_anchor(session_id)
+        })
     }
 
     fn unregister(&self, context: &ProtocolContextMutRef) {

network/src/protocols/identify/mod.rs

@@ -38,9 +38,22 @@ impl<'a> GetBlockFiltersProcess<'a> {
         if latest >= start_number {
             let mut block_hashes = Vec::new();
             let mut filters = Vec::new();
+            let mut current_content_size = 0;
+            current_content_size += 8; // Size of start_number
+
             for block_number in start_number..start_number + BATCH_SIZE {
                 if let Some(block_hash) = active_chain.get_block_hash(block_number) {
                     if let Some(block_filter) = active_chain.get_block_filter(&block_hash) {
+                        if current_content_size
+                            + block_hash.as_slice().len()
+                            + block_filter.as_slice().len()
+                            >= 2 * 1024 * 1024
+                        {
+                            // Break if the encoded size of `block_hash` + `block_filter` + `start_number` reaches 2MB, to avoid frame size too large
+                            break;
+                        }
+                        current_content_size +=
+                            block_hash.as_slice().len() + block_filter.as_slice().len();
                         block_hashes.push(block_hash);
                         filters.push(block_filter);
                     } else {

sync/src/filter/get_block_filters_process.rs

@@ -138,10 +138,10 @@ impl IndexerSyncService {
     where
         I: IndexerSync + Clone + Send + 'static,
     {
-        if let Err(e) = self.secondary_db.try_catch_up_with_primary() {
-            error!("secondary_db try_catch_up_with_primary error {}", e);
-        }
         loop {
+            if let Err(e) = self.secondary_db.try_catch_up_with_primary() {
+                error!("secondary_db try_catch_up_with_primary error {}", e);
+            }
             if has_received_stop_signal() {
                 info!("try_loop_sync received exit signal, exit now");
                 break;