Remove unneeded jsonrpc dependencies/features; update do-audit (#23436)

* Update generic-array note

* Remove unneeded jsonrpc deps

* Remove unneeded jsonrpc features

* Rewrite slot-update test without websocket crate

* Rewrite rpc-subscription test without websocket crate, and remove jsonrpc deps

* Update expected balance to accommodate rent-exempt minimum transfer amount

* Remove obsolete audit ignores

ci/do-audit.sh

@@ -25,22 +25,10 @@ cargo_audit_ignores=(
 
   # generic-array: arr! macro erases lifetimes
   #
-  # Blocked on libsecp256k1 releasing with upgraded dependencies
+  # Blocked on new spl dependencies on solana-program v1.9
-  # https://github.com/paritytech/libsecp256k1/issues/66
+  # due to curve25519-dalek dependency
   --ignore RUSTSEC-2020-0146
 
-  # hyper: Lenient `hyper` header parsing of `Content-Length` could allow request smuggling
-  #
-  # Blocked on jsonrpc removing dependency on unmaintained `websocket`
-  # https://github.com/paritytech/jsonrpc/issues/605
-  --ignore RUSTSEC-2021-0078
-
-  # hyper: Integer overflow in `hyper`'s parsing of the `Transfer-Encoding` header leads to data loss
-  #
-  # Blocked on jsonrpc removing dependency on unmaintained `websocket`
-  # https://github.com/paritytech/jsonrpc/issues/605
-  --ignore RUSTSEC-2021-0079
-
   # chrono: Potential segfault in `localtime_r` invocations
   #
   # Blocked due to no safe upgrade

core/Cargo.toml

@@ -64,10 +64,6 @@ tokio = { version = "1", features = ["full"] }
 trees = "0.4.2"
 
 [dev-dependencies]
-jsonrpc-core = "18.0.0"
-jsonrpc-core-client = { version = "18.0.0", features = ["ipc", "ws"] }
-jsonrpc-derive = "18.0.0"
-jsonrpc-pubsub = "18.0.0"
 matches = "0.1.9"
 raptorq = "1.6.5"
 reqwest = { version = "0.11.9", default-features = false, features = ["blocking", "rustls-tls", "json"] }

rpc-test/Cargo.toml

@@ -13,8 +13,7 @@ edition = "2021"
 bincode = "1.3.3"
 bs58 = "0.4.0"
 crossbeam-channel = "0.5"
-jsonrpc-core = "18.0.0"
+futures-util = "0.3.21"
-jsonrpc-core-client = { version = "18.0.0", features = ["ipc", "ws"] }
 log = "0.4.11"
 reqwest = { version = "0.11.9", default-features = false, features = ["blocking", "rustls-tls", "json"] }
 serde = "1.0.136"

rpc-test/tests/rpc.rs

@@ -1,27 +1,26 @@
 use {
     bincode::serialize,
     crossbeam_channel::unbounded,
-    jsonrpc_core::futures::StreamExt,
+    futures_util::StreamExt,
-    jsonrpc_core_client::transports::ws,
     log::*,
     reqwest::{self, header::CONTENT_TYPE},
     serde_json::{json, Value},
     solana_account_decoder::UiAccount,
     solana_client::{
         client_error::{ClientErrorKind, Result as ClientResult},
+        nonblocking::pubsub_client::PubsubClient,
         rpc_client::RpcClient,
         rpc_config::{RpcAccountInfoConfig, RpcSignatureSubscribeConfig},
         rpc_request::RpcError,
         rpc_response::{Response as RpcResponse, RpcSignatureResult, SlotUpdate},
         tpu_client::{TpuClient, TpuClientConfig},
     },
-    solana_rpc::rpc_pubsub::gen_client::Client as PubsubClient,
     solana_sdk::{
         commitment_config::CommitmentConfig,
         hash::Hash,
         pubkey::Pubkey,
         rent::Rent,
-        signature::{Keypair, Signer},
+        signature::{Keypair, Signature, Signer},
         system_transaction,
         transaction::Transaction,
     },
@@ -171,23 +170,21 @@ fn test_rpc_slot_updates() {
     let test_validator =
         TestValidator::with_no_fees(Pubkey::new_unique(), None, SocketAddrSpace::Unspecified);
 
+    // Track when slot updates are ready
+    let (update_sender, update_receiver) = unbounded::<SlotUpdate>();
     // Create the pub sub runtime
     let rt = Runtime::new().unwrap();
     let rpc_pubsub_url = test_validator.rpc_pubsub_url();
-    let (update_sender, update_receiver) = unbounded::<Arc<SlotUpdate>>();
 
-    // Subscribe to slot updates
     rt.spawn(async move {
-        let connect = ws::try_connect::<PubsubClient>(&rpc_pubsub_url).unwrap();
+        let pubsub_client = PubsubClient::new(&rpc_pubsub_url).await.unwrap();
-        let client = connect.await.unwrap();
+        let (mut slot_notifications, slot_unsubscribe) =
+            pubsub_client.slot_updates_subscribe().await.unwrap();
 
-        tokio::spawn(async move {
+        while let Some(slot_update) = slot_notifications.next().await {
-            let mut update_sub = client.slots_updates_subscribe().unwrap();
+            update_sender.send(slot_update).unwrap();
-            loop {
+        }
-                let response = update_sub.next().await.unwrap();
+        slot_unsubscribe().await;
-                update_sender.send(response.unwrap()).unwrap();
-            }
-        });
     });
 
     let first_update = update_receiver
@@ -212,7 +209,7 @@ fn test_rpc_slot_updates() {
             .recv_timeout(Duration::from_secs(2))
             .unwrap();
         if update.slot() == verify_slot {
-            let update_name = match *update {
+            let update_name = match update {
                 SlotUpdate::CreatedBank { .. } => "CreatedBank",
                 SlotUpdate::Completed { .. } => "Completed",
                 SlotUpdate::Frozen { .. } => "Frozen",
@@ -244,23 +241,22 @@ fn test_rpc_subscriptions() {
     let recent_blockhash = rpc_client.get_latest_blockhash().unwrap();
 
     // Create transaction signatures to subscribe to
+    let transfer_amount = Rent::default().minimum_balance(0);
     let transactions: Vec<Transaction> = (0..1000)
         .map(|_| {
             system_transaction::transfer(
                 &alice,
                 &solana_sdk::pubkey::new_rand(),
-                Rent::default().minimum_balance(0),
+                transfer_amount,
                 recent_blockhash,
             )
         })
         .collect();
-    let mut signature_set: HashSet<String> = transactions
+    let mut signature_set: HashSet<Signature> =
+        transactions.iter().map(|tx| tx.signatures[0]).collect();
+    let account_set: HashSet<Pubkey> = transactions
         .iter()
-        .map(|tx| tx.signatures[0].to_string())
+        .map(|tx| tx.message.account_keys[1])
-        .collect();
-    let account_set: HashSet<String> = transactions
-        .iter()
-        .map(|tx| tx.message.account_keys[1].to_string())
         .collect();
 
     // Track when subscriptions are ready
@@ -268,62 +264,74 @@ fn test_rpc_subscriptions() {
     // Track account notifications are received
     let (account_sender, account_receiver) = unbounded::<RpcResponse<UiAccount>>();
     // Track when status notifications are received
-    let (status_sender, status_receiver) = unbounded::<(String, RpcResponse<RpcSignatureResult>)>();
+    let (status_sender, status_receiver) =
+        unbounded::<(Signature, RpcResponse<RpcSignatureResult>)>();
 
     // Create the pub sub runtime
     let rt = Runtime::new().unwrap();
     let rpc_pubsub_url = test_validator.rpc_pubsub_url();
     let signature_set_clone = signature_set.clone();
     rt.spawn(async move {
-        let connect = ws::try_connect::<PubsubClient>(&rpc_pubsub_url).unwrap();
+        let pubsub_client = Arc::new(PubsubClient::new(&rpc_pubsub_url).await.unwrap());
-        let client = connect.await.unwrap();
 
         // Subscribe to signature notifications
-        for sig in signature_set_clone {
+        for signature in signature_set_clone {
             let status_sender = status_sender.clone();
-            let mut sig_sub = client
+            tokio::spawn({
-                .signature_subscribe(
+                let _pubsub_client = Arc::clone(&pubsub_client);
-                    sig.clone(),
+                async move {
-                    Some(RpcSignatureSubscribeConfig {
+                    let (mut sig_notifications, sig_unsubscribe) = _pubsub_client
-                        commitment: Some(CommitmentConfig::confirmed()),
+                        .signature_subscribe(
-                        ..RpcSignatureSubscribeConfig::default()
+                            &signature,
-                    }),
+                            Some(RpcSignatureSubscribeConfig {
-                )
+                                commitment: Some(CommitmentConfig::confirmed()),
-                .unwrap_or_else(|err| panic!("sig sub err: {:#?}", err));
+                                ..RpcSignatureSubscribeConfig::default()
+                            }),
+                        )
+                        .await
+                        .unwrap();
 
-            tokio::spawn(async move {
+                    let response = sig_notifications.next().await.unwrap();
-                let response = sig_sub.next().await.unwrap();
+                    status_sender.send((signature, response)).unwrap();
-                status_sender
+                    sig_unsubscribe().await;
-                    .send((sig.clone(), response.unwrap()))
+                }
-                    .unwrap();
             });
         }
 
         // Subscribe to account notifications
         for pubkey in account_set {
             let account_sender = account_sender.clone();
-            let mut client_sub = client
+            tokio::spawn({
-                .account_subscribe(
+                let _pubsub_client = Arc::clone(&pubsub_client);
-                    pubkey,
+                async move {
-                    Some(RpcAccountInfoConfig {
+                    let (mut account_notifications, account_unsubscribe) = _pubsub_client
-                        commitment: Some(CommitmentConfig::confirmed()),
+                        .account_subscribe(
-                        ..RpcAccountInfoConfig::default()
+                            &pubkey,
-                    }),
+                            Some(RpcAccountInfoConfig {
-                )
+                                commitment: Some(CommitmentConfig::confirmed()),
-                .unwrap_or_else(|err| panic!("acct sub err: {:#?}", err));
+                                ..RpcAccountInfoConfig::default()
-            tokio::spawn(async move {
+                            }),
-                let response = client_sub.next().await.unwrap();
+                        )
-                account_sender.send(response.unwrap()).unwrap();
+                        .await
+                        .unwrap();
+
+                    let response = account_notifications.next().await.unwrap();
+                    account_sender.send(response).unwrap();
+                    account_unsubscribe().await;
+                }
             });
         }
 
         // Signal ready after the next slot notification
-        let mut slot_sub = client
+        tokio::spawn({
-            .slot_subscribe()
+            let _pubsub_client = Arc::clone(&pubsub_client);
-            .unwrap_or_else(|err| panic!("sig sub err: {:#?}", err));
+            async move {
-        tokio::spawn(async move {
+                let (mut slot_notifications, slot_unsubscribe) =
-            let _response = slot_sub.next().await.unwrap();
+                    _pubsub_client.slot_subscribe().await.unwrap();
-            ready_sender.send(()).unwrap();
+                let _response = slot_notifications.next().await.unwrap();
+                ready_sender.send(()).unwrap();
+                slot_unsubscribe().await;
+            }
         });
     });
 
@@ -346,7 +354,7 @@ fn test_rpc_subscriptions() {
 
     // Track mint balance to know when transactions have completed
     let now = Instant::now();
-    let expected_mint_balance = mint_balance - transactions.len() as u64;
+    let expected_mint_balance = mint_balance - (transfer_amount * transactions.len() as u64);
     while mint_balance != expected_mint_balance && now.elapsed() < Duration::from_secs(15) {
         mint_balance = rpc_client
             .get_balance_with_commitment(&alice.pubkey(), CommitmentConfig::processed())

rpc/Cargo.toml

@@ -17,7 +17,7 @@ crossbeam-channel = "0.5"
 dashmap = "4.0.2"
 itertools = "0.10.3"
 jsonrpc-core = "18.0.0"
-jsonrpc-core-client = { version = "18.0.0", features = ["ipc", "ws"] }
+jsonrpc-core-client = { version = "18.0.0" }
 jsonrpc-derive = "18.0.0"
 jsonrpc-http-server = "18.0.0"
 jsonrpc-pubsub = "18.0.0"

validator/Cargo.toml

@@ -19,7 +19,7 @@ crossbeam-channel = "0.5"
 fd-lock = "3.0.4"
 indicatif = "0.16.2"
 jsonrpc-core = "18.0.0"
-jsonrpc-core-client = { version = "18.0.0", features = ["ipc", "ws"] }
+jsonrpc-core-client = { version = "18.0.0", features = ["ipc"] }
 jsonrpc-derive = "18.0.0"
 jsonrpc-ipc-server = "18.0.0"
 jsonrpc-server-utils= "18.0.0"