use crate::beacon_chain::BeaconChain; use crossbeam_channel; use eth2_libp2p::rpc::methods::BlockRootSlot; use eth2_libp2p::PubsubMessage; use futures::Future; use grpcio::{RpcContext, RpcStatus, RpcStatusCode, UnarySink}; use network::NetworkMessage; use protos::services::{ BeaconBlock as BeaconBlockProto, ProduceBeaconBlockRequest, ProduceBeaconBlockResponse, PublishBeaconBlockRequest, PublishBeaconBlockResponse, }; use protos::services_grpc::BeaconBlockService; use slog::Logger; use slog::{error, info, trace, warn}; use ssz::{ssz_encode, Decodable, TreeHash}; use std::sync::Arc; use types::{BeaconBlock, Hash256, Signature, Slot}; #[derive(Clone)] pub struct BeaconBlockServiceInstance { pub chain: Arc, pub network_chan: crossbeam_channel::Sender, pub log: Logger, } impl BeaconBlockService for BeaconBlockServiceInstance { /// Produce a `BeaconBlock` for signing by a validator. fn produce_beacon_block( &mut self, ctx: RpcContext, req: ProduceBeaconBlockRequest, sink: UnarySink, ) { trace!(self.log, "Generating a beacon block"; "req" => format!("{:?}", req)); // decode the request // TODO: requested slot currently unused, see: https://github.com/sigp/lighthouse/issues/336 let _requested_slot = Slot::from(req.get_slot()); let (randao_reveal, _index) = match Signature::ssz_decode(req.get_randao_reveal(), 0) { Ok(v) => v, Err(_) => { // decode error, incorrect signature let log_clone = self.log.clone(); let f = sink .fail(RpcStatus::new( RpcStatusCode::InvalidArgument, Some(format!("Invalid randao reveal signature")), )) .map_err(move |e| warn!(log_clone, "failed to reply {:?}: {:?}", req, e)); return ctx.spawn(f); } }; let produced_block = match self.chain.produce_block(randao_reveal) { Ok((block, _state)) => block, Err(e) => { // could not produce a block let log_clone = self.log.clone(); warn!(self.log, "RPC Error"; "Error" => format!("Could not produce a block:{:?}",e)); let f = sink .fail(RpcStatus::new( RpcStatusCode::Unknown, Some(format!("Could not produce a block: {:?}", e)), )) .map_err(move |e| warn!(log_clone, "failed to reply {:?}: {:?}", req, e)); return ctx.spawn(f); } }; let mut block = BeaconBlockProto::new(); block.set_ssz(ssz_encode(&produced_block)); let mut resp = ProduceBeaconBlockResponse::new(); resp.set_block(block); let f = sink .success(resp) .map_err(move |e| println!("failed to reply {:?}: {:?}", req, e)); ctx.spawn(f) } /// Accept some fully-formed `BeaconBlock`, process and publish it. fn publish_beacon_block( &mut self, ctx: RpcContext, req: PublishBeaconBlockRequest, sink: UnarySink, ) { let mut resp = PublishBeaconBlockResponse::new(); let ssz_serialized_block = req.get_block().get_ssz(); match BeaconBlock::ssz_decode(ssz_serialized_block, 0) { Ok((block, _i)) => { let block_root = Hash256::from_slice(&block.hash_tree_root()[..]); match self.chain.process_block(block.clone()) { Ok(outcome) => { if outcome.sucessfully_processed() { // Block was successfully processed. info!( self.log, "PublishBeaconBlock"; "type" => "valid_block", "block_slot" => block.slot, "outcome" => format!("{:?}", outcome) ); // TODO: Obtain topics from the network service properly. let topic = types::TopicBuilder::new("beacon_chain".to_string()).build(); let message = PubsubMessage::Block(BlockRootSlot { block_root, slot: block.slot, }); match self.network_chan.send(NetworkMessage::Publish { topics: vec![topic], message, }) { Ok(_) => {} Err(_) => warn!( self.log, "Could not send published block to the network service" ), } resp.set_success(true); } else if outcome.is_invalid() { // Block was invalid. warn!( self.log, "PublishBeaconBlock"; "type" => "invalid_block", "outcome" => format!("{:?}", outcome) ); resp.set_success(false); resp.set_msg( format!("InvalidBlock: {:?}", outcome).as_bytes().to_vec(), ); } else { // Some failure during processing. warn!( self.log, "PublishBeaconBlock"; "type" => "unable_to_import", "outcome" => format!("{:?}", outcome) ); resp.set_success(false); resp.set_msg(format!("other: {:?}", outcome).as_bytes().to_vec()); } } Err(e) => { // Some failure during processing. error!( self.log, "PublishBeaconBlock"; "type" => "failed_to_process", "error" => format!("{:?}", e) ); resp.set_success(false); resp.set_msg(format!("failed_to_process: {:?}", e).as_bytes().to_vec()); } } resp.set_success(true); } Err(_) => { resp.set_success(false); resp.set_msg(b"Invalid SSZ".to_vec()); } }; let f = sink .success(resp) .map_err(move |e| println!("failed to reply {:?}: {:?}", req, e)); ctx.spawn(f) } }