2020-09-18 12:18:22 -07:00
|
|
|
use std::{
|
|
|
|
future::Future,
|
|
|
|
pin::Pin,
|
|
|
|
sync::{Arc, Mutex},
|
|
|
|
task::{Context, Poll},
|
|
|
|
};
|
|
|
|
|
2020-09-18 13:47:31 -07:00
|
|
|
use futures::{
|
|
|
|
future::{FutureExt, TryFutureExt},
|
2021-01-24 20:55:36 -08:00
|
|
|
stream::Stream,
|
2020-09-18 13:47:31 -07:00
|
|
|
};
|
2021-01-27 14:18:26 -08:00
|
|
|
use oneshot::error::TryRecvError;
|
2020-09-18 12:18:22 -07:00
|
|
|
use tokio::sync::oneshot;
|
2021-01-13 01:39:33 -08:00
|
|
|
use tower::{buffer::Buffer, timeout::Timeout, util::BoxService, Service, ServiceExt};
|
2020-09-18 12:18:22 -07:00
|
|
|
|
|
|
|
use zebra_network as zn;
|
2020-09-18 13:47:31 -07:00
|
|
|
use zebra_state as zs;
|
2020-09-18 12:18:22 -07:00
|
|
|
|
2020-11-24 14:53:57 -08:00
|
|
|
use zebra_chain::block::{self, Block};
|
|
|
|
use zebra_consensus::chain::VerifyChainError;
|
|
|
|
use zebra_network::AddressBook;
|
|
|
|
|
2021-01-13 01:39:33 -08:00
|
|
|
// Re-use the syncer timeouts for consistency.
|
|
|
|
use super::sync::{BLOCK_DOWNLOAD_TIMEOUT, BLOCK_VERIFY_TIMEOUT};
|
|
|
|
|
2020-11-19 09:55:06 -08:00
|
|
|
mod downloads;
|
|
|
|
use downloads::Downloads;
|
|
|
|
|
2020-09-18 12:18:22 -07:00
|
|
|
type Outbound = Buffer<BoxService<zn::Request, zn::Response, zn::BoxError>, zn::Request>;
|
2020-09-18 13:47:31 -07:00
|
|
|
type State = Buffer<BoxService<zs::Request, zs::Response, zs::BoxError>, zs::Request>;
|
2020-11-24 14:53:57 -08:00
|
|
|
type Verifier = Buffer<BoxService<Arc<Block>, block::Hash, VerifyChainError>, Arc<Block>>;
|
2021-01-24 22:11:19 -08:00
|
|
|
type InboundDownloads = Downloads<Timeout<Outbound>, Timeout<Verifier>, State>;
|
2020-09-18 12:18:22 -07:00
|
|
|
|
2021-01-24 22:11:19 -08:00
|
|
|
pub type NetworkSetupData = (Outbound, Arc<Mutex<AddressBook>>);
|
|
|
|
|
|
|
|
/// Tracks the internal state of the [`Inbound`] service during network setup.
|
|
|
|
pub enum Setup {
|
|
|
|
/// Waiting for network setup to complete.
|
|
|
|
///
|
|
|
|
/// Requests that depend on Zebra's internal network setup are ignored.
|
|
|
|
/// Other requests are answered.
|
|
|
|
AwaitingNetwork {
|
|
|
|
/// A oneshot channel used to receive the address_book and outbound services
|
|
|
|
/// after the network is set up.
|
|
|
|
network_setup: oneshot::Receiver<NetworkSetupData>,
|
|
|
|
|
|
|
|
/// A service that verifies downloaded blocks. Given to `downloads`
|
|
|
|
/// after the network is set up.
|
|
|
|
verifier: Verifier,
|
|
|
|
},
|
|
|
|
|
|
|
|
/// Network setup is complete.
|
|
|
|
///
|
|
|
|
/// All requests are answered.
|
|
|
|
Initialized {
|
|
|
|
/// A shared list of peer addresses.
|
|
|
|
address_book: Arc<Mutex<zn::AddressBook>>,
|
|
|
|
|
|
|
|
/// A `futures::Stream` that downloads and verifies gossipped blocks.
|
|
|
|
downloads: Pin<Box<InboundDownloads>>,
|
|
|
|
},
|
|
|
|
|
|
|
|
/// Temporary state used in the service's internal network initialization
|
|
|
|
/// code.
|
|
|
|
///
|
|
|
|
/// If this state occurs outside the service initialization code, the
|
|
|
|
/// service panics.
|
|
|
|
FailedInit,
|
|
|
|
|
|
|
|
/// Network setup failed, because the setup channel permanently failed.
|
|
|
|
/// The service keeps returning readiness errors for every request.
|
|
|
|
///
|
|
|
|
/// We keep hold of the closed oneshot, so we can use it to create a
|
|
|
|
/// new error for each `poll_ready` call.
|
2021-01-27 14:18:26 -08:00
|
|
|
FailedRecv { error: SharedRecvError },
|
|
|
|
}
|
|
|
|
|
|
|
|
#[derive(thiserror::Error, Debug, Clone)]
|
|
|
|
#[error(transparent)]
|
|
|
|
pub struct SharedRecvError(Arc<TryRecvError>);
|
|
|
|
|
|
|
|
impl From<TryRecvError> for SharedRecvError {
|
|
|
|
fn from(source: TryRecvError) -> Self {
|
|
|
|
Self(Arc::new(source))
|
|
|
|
}
|
2021-01-24 22:11:19 -08:00
|
|
|
}
|
2020-09-18 12:18:22 -07:00
|
|
|
|
2020-09-18 16:13:57 -07:00
|
|
|
/// Uses the node state to respond to inbound peer requests.
|
|
|
|
///
|
|
|
|
/// This service, wrapped in appropriate middleware, is passed to
|
|
|
|
/// `zebra_network::init` to respond to inbound peer requests.
|
|
|
|
///
|
|
|
|
/// The `Inbound` service is responsible for:
|
|
|
|
///
|
|
|
|
/// - supplying network data like peer addresses to other nodes;
|
|
|
|
/// - supplying chain data like blocks to other nodes;
|
|
|
|
/// - performing transaction diffusion;
|
|
|
|
/// - performing block diffusion.
|
|
|
|
///
|
|
|
|
/// Because the `Inbound` service is responsible for participating in the gossip
|
|
|
|
/// protocols used for transaction and block diffusion, there is a potential
|
|
|
|
/// overlap with the `ChainSync` component.
|
|
|
|
///
|
|
|
|
/// The division of responsibility is that the `ChainSync` component is
|
|
|
|
/// *internally driven*, periodically polling the network to check whether it is
|
|
|
|
/// behind the current tip, while the `Inbound` service is *externally driven*,
|
|
|
|
/// responding to block gossip by attempting to download and validate advertised
|
|
|
|
/// blocks.
|
2020-09-18 12:18:22 -07:00
|
|
|
pub struct Inbound {
|
2021-01-24 22:11:19 -08:00
|
|
|
/// Provides network-dependent services, if they are available.
|
2021-01-13 02:26:50 -08:00
|
|
|
///
|
2021-01-24 22:11:19 -08:00
|
|
|
/// Some services are unavailable until Zebra has completed network setup.
|
|
|
|
network: Setup,
|
2021-01-13 02:26:50 -08:00
|
|
|
|
|
|
|
/// A service that manages cached blockchain state.
|
2020-09-18 13:47:31 -07:00
|
|
|
state: State,
|
2020-09-18 12:18:22 -07:00
|
|
|
}
|
|
|
|
|
|
|
|
impl Inbound {
|
2020-11-19 09:55:06 -08:00
|
|
|
pub fn new(
|
2021-01-24 22:11:19 -08:00
|
|
|
network_setup: oneshot::Receiver<NetworkSetupData>,
|
2020-11-19 09:55:06 -08:00
|
|
|
state: State,
|
|
|
|
verifier: Verifier,
|
|
|
|
) -> Self {
|
2020-09-18 12:18:22 -07:00
|
|
|
Self {
|
2021-01-24 22:11:19 -08:00
|
|
|
network: Setup::AwaitingNetwork {
|
|
|
|
network_setup,
|
|
|
|
verifier,
|
|
|
|
},
|
2020-09-18 13:47:31 -07:00
|
|
|
state,
|
2020-09-18 12:18:22 -07:00
|
|
|
}
|
|
|
|
}
|
2021-01-27 15:09:40 -08:00
|
|
|
|
|
|
|
fn take_setup(&mut self) -> Setup {
|
|
|
|
let mut network = Setup::FailedInit;
|
|
|
|
std::mem::swap(&mut self.network, &mut network);
|
|
|
|
network
|
|
|
|
}
|
2020-09-18 12:18:22 -07:00
|
|
|
}
|
|
|
|
|
|
|
|
impl Service<zn::Request> for Inbound {
|
|
|
|
type Response = zn::Response;
|
|
|
|
type Error = zn::BoxError;
|
|
|
|
type Future =
|
|
|
|
Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;
|
|
|
|
|
2020-09-18 14:11:29 -07:00
|
|
|
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
|
2020-09-19 23:50:51 -07:00
|
|
|
// Check whether the network setup is finished, but don't wait for it to
|
|
|
|
// become ready before reporting readiness. We expect to get it "soon",
|
|
|
|
// and reporting unreadiness might cause unwanted load-shedding, since
|
|
|
|
// the load-shed middleware is unable to distinguish being unready due
|
|
|
|
// to load from being unready while waiting on setup.
|
2021-01-27 15:09:40 -08:00
|
|
|
let mut result = Ok(());
|
|
|
|
|
|
|
|
self.network = match self.take_setup() {
|
|
|
|
Setup::AwaitingNetwork {
|
2021-01-24 22:11:19 -08:00
|
|
|
mut network_setup,
|
|
|
|
verifier,
|
2021-01-27 15:09:40 -08:00
|
|
|
} => match network_setup.try_recv() {
|
|
|
|
Ok((outbound, address_book)) => {
|
|
|
|
let downloads = Box::pin(Downloads::new(
|
|
|
|
Timeout::new(outbound, BLOCK_DOWNLOAD_TIMEOUT),
|
|
|
|
Timeout::new(verifier, BLOCK_VERIFY_TIMEOUT),
|
|
|
|
self.state.clone(),
|
|
|
|
));
|
|
|
|
Setup::Initialized {
|
|
|
|
address_book,
|
|
|
|
downloads,
|
2021-01-24 22:11:19 -08:00
|
|
|
}
|
2021-01-27 15:09:40 -08:00
|
|
|
}
|
|
|
|
Err(TryRecvError::Empty) => {
|
|
|
|
// There's no setup data yet, so keep waiting for it
|
|
|
|
Setup::AwaitingNetwork {
|
|
|
|
network_setup,
|
|
|
|
verifier,
|
2021-01-24 22:11:19 -08:00
|
|
|
}
|
|
|
|
}
|
2021-01-27 15:09:40 -08:00
|
|
|
Err(error @ TryRecvError::Closed) => {
|
|
|
|
// Mark the service as failed, because network setup failed
|
|
|
|
error!(?error, "inbound network setup failed");
|
|
|
|
let error: SharedRecvError = error.into();
|
|
|
|
result = Err(error.clone().into());
|
|
|
|
Setup::FailedRecv { error }
|
|
|
|
}
|
|
|
|
},
|
|
|
|
// Make sure we left the network setup in a valid state
|
|
|
|
Setup::FailedInit => unreachable!("incomplete Inbound initialization"),
|
|
|
|
// If network setup failed, report service failure
|
|
|
|
Setup::FailedRecv { error } => {
|
|
|
|
result = Err(error.clone().into());
|
|
|
|
Setup::FailedRecv { error }
|
2021-01-24 22:11:19 -08:00
|
|
|
}
|
2021-01-27 15:09:40 -08:00
|
|
|
// Clean up completed download tasks, ignoring their results
|
|
|
|
Setup::Initialized {
|
|
|
|
address_book,
|
|
|
|
mut downloads,
|
|
|
|
} => {
|
|
|
|
while let Poll::Ready(Some(_)) = downloads.as_mut().poll_next(cx) {}
|
|
|
|
Setup::Initialized {
|
|
|
|
address_book,
|
|
|
|
downloads,
|
|
|
|
}
|
|
|
|
}
|
|
|
|
};
|
2020-11-19 09:55:06 -08:00
|
|
|
|
2021-01-22 05:03:33 -08:00
|
|
|
// TODO:
|
2021-01-24 20:55:36 -08:00
|
|
|
// * do we want to propagate backpressure from the download queue or its outbound network?
|
|
|
|
// currently, the download queue waits for the outbound network in the download future,
|
|
|
|
// and drops new requests after it reaches a hard-coded limit. This is the
|
|
|
|
// "load shed directly" pattern from #1618.
|
|
|
|
// * currently, the state service is always ready, unless its buffer is full.
|
|
|
|
// So we might also want to propagate backpressure from its buffer.
|
|
|
|
// * if we want to propagate backpressure, add a ReadyCache for each service, to ensure
|
|
|
|
// that each poll_ready has a matching call. See #1593 for details.
|
2021-01-27 15:09:40 -08:00
|
|
|
Poll::Ready(result)
|
2020-09-18 12:18:22 -07:00
|
|
|
}
|
|
|
|
|
2020-11-24 16:05:11 -08:00
|
|
|
#[instrument(name = "inbound", skip(self, req))]
|
2020-09-18 12:18:22 -07:00
|
|
|
fn call(&mut self, req: zn::Request) -> Self::Future {
|
|
|
|
match req {
|
2021-01-24 22:11:19 -08:00
|
|
|
zn::Request::Peers => {
|
|
|
|
if let Setup::Initialized { address_book, .. } = &self.network {
|
2020-09-19 23:50:51 -07:00
|
|
|
// We could truncate the list to try to not reveal our entire
|
|
|
|
// peer set. But because we don't monitor repeated requests,
|
|
|
|
// this wouldn't actually achieve anything, because a crawler
|
|
|
|
// could just repeatedly query it.
|
2021-01-24 22:11:19 -08:00
|
|
|
let mut peers = address_book.lock().unwrap().sanitized();
|
2020-09-19 23:50:51 -07:00
|
|
|
const MAX_ADDR: usize = 1000; // bitcoin protocol constant
|
|
|
|
peers.truncate(MAX_ADDR);
|
|
|
|
async { Ok(zn::Response::Peers(peers)) }.boxed()
|
2021-01-24 22:11:19 -08:00
|
|
|
} else {
|
2021-01-24 21:16:04 -08:00
|
|
|
info!("ignoring `Peers` request from remote peer during network setup");
|
|
|
|
async { Ok(zn::Response::Nil) }.boxed()
|
|
|
|
}
|
2021-01-24 22:11:19 -08:00
|
|
|
}
|
2020-09-18 13:47:31 -07:00
|
|
|
zn::Request::BlocksByHash(hashes) => {
|
2021-01-22 05:03:33 -08:00
|
|
|
// Correctness:
|
|
|
|
//
|
2021-01-24 20:55:36 -08:00
|
|
|
// We can't use `call_all` here, because it leaks buffer slots:
|
2021-01-22 05:03:33 -08:00
|
|
|
// https://github.com/tower-rs/tower/blob/master/tower/src/util/call_all/common.rs#L112
|
2021-01-27 14:44:22 -08:00
|
|
|
use futures::stream::TryStreamExt;
|
|
|
|
hashes
|
|
|
|
.into_iter()
|
|
|
|
.map(|hash| zs::Request::Block(hash.into()))
|
|
|
|
.map(|request| self.state.clone().oneshot(request))
|
|
|
|
.collect::<futures::stream::FuturesOrdered<_>>()
|
|
|
|
.try_filter_map(|response| async move {
|
|
|
|
Ok(match response {
|
|
|
|
zs::Response::Block(Some(block)) => Some(block),
|
2021-01-03 19:25:35 -08:00
|
|
|
// `zcashd` ignores missing blocks in GetData responses,
|
|
|
|
// rather than including them in a trailing `NotFound`
|
|
|
|
// message
|
2021-01-27 14:44:22 -08:00
|
|
|
zs::Response::Block(None) => None,
|
2020-09-18 13:47:31 -07:00
|
|
|
_ => unreachable!("wrong response from state"),
|
2021-01-27 14:44:22 -08:00
|
|
|
})
|
|
|
|
})
|
|
|
|
.try_collect::<Vec<_>>()
|
|
|
|
.map_ok(zn::Response::Blocks)
|
|
|
|
.boxed()
|
2020-09-18 13:47:31 -07:00
|
|
|
}
|
|
|
|
zn::Request::TransactionsByHash(_transactions) => {
|
2021-01-03 19:25:35 -08:00
|
|
|
// `zcashd` returns a list of found transactions, followed by a
|
|
|
|
// `NotFound` message if any transactions are missing. `zcashd`
|
|
|
|
// says that Simplified Payment Verification (SPV) clients rely on
|
|
|
|
// this behaviour - are there any of them on the Zcash network?
|
|
|
|
// https://github.com/zcash/zcash/blob/e7b425298f6d9a54810cb7183f00be547e4d9415/src/main.cpp#L5632
|
|
|
|
// We'll implement this request once we have a mempool:
|
|
|
|
// https://en.bitcoin.it/wiki/Protocol_documentation#getdata
|
2020-09-18 13:47:31 -07:00
|
|
|
debug!("ignoring unimplemented request");
|
|
|
|
async { Ok(zn::Response::Nil) }.boxed()
|
|
|
|
}
|
2020-11-30 13:30:37 -08:00
|
|
|
zn::Request::FindBlocks { known_blocks, stop } => {
|
|
|
|
let request = zs::Request::FindBlockHashes { known_blocks, stop };
|
2021-01-22 05:03:33 -08:00
|
|
|
self.state.clone().oneshot(request).map_ok(|resp| match resp {
|
2020-11-30 13:30:37 -08:00
|
|
|
zs::Response::BlockHashes(hashes) if hashes.is_empty() => zn::Response::Nil,
|
|
|
|
zs::Response::BlockHashes(hashes) => zn::Response::BlockHashes(hashes),
|
|
|
|
_ => unreachable!("zebra-state should always respond to a `FindBlockHashes` request with a `BlockHashes` response"),
|
|
|
|
})
|
|
|
|
.boxed()
|
2020-09-18 13:47:31 -07:00
|
|
|
}
|
2020-11-30 13:30:37 -08:00
|
|
|
zn::Request::FindHeaders { known_blocks, stop } => {
|
|
|
|
let request = zs::Request::FindBlockHeaders { known_blocks, stop };
|
2021-01-22 05:03:33 -08:00
|
|
|
self.state.clone().oneshot(request).map_ok(|resp| match resp {
|
2020-11-30 13:30:37 -08:00
|
|
|
zs::Response::BlockHeaders(headers) if headers.is_empty() => zn::Response::Nil,
|
|
|
|
zs::Response::BlockHeaders(headers) => zn::Response::BlockHeaders(headers),
|
|
|
|
_ => unreachable!("zebra-state should always respond to a `FindBlockHeaders` request with a `BlockHeaders` response"),
|
|
|
|
})
|
|
|
|
.boxed()
|
2020-09-18 22:16:59 -07:00
|
|
|
}
|
2020-09-18 13:47:31 -07:00
|
|
|
zn::Request::PushTransaction(_transaction) => {
|
|
|
|
debug!("ignoring unimplemented request");
|
|
|
|
async { Ok(zn::Response::Nil) }.boxed()
|
|
|
|
}
|
|
|
|
zn::Request::AdvertiseTransactions(_transactions) => {
|
2020-09-18 12:18:22 -07:00
|
|
|
debug!("ignoring unimplemented request");
|
|
|
|
async { Ok(zn::Response::Nil) }.boxed()
|
|
|
|
}
|
2020-11-19 09:55:06 -08:00
|
|
|
zn::Request::AdvertiseBlock(hash) => {
|
2021-01-24 22:11:19 -08:00
|
|
|
if let Setup::Initialized { downloads, .. } = &mut self.network {
|
2020-11-24 14:20:34 -08:00
|
|
|
downloads.download_and_verify(hash);
|
2021-01-24 21:16:04 -08:00
|
|
|
} else {
|
|
|
|
info!(
|
|
|
|
?hash,
|
|
|
|
"ignoring `AdvertiseBlock` request from remote peer during network setup"
|
|
|
|
);
|
2020-11-19 09:55:06 -08:00
|
|
|
}
|
2020-11-24 14:20:34 -08:00
|
|
|
async { Ok(zn::Response::Nil) }.boxed()
|
2020-09-18 13:47:31 -07:00
|
|
|
}
|
2020-09-18 22:16:59 -07:00
|
|
|
zn::Request::MempoolTransactions => {
|
|
|
|
debug!("ignoring unimplemented request");
|
|
|
|
async { Ok(zn::Response::Nil) }.boxed()
|
|
|
|
}
|
2020-09-18 13:47:31 -07:00
|
|
|
zn::Request::Ping(_) => {
|
|
|
|
unreachable!("ping requests are handled internally");
|
|
|
|
}
|
2020-09-18 12:18:22 -07:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|