Repository: nervosnetwork/ckb (language: rust)

Issue:
Performance issue about websocket of CKB node
## Bug Report

<!-- Thanks for your bug report! We appreciate your time to fill following sections to help us locate the root cause. -->

### Current Behavior
Recently we have done a test about fee rate estimation. During the test, numerous transactions(40k+) have been submitted onto the chain and the tx pool is full of pending transactions(30k+) according to the RPC `get_raw_tx_pool`.

Our app subscribes to the CKB node via WebSocket, and pending transactions are expected to be pushed to our end. However, only 3k transactions had been received during the test, and the rest were reached in blocks.

Here's our question, is there a limit on the number of transactions pushed by WebSocket?

### Expected Behavior
Most/all pending transactions should be pushed via WebSocket.

### Environment

- **CKB version**: v0.116.1
- **Chain**: testnet
- **Operating system**: Ubuntu 18.0.6 LTS
- **Arch**: X86_64
- **Installation**: GitHub Release

### Additional context/Screenshots
<!-- Add any other context about the problem here. If applicable, add ckb.toml and screenshots to help explain. -->
The original issue could be found at https://github.com/Magickbase/ckb-explorer-public-issues/issues/704 and a question post could be found at https://talk.nervos.org/t/performance-issue-about-websocket-of-ckb-node/8248

After a brief inquiry, I was led to open a bug report here.



Discussion / hints:
Code: https://github.com/nervosnetwork/ckb-explorer/blob/develop/lib/websocket.rb
How is your CPU performance?

Fix patch:
diff --git a/rpc/src/module/subscription.rs b/rpc/src/module/subscription.rs
index 1c6e846dce..4036445929 100644
--- a/rpc/src/module/subscription.rs
+++ b/rpc/src/module/subscription.rs
@@ -1,7 +1,10 @@
 use async_trait::async_trait;
+use broadcast::error::RecvError;
 use ckb_async_runtime::Handle;
 use ckb_jsonrpc_types::Topic;
+use ckb_logger::error;
 use ckb_notify::NotifyController;
+use ckb_notify::NOTIFY_CHANNEL_SIZE;
 use ckb_stop_handler::new_tokio_exit_rx;
 use futures_util::{stream::BoxStream, Stream};
 use jsonrpc_core::Result;
@@ -211,9 +214,19 @@ impl SubscriptionRpc for SubscriptionRpcImpl {
         };
         let mut rx = tx.subscribe();
         Ok(Box::pin(async_stream::stream! {
-               while let Ok(msg) = rx.recv().await {
-                    yield msg;
-               }
+                loop {
+                    match rx.recv().await {
+                        Ok(msg) => {
+                            yield msg;
+                        }
+                        Err(RecvError::Lagged(cnt)) => {
+                            error!("subscription lagged error: {:?}", cnt);
+                        }
+                        Err(RecvError::Closed) => {
+                            break;
+                        }
+                    }
+                }
         }))
     }
 }
@@ -232,11 +245,11 @@ impl SubscriptionRpcImpl {
         let mut reject_transaction_receiver = handle
             .block_on(notify_controller.subscribe_reject_transaction(SUBSCRIBER_NAME.to_string()));
 
-        let (new_tip_header_sender, _) = broadcast::channel(10);
-        let (new_tip_block_sender, _) = broadcast::channel(10);
-        let (proposed_transaction_sender, _) = broadcast::channel(10);
-        let (new_transaction_sender, _) = broadcast::channel(10);
-        let (new_reject_transaction_sender, _) = broadcast::channel(10);
+        let (new_tip_header_sender, _) = broadcast::channel(NOTIFY_CHANNEL_SIZE);
+        let (new_tip_block_sender, _) = broadcast::channel(NOTIFY_CHANNEL_SIZE);
+        let (proposed_transaction_sender, _) = broadcast::channel(NOTIFY_CHANNEL_SIZE);
+        let (new_transaction_sender, _) = broadcast::channel(NOTIFY_CHANNEL_SIZE);
+        let (new_reject_transaction_sender, _) = broadcast::channel(NOTIFY_CHANNEL_SIZE);
 
         let stop_rx = new_tokio_exit_rx();
         handle.spawn({
@@ -266,7 +279,10 @@ impl SubscriptionRpcImpl {
                         _ = stop_rx.cancelled() => {
                             break;
                         },
-                        else => break,
+                        else => {
+                            error!("SubscriptionRpcImpl tokio::select! unexpected error");
+                            break;
+                        }
                     }
                 }
             }


Test patch: