Respect confirmations param for signature subscription notifications (#9019)

automerge

core/tests/rpc.rs

@@ -7,9 +7,18 @@ use jsonrpc_core_client::transports::ws;
 use log::*;
 use reqwest::{self, header::CONTENT_TYPE};
 use serde_json::{json, Value};
-use solana_client::{rpc_client::get_rpc_request_str, rpc_response::Response};
+use solana_client::{
+    rpc_client::{get_rpc_request_str, RpcClient},
+    rpc_response::Response,
+};
 use solana_core::{rpc_pubsub::gen_client::Client as PubsubClient, validator::TestValidator};
-use solana_sdk::{hash::Hash, pubkey::Pubkey, system_transaction, transaction};
+use solana_sdk::{
+    commitment_config::CommitmentConfig,
+    hash::Hash,
+    pubkey::Pubkey,
+    system_transaction,
+    transaction::{self, Transaction},
+};
 use std::{
     collections::HashSet,
     fs::remove_dir_all,
@@ -17,8 +26,7 @@ use std::{
     sync::mpsc::channel,
     sync::{Arc, Mutex},
     thread::sleep,
-    time::Duration,
-    time::SystemTime,
+    time::{Duration, Instant},
 };
 use tokio::runtime::Runtime;
 
@@ -204,35 +212,43 @@ fn test_rpc_subscriptions() {
 
     // Create transaction signatures to subscribe to
     let transactions_socket = UdpSocket::bind("0.0.0.0:0").unwrap();
-    let mut signature_set: HashSet<String> = (0..1000)
-        .map(|_| {
-            let tx = system_transaction::transfer(&alice, &Pubkey::new_rand(), 1, genesis_hash);
-            transactions_socket
-                .send_to(&bincode::serialize(&tx).unwrap(), leader_data.tpu)
-                .unwrap();
-            tx.signatures[0].to_string()
-        })
+    let transactions: Vec<Transaction> = (0..500)
+        .map(|_| system_transaction::transfer(&alice, &Pubkey::new_rand(), 1, genesis_hash))
+        .collect();
+    let mut signature_set: HashSet<String> = transactions
+        .iter()
+        .map(|tx| tx.signatures[0].to_string())
         .collect();
 
     // Create the pub sub runtime
     let mut rt = Runtime::new().unwrap();
     let rpc_pubsub_url = format!("ws://{}/", leader_data.rpc_pubsub);
-    let (sender, receiver) = channel::<(String, Response<transaction::Result<()>>)>();
-    let sender = Arc::new(Mutex::new(sender));
 
+    let (status_sender, status_receiver) = channel::<(String, Response<transaction::Result<()>>)>();
+    let status_sender = Arc::new(Mutex::new(status_sender));
+    let (sent_sender, sent_receiver) = channel::<()>();
+    let sent_sender = Arc::new(Mutex::new(sent_sender));
+
+    // Subscribe to all signatures
     rt.spawn({
         let connect = ws::try_connect::<PubsubClient>(&rpc_pubsub_url).unwrap();
         let signature_set = signature_set.clone();
         connect
             .and_then(move |client| {
                 for sig in signature_set {
-                    let sender = sender.clone();
+                    let status_sender = status_sender.clone();
+                    let sent_sender = sent_sender.clone();
                     tokio::spawn(
                         client
                             .signature_subscribe(sig.clone(), None)
                             .and_then(move |sig_stream| {
+                                sent_sender.lock().unwrap().send(()).unwrap();
                                 sig_stream.for_each(move |result| {
-                                    sender.lock().unwrap().send((sig.clone(), result)).unwrap();
+                                    status_sender
+                                        .lock()
+                                        .unwrap()
+                                        .send((sig.clone(), result))
+                                        .unwrap();
                                     future::ok(())
                                 })
                             })
@@ -246,18 +262,49 @@ fn test_rpc_subscriptions() {
             .map_err(|_| ())
     });
 
+    // Wait for signature subscriptions
+    let deadline = Instant::now() + Duration::from_secs(2);
+    (0..transactions.len()).for_each(|_| {
+        sent_receiver
+            .recv_timeout(deadline.saturating_duration_since(Instant::now()))
+            .unwrap();
+    });
+
+    let rpc_client = RpcClient::new_socket(leader_data.rpc);
+    let transaction_count = rpc_client
+        .get_transaction_count_with_commitment(CommitmentConfig::recent())
+        .unwrap();
+
+    // Send all transactions to tpu socket for processing
+    transactions.iter().for_each(|tx| {
+        transactions_socket
+            .send_to(&bincode::serialize(&tx).unwrap(), leader_data.tpu)
+            .unwrap();
+    });
+    let mut x = 0;
+    let now = Instant::now();
+    while x < transaction_count + 500 || now.elapsed() > Duration::from_secs(5) {
+        x = rpc_client
+            .get_transaction_count_with_commitment(CommitmentConfig::recent())
+            .unwrap();
+        sleep(Duration::from_millis(200));
+    }
+
     // Wait for all signature subscriptions
-    let now = SystemTime::now();
-    let timeout = Duration::from_secs(5);
+    let deadline = Instant::now() + Duration::from_secs(5);
     while !signature_set.is_empty() {
-        assert!(now.elapsed().unwrap() < timeout);
-        match receiver.recv_timeout(Duration::from_secs(1)) {
+        let timeout = deadline.saturating_duration_since(Instant::now());
+        match status_receiver.recv_timeout(timeout) {
             Ok((sig, result)) => {
                 assert!(result.value.is_ok());
                 assert!(signature_set.remove(&sig));
             }
             Err(_err) => {
-                eprintln!("unexpected receive timeout");
+                eprintln!(
+                    "recv_timeout, {}/{} signatures remaining",
+                    signature_set.len(),
+                    transactions.len()
+                );
                 assert!(false)
             }
         }