Enable cluster tests (#3372)

* Cluster tests

* stable!

* fixup! stable!

* fixup! fixup! stable!

* fixup! fixup! fixup! stable!

* fixup! fixup! fixup! fixup! stable!

* fixed space

* add getNumBlocksSinceSignatureConfirmation entry for the json rpc docs

* Check in upcoming epochs for potential leadership slots in next_leader_slot()

client/src/rpc_client.rs

@@ -438,6 +438,78 @@ impl RpcClient {
             };
         }
     }
+
+    /// Poll the server to confirm a transaction.
+    pub fn poll_for_signature_confirmation(
+        &self,
+        signature: &Signature,
+        min_confirmed_blocks: usize,
+    ) -> io::Result<()> {
+        let mut now = Instant::now();
+        let mut confirmed_blocks = 0;
+        loop {
+            let response = self.get_num_blocks_since_signature_confirmation(signature);
+            match response {
+                Ok(count) => {
+                    if confirmed_blocks != count {
+                        info!(
+                            "signature {} confirmed {} out of {}",
+                            signature, count, min_confirmed_blocks
+                        );
+                        now = Instant::now();
+                        confirmed_blocks = count;
+                    }
+                    if count >= min_confirmed_blocks {
+                        break;
+                    }
+                }
+                Err(err) => {
+                    debug!("check_confirmations request failed: {:?}", err);
+                }
+            };
+            if now.elapsed().as_secs() > 15 {
+                // TODO: Return a better error.
+                return Err(io::Error::new(io::ErrorKind::Other, "signature not found"));
+            }
+            sleep(Duration::from_millis(250));
+        }
+        Ok(())
+    }
+
+    pub fn get_num_blocks_since_signature_confirmation(
+        &self,
+        sig: &Signature,
+    ) -> io::Result<usize> {
+        let params = json!([format!("{}", sig)]);
+        let response = self
+            .client
+            .send(
+                &RpcRequest::GetNumBlocksSinceSignatureConfirmation,
+                Some(params.clone()),
+                1,
+            )
+            .map_err(|error| {
+                debug!(
+                    "Response get_num_blocks_since_signature_confirmation: {}",
+                    error
+                );
+                io::Error::new(
+                    io::ErrorKind::Other,
+                    "GetNumBlocksSinceSignatureConfirmation request failure",
+                )
+            })?;
+        serde_json::from_value(response).map_err(|error| {
+            debug!(
+                "ParseError: get_num_blocks_since_signature_confirmation: {}",
+                error
+            );
+            io::Error::new(
+                io::ErrorKind::Other,
+                "GetNumBlocksSinceSignatureConfirmation parse failure",
+            )
+        })
+    }
+
     pub fn fullnode_exit(&self) -> io::Result<bool> {
         let response = self
             .client

client/src/rpc_request.rs

@@ -18,6 +18,7 @@ pub enum RpcRequest {
     GetStorageEntryHeight,
     GetStoragePubkeysForEntryHeight,
     FullnodeExit,
+    GetNumBlocksSinceSignatureConfirmation,
 }
 
 impl RpcRequest {
@@ -39,6 +40,9 @@ impl RpcRequest {
             RpcRequest::GetStorageEntryHeight => "getStorageEntryHeight",
             RpcRequest::GetStoragePubkeysForEntryHeight => "getStoragePubkeysForEntryHeight",
             RpcRequest::FullnodeExit => "fullnodeExit",
+            RpcRequest::GetNumBlocksSinceSignatureConfirmation => {
+                "getNumBlocksSinceSignatureConfirmation"
+            }
         };
         let mut request = json!({
            "jsonrpc": jsonrpc,

client/src/thin_client.rs

@@ -73,6 +73,36 @@ impl ThinClient {
         Ok(transaction.signatures[0])
     }
 
+    /// Retry a sending a signed Transaction to the server for processing.
+    pub fn retry_transfer_until_confirmed(
+        &self,
+        keypair: &Keypair,
+        transaction: &mut Transaction,
+        tries: usize,
+        min_confirmed_blocks: usize,
+    ) -> io::Result<Signature> {
+        for x in 0..tries {
+            transaction.sign(&[keypair], self.get_recent_blockhash()?);
+            let mut buf = vec![0; transaction.serialized_size().unwrap() as usize];
+            let mut wr = std::io::Cursor::new(&mut buf[..]);
+            serialize_into(&mut wr, &transaction)
+                .expect("serialize Transaction in pub fn transfer_signed");
+            self.transactions_socket
+                .send_to(&buf[..], &self.transactions_addr)?;
+            if self
+                .poll_for_signature_confirmation(&transaction.signatures[0], min_confirmed_blocks)
+                .is_ok()
+            {
+                return Ok(transaction.signatures[0]);
+            }
+            info!("{} tries failed transfer to {}", x, self.transactions_addr);
+        }
+        Err(io::Error::new(
+            io::ErrorKind::Other,
+            "retry_transfer failed",
+        ))
+    }
+
     /// Retry a sending a signed Transaction to the server for processing.
     pub fn retry_transfer(
         &self,
@@ -140,7 +170,18 @@ impl ThinClient {
     pub fn poll_for_signature(&self, signature: &Signature) -> io::Result<()> {
         self.rpc_client.poll_for_signature(signature)
     }
+    /// Poll the server until the signature has been confirmed by at least `min_confirmed_blocks`
+    pub fn poll_for_signature_confirmation(
+        &self,
+        signature: &Signature,
+        min_confirmed_blocks: usize,
+    ) -> io::Result<()> {
+        self.rpc_client
+            .poll_for_signature_confirmation(signature, min_confirmed_blocks)
+    }
 
+    /// Check a signature in the bank. This method blocks
+    /// until the server sends a response.
     pub fn check_signature(&self, signature: &Signature) -> bool {
         self.rpc_client.check_signature(signature)
     }
@@ -148,6 +189,13 @@ impl ThinClient {
     pub fn fullnode_exit(&self) -> io::Result<bool> {
         self.rpc_client.fullnode_exit()
     }
+    pub fn get_num_blocks_since_signature_confirmation(
+        &mut self,
+        sig: &Signature,
+    ) -> io::Result<usize> {
+        self.rpc_client
+            .get_num_blocks_since_signature_confirmation(sig)
+    }
 }
 
 pub fn create_client((rpc, tpu): (SocketAddr, SocketAddr), range: (u16, u16)) -> ThinClient {