From a3a47de3c09797a25b776e08c535ccfaf6707a1c Mon Sep 17 00:00:00 2001 From: Tyera Eulberg Date: Wed, 22 Aug 2018 11:53:24 -0600 Subject: [PATCH 1/4] Add json-rpc sendTransaction endpoint --- src/fullnode.rs | 14 ++++++++++++-- src/rpc.rs | 49 ++++++++++++++++++++++++++----------------------- 2 files changed, 38 insertions(+), 25 deletions(-) diff --git a/src/fullnode.rs b/src/fullnode.rs index b8fb60b6d22259..3d2d4d21fbedbb 100644 --- a/src/fullnode.rs +++ b/src/fullnode.rs @@ -203,7 +203,12 @@ impl Fullnode { thread_hdls.extend(rpu.thread_hdls()); let rpc_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), RPC_PORT); - let rpc_service = JsonRpcService::new(bank.clone(), rpc_addr, exit.clone()); + let rpc_service = JsonRpcService::new( + bank.clone(), + node.data.contact_info.tpu, + rpc_addr, + exit.clone(), + ); thread_hdls.extend(rpc_service.thread_hdls()); let blob_recycler = BlobRecycler::default(); @@ -298,7 +303,12 @@ impl Fullnode { thread_hdls.extend(rpu.thread_hdls()); let rpc_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), RPC_PORT); - let rpc_service = JsonRpcService::new(bank.clone(), rpc_addr, exit.clone()); + let rpc_service = JsonRpcService::new( + bank.clone(), + node.data.contact_info.tpu, + rpc_addr, + exit.clone(), + ); thread_hdls.extend(rpc_service.thread_hdls()); let blob_recycler = BlobRecycler::default(); diff --git a/src/rpc.rs b/src/rpc.rs index b06a080f066e5e..79c5b7861d90ad 100644 --- a/src/rpc.rs +++ b/src/rpc.rs @@ -1,16 +1,19 @@ //! The `rpc` module implements the Solana RPC interface. use bank::Bank; +use bincode::deserialize; use bs58; use jsonrpc_core::*; use jsonrpc_http_server::*; use service::Service; use signature::{Pubkey, Signature}; use std::mem; -use std::net::SocketAddr; +use std::net::{SocketAddr, UdpSocket}; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use std::thread::{self, Builder, JoinHandle}; +use transaction::Transaction; +use wallet::request_airdrop; pub const RPC_PORT: u16 = 8899; @@ -19,7 +22,12 @@ pub struct JsonRpcService { } impl JsonRpcService { - pub fn new(bank: Arc, rpc_addr: SocketAddr, exit: Arc) -> Self { + pub fn new( + bank: Arc, + transactions_addr: SocketAddr, + rpc_addr: SocketAddr, + exit: Arc, + ) -> Self { let request_processor = JsonRpcRequestProcessor::new(bank); let thread_hdl = Builder::new() .name("solana-jsonrpc".to_string()) @@ -31,6 +39,7 @@ impl JsonRpcService { let server = ServerBuilder::with_meta_extractor(io, move |_req: &hyper::Request| Meta { request_processor: request_processor.clone(), + transactions_addr, }).threads(4) .cors(DomainsValidation::AllowOnly(vec![ AccessControlAllowOrigin::Any, @@ -66,6 +75,7 @@ impl Service for JsonRpcService { #[derive(Clone)] pub struct Meta { pub request_processor: JsonRpcRequestProcessor, + pub transactions_addr: SocketAddr, } impl Metadata for Meta {} @@ -88,8 +98,8 @@ build_rpc_trait! { #[rpc(meta, name = "getTransactionCount")] fn get_transaction_count(&self, Self::Metadata) -> Result; - // #[rpc(meta, name = "sendTransaction")] - // fn send_transaction(&self, Self::Metadata, String, i64) -> Result; + #[rpc(meta, name = "sendTransaction")] + fn send_transaction(&self, Self::Metadata, Vec) -> Result; } } @@ -126,24 +136,14 @@ impl RpcSol for RpcSolImpl { fn get_transaction_count(&self, meta: Self::Metadata) -> Result { meta.request_processor.get_transaction_count() } - // fn send_transaction(&self, meta: Self::Metadata, to: String, tokens: i64) -> Result { - // let client_keypair = read_keypair(&meta.keypair_location.unwrap()).unwrap(); - // let mut client = mk_client(&meta.leader.unwrap()); - // let last_id = client.get_last_id(); - // let to_pubkey_vec = bs58::decode(to) - // .into_vec() - // .expect("base58-encoded public key"); - // - // if to_pubkey_vec.len() != mem::size_of::() { - // Err(Error::invalid_request()) - // } else { - // let to_pubkey = Pubkey::new(&to_pubkey_vec); - // let signature = client - // .transfer(tokens, &client_keypair, to_pubkey, &last_id) - // .unwrap(); - // Ok(bs58::encode(signature).into_string()) - // } - // } + fn send_transaction(&self, meta: Self::Metadata, data: Vec) -> Result { + let tx: Transaction = deserialize(&data).map_err(|_| Error::invalid_request())?; + let transactions_socket = UdpSocket::bind("0.0.0.0:0").unwrap(); + transactions_socket + .send_to(&data, &meta.transactions_addr) + .map_err(|_| Error::internal_error())?; + Ok(bs58::encode(tx.signature).into_string()) + } } #[derive(Clone)] pub struct JsonRpcRequestProcessor { @@ -196,11 +196,12 @@ mod tests { bank.process_transaction(&tx).expect("process transaction"); let request_processor = JsonRpcRequestProcessor::new(Arc::new(bank)); + let transactions_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), 0); let mut io = MetaIoHandler::default(); let rpc = RpcSolImpl; io.extend_with(rpc.to_delegate()); - let meta = Meta { request_processor }; + let meta = Meta { request_processor, transactions_addr }; let req = format!( r#"{{"jsonrpc":"2.0","id":1,"method":"getBalance","params":["{}"]}}"#, @@ -236,6 +237,7 @@ mod tests { let req = r#"{"jsonrpc":"2.0","id":1,"method":"confirmTransaction","params":[1234567890]}"#; let meta = Meta { request_processor: JsonRpcRequestProcessor::new(Arc::new(bank)), + transactions_addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), 0), }; let res = io.handle_request_sync(req, meta); @@ -259,6 +261,7 @@ mod tests { r#"{"jsonrpc":"2.0","id":1,"method":"confirmTransaction","params":["a1b2c3d4e5"]}"#; let meta = Meta { request_processor: JsonRpcRequestProcessor::new(Arc::new(bank)), + transactions_addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), 0), }; let res = io.handle_request_sync(req, meta); From eecda441d1a4d3f5a164ae65201bf0d21810380f Mon Sep 17 00:00:00 2001 From: Tyera Eulberg Date: Wed, 22 Aug 2018 11:54:37 -0600 Subject: [PATCH 2/4] Add json-rpc requestAirdrop endpoint --- src/fullnode.rs | 7 +++++++ src/rpc.rs | 28 +++++++++++++++++++++++++++- 2 files changed, 34 insertions(+), 1 deletion(-) diff --git a/src/fullnode.rs b/src/fullnode.rs index 3d2d4d21fbedbb..2bf97640f73b7d 100644 --- a/src/fullnode.rs +++ b/src/fullnode.rs @@ -3,6 +3,7 @@ use bank::Bank; use broadcast_stage::BroadcastStage; use crdt::{Crdt, NodeInfo, TestNode}; +use drone::DRONE_PORT; use entry::Entry; use ledger::read_ledger; use ncp::Ncp; @@ -202,10 +203,13 @@ impl Fullnode { ); thread_hdls.extend(rpu.thread_hdls()); + let mut drone_addr = node.data.contact_info.tpu; + drone_addr.set_port(DRONE_PORT); let rpc_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), RPC_PORT); let rpc_service = JsonRpcService::new( bank.clone(), node.data.contact_info.tpu, + drone_addr, rpc_addr, exit.clone(), ); @@ -302,10 +306,13 @@ impl Fullnode { ); thread_hdls.extend(rpu.thread_hdls()); + let mut drone_addr = entry_point.contact_info.ncp; + drone_addr.set_port(DRONE_PORT); let rpc_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), RPC_PORT); let rpc_service = JsonRpcService::new( bank.clone(), node.data.contact_info.tpu, + drone_addr, rpc_addr, exit.clone(), ); diff --git a/src/rpc.rs b/src/rpc.rs index 79c5b7861d90ad..347b7cd8be5ee3 100644 --- a/src/rpc.rs +++ b/src/rpc.rs @@ -25,6 +25,7 @@ impl JsonRpcService { pub fn new( bank: Arc, transactions_addr: SocketAddr, + drone_addr: SocketAddr, rpc_addr: SocketAddr, exit: Arc, ) -> Self { @@ -40,6 +41,7 @@ impl JsonRpcService { ServerBuilder::with_meta_extractor(io, move |_req: &hyper::Request| Meta { request_processor: request_processor.clone(), transactions_addr, + drone_addr, }).threads(4) .cors(DomainsValidation::AllowOnly(vec![ AccessControlAllowOrigin::Any, @@ -76,6 +78,7 @@ impl Service for JsonRpcService { pub struct Meta { pub request_processor: JsonRpcRequestProcessor, pub transactions_addr: SocketAddr, + pub drone_addr: SocketAddr, } impl Metadata for Meta {} @@ -98,6 +101,9 @@ build_rpc_trait! { #[rpc(meta, name = "getTransactionCount")] fn get_transaction_count(&self, Self::Metadata) -> Result; + #[rpc(meta, name= "requestAirdrop")] + fn request_airdrop(&self, Self::Metadata, String, u64) -> Result; + #[rpc(meta, name = "sendTransaction")] fn send_transaction(&self, Self::Metadata, Vec) -> Result; } @@ -136,6 +142,18 @@ impl RpcSol for RpcSolImpl { fn get_transaction_count(&self, meta: Self::Metadata) -> Result { meta.request_processor.get_transaction_count() } + fn request_airdrop(&self, meta: Self::Metadata, id: String, tokens: u64) -> Result { + let pubkey_vec = bs58::decode(id) + .into_vec() + .map_err(|_| Error::invalid_request())?; + if pubkey_vec.len() != mem::size_of::() { + return Err(Error::invalid_request()); + } + let pubkey = Pubkey::new(&pubkey_vec); + request_airdrop(&meta.drone_addr, &pubkey, tokens) + .map_err(|_| Error::internal_error())?; + Ok(true) + } fn send_transaction(&self, meta: Self::Metadata, data: Vec) -> Result { let tx: Transaction = deserialize(&data).map_err(|_| Error::invalid_request())?; let transactions_socket = UdpSocket::bind("0.0.0.0:0").unwrap(); @@ -182,6 +200,7 @@ mod tests { use jsonrpc_core::Response; use mint::Mint; use signature::{Keypair, KeypairUtil}; + use std::net::{IpAddr, Ipv4Addr, SocketAddr}; use std::sync::Arc; use transaction::Transaction; @@ -197,11 +216,16 @@ mod tests { let request_processor = JsonRpcRequestProcessor::new(Arc::new(bank)); let transactions_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), 0); + let drone_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), 0); let mut io = MetaIoHandler::default(); let rpc = RpcSolImpl; io.extend_with(rpc.to_delegate()); - let meta = Meta { request_processor, transactions_addr }; + let meta = Meta { + request_processor, + transactions_addr, + drone_addr, + }; let req = format!( r#"{{"jsonrpc":"2.0","id":1,"method":"getBalance","params":["{}"]}}"#, @@ -238,6 +262,7 @@ mod tests { let meta = Meta { request_processor: JsonRpcRequestProcessor::new(Arc::new(bank)), transactions_addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), 0), + drone_addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), 0), }; let res = io.handle_request_sync(req, meta); @@ -262,6 +287,7 @@ mod tests { let meta = Meta { request_processor: JsonRpcRequestProcessor::new(Arc::new(bank)), transactions_addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), 0), + drone_addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), 0), }; let res = io.handle_request_sync(req, meta); From 2b3ccbda4d5b1229b7cbe8e0d1369ff5a0506d82 Mon Sep 17 00:00:00 2001 From: Tyera Eulberg Date: Wed, 22 Aug 2018 12:25:21 -0600 Subject: [PATCH 3/4] Use balance to verify requestAirdrop success --- src/rpc.rs | 24 +++++++++++++++++++++--- 1 file changed, 21 insertions(+), 3 deletions(-) diff --git a/src/rpc.rs b/src/rpc.rs index 347b7cd8be5ee3..2382acca3e8331 100644 --- a/src/rpc.rs +++ b/src/rpc.rs @@ -11,7 +11,9 @@ use std::mem; use std::net::{SocketAddr, UdpSocket}; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; -use std::thread::{self, Builder, JoinHandle}; +use std::thread::{self, sleep, Builder, JoinHandle}; +use std::time::Duration; +use std::time::Instant; use transaction::Transaction; use wallet::request_airdrop; @@ -150,9 +152,25 @@ impl RpcSol for RpcSolImpl { return Err(Error::invalid_request()); } let pubkey = Pubkey::new(&pubkey_vec); - request_airdrop(&meta.drone_addr, &pubkey, tokens) + let previous_balance = meta + .request_processor + .get_balance(pubkey) .map_err(|_| Error::internal_error())?; - Ok(true) + request_airdrop(&meta.drone_addr, &pubkey, tokens).map_err(|_| Error::internal_error())?; + let now = Instant::now(); + let mut balance; + loop { + balance = meta + .request_processor + .get_balance(pubkey) + .map_err(|_| Error::internal_error())?; + if balance > previous_balance { + return Ok(true); + } else if now.elapsed().as_secs() > 5 { + return Err(Error::internal_error()); + } + sleep(Duration::from_millis(100)); + } } fn send_transaction(&self, meta: Self::Metadata, data: Vec) -> Result { let tx: Transaction = deserialize(&data).map_err(|_| Error::invalid_request())?; From b5d09e77d0d84e48fc3c404c6e326238c230a4d2 Mon Sep 17 00:00:00 2001 From: Tyera Eulberg Date: Wed, 22 Aug 2018 16:44:26 -0600 Subject: [PATCH 4/4] Pass bank to rpc as reference --- doc/json-rpc.md | 0 src/fullnode.rs | 4 ++-- src/rpc.rs | 4 ++-- 3 files changed, 4 insertions(+), 4 deletions(-) create mode 100644 doc/json-rpc.md diff --git a/doc/json-rpc.md b/doc/json-rpc.md new file mode 100644 index 00000000000000..e69de29bb2d1d6 diff --git a/src/fullnode.rs b/src/fullnode.rs index 2bf97640f73b7d..9349456e5dd8ae 100644 --- a/src/fullnode.rs +++ b/src/fullnode.rs @@ -207,7 +207,7 @@ impl Fullnode { drone_addr.set_port(DRONE_PORT); let rpc_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), RPC_PORT); let rpc_service = JsonRpcService::new( - bank.clone(), + &bank, node.data.contact_info.tpu, drone_addr, rpc_addr, @@ -310,7 +310,7 @@ impl Fullnode { drone_addr.set_port(DRONE_PORT); let rpc_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), RPC_PORT); let rpc_service = JsonRpcService::new( - bank.clone(), + &bank, node.data.contact_info.tpu, drone_addr, rpc_addr, diff --git a/src/rpc.rs b/src/rpc.rs index 2382acca3e8331..5e45b6139211fe 100644 --- a/src/rpc.rs +++ b/src/rpc.rs @@ -25,13 +25,13 @@ pub struct JsonRpcService { impl JsonRpcService { pub fn new( - bank: Arc, + bank: &Arc, transactions_addr: SocketAddr, drone_addr: SocketAddr, rpc_addr: SocketAddr, exit: Arc, ) -> Self { - let request_processor = JsonRpcRequestProcessor::new(bank); + let request_processor = JsonRpcRequestProcessor::new(bank.clone()); let thread_hdl = Builder::new() .name("solana-jsonrpc".to_string()) .spawn(move || {