zebra_rpc/lightwalletd/
server.rs

1//! A tonic gRPC server for the lightwalletd `CompactTxStreamer` client API.
2
3use std::{net::SocketAddr, time::Duration};
4
5use tokio::task::JoinHandle;
6use tonic::transport::{server::TcpIncoming, Server};
7use tower::BoxError;
8use zebra_chain::{chain_tip::ChainTip, parameters::Network};
9use zebra_node_services::mempool::{MempoolService, MempoolTxSubscriber};
10use zebra_state::ReadState;
11
12use crate::{
13    lightwalletd::{
14        compact_tx_streamer_server::CompactTxStreamerServer, tracing::GrpcTracingLayer,
15    },
16    methods::RpcServer as RpcMethods,
17    server::OPENED_LIGHTWALLETD_ENDPOINT_MSG,
18};
19
20/// Maximum concurrent HTTP/2 streams per connection.
21///
22/// Must be set explicitly: tonic defaults it to `None`, which removes hyper's own
23/// 200-stream default rather than applying it. Matches the indexer server.
24const MAX_CONCURRENT_STREAMS: u32 = 20;
25
26/// Interval between HTTP/2 keepalive pings sent to detect dead connections.
27const HTTP2_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(30);
28
29/// Timeout for HTTP/2 keepalive pings before the connection is closed.
30const HTTP2_KEEPALIVE_TIMEOUT: Duration = Duration::from_secs(10);
31
32type ServerTask = JoinHandle<Result<(), BoxError>>;
33
34/// Lightwalletd gRPC service.
35///
36/// Most methods are implemented by calling Zebra's JSON-RPC methods in-process, in the
37/// same way lightwalletd calls them on zcashd. Compact blocks and mempool streams are
38/// built directly from the state and mempool services.
39pub struct LightwalletdRPC<Rpc, ReadStateService, Mempool, Tip>
40where
41    Rpc: RpcMethods + Clone,
42    ReadStateService: ReadState,
43    Mempool: MempoolService,
44    Tip: ChainTip + Clone + Send + Sync + 'static,
45{
46    pub(super) rpc: Rpc,
47    pub(super) read_state: ReadStateService,
48    pub(super) mempool: Mempool,
49    pub(super) latest_chain_tip: Tip,
50    pub(super) mempool_change: MempoolTxSubscriber,
51    pub(super) network: Network,
52}
53
54/// Initializes the lightwalletd gRPC server.
55#[tracing::instrument(skip_all)]
56pub async fn init<Rpc, ReadStateService, Mempool, Tip>(
57    listen_addr: SocketAddr,
58    rpc: Rpc,
59    read_state: ReadStateService,
60    mempool: Mempool,
61    latest_chain_tip: Tip,
62    mempool_change: MempoolTxSubscriber,
63    network: Network,
64) -> Result<(ServerTask, SocketAddr), BoxError>
65where
66    Rpc: RpcMethods + Clone,
67    ReadStateService: ReadState,
68    Mempool: MempoolService,
69    Tip: ChainTip + Clone + Send + Sync + 'static,
70{
71    let lightwalletd_service = LightwalletdRPC {
72        rpc,
73        read_state,
74        mempool,
75        latest_chain_tip,
76        mempool_change,
77        network,
78    };
79
80    let reflection_service = tonic_reflection::server::Builder::configure()
81        .register_encoded_file_descriptor_set(crate::lightwalletd::FILE_DESCRIPTOR_SET)
82        .build_v1()?;
83
84    tracing::info!(
85        "Trying to open lightwalletd gRPC endpoint at {}...",
86        listen_addr,
87    );
88
89    let tcp_listener = tokio::net::TcpListener::bind(listen_addr).await?;
90
91    let listen_addr = tcp_listener.local_addr()?;
92    tracing::info!("{OPENED_LIGHTWALLETD_ENDPOINT_MSG}{}", listen_addr);
93
94    let server_task: JoinHandle<Result<(), BoxError>> = tokio::spawn(async move {
95        Server::builder()
96            .max_concurrent_streams(Some(MAX_CONCURRENT_STREAMS))
97            .http2_keepalive_interval(Some(HTTP2_KEEPALIVE_INTERVAL))
98            .http2_keepalive_timeout(Some(HTTP2_KEEPALIVE_TIMEOUT))
99            .layer(GrpcTracingLayer)
100            .add_service(reflection_service)
101            .add_service(CompactTxStreamerServer::new(lightwalletd_service))
102            .serve_with_incoming(TcpIncoming::from(tcp_listener))
103            .await?;
104
105        Ok(())
106    });
107
108    Ok((server_task, listen_addr))
109}
zebra_rpc/lightwalletd/server.rs

zebra_rpc/lightwalletd/
server.rs
