zebra_rpc/indexer/
server.rs

1//! A tonic RPC server for Zebra's indexer 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;
9use zebra_node_services::mempool::MempoolTxSubscriber;
10use zebra_state::ReadState;
11
12use crate::{indexer::indexer_server::IndexerServer, server::OPENED_RPC_ENDPOINT_MSG};
13
14/// Maximum concurrent HTTP/2 streams per connection.
15///
16/// The primary clients (Zaino and Zallet) each open 1-2 streams via
17/// `init_read_state_with_syncer`. 20 provides generous headroom while
18/// bounding resource usage from untrusted clients.
19const MAX_CONCURRENT_STREAMS: u32 = 20;
20
21/// Interval between HTTP/2 keepalive pings sent to detect dead connections.
22const HTTP2_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(30);
23
24/// Timeout for HTTP/2 keepalive pings before the connection is closed.
25const HTTP2_KEEPALIVE_TIMEOUT: Duration = Duration::from_secs(10);
26
27type ServerTask = JoinHandle<Result<(), BoxError>>;
28
29/// Indexer RPC service.
30pub struct IndexerRPC<ReadStateService, Tip>
31where
32    ReadStateService: ReadState,
33    Tip: ChainTip + Clone + Send + Sync + 'static,
34{
35    pub(super) read_state: ReadStateService,
36    pub(super) chain_tip_change: Tip,
37    pub(super) mempool_change: MempoolTxSubscriber,
38}
39
40/// Initializes the indexer RPC server
41#[tracing::instrument(skip_all)]
42pub async fn init<ReadStateService, Tip>(
43    listen_addr: SocketAddr,
44    read_state: ReadStateService,
45    chain_tip_change: Tip,
46    mempool_change: MempoolTxSubscriber,
47) -> Result<(ServerTask, SocketAddr), BoxError>
48where
49    ReadStateService: ReadState,
50    Tip: ChainTip + Clone + Send + Sync + 'static,
51{
52    let indexer_service = IndexerRPC {
53        read_state,
54        chain_tip_change,
55        mempool_change,
56    };
57
58    let reflection_service = tonic_reflection::server::Builder::configure()
59        .register_encoded_file_descriptor_set(crate::indexer::FILE_DESCRIPTOR_SET)
60        .build_v1()
61        .unwrap();
62
63    tracing::info!("Trying to open indexer RPC endpoint at {}...", listen_addr,);
64
65    let tcp_listener = tokio::net::TcpListener::bind(listen_addr).await?;
66
67    let listen_addr = tcp_listener.local_addr()?;
68    tracing::info!("{OPENED_RPC_ENDPOINT_MSG}{}", listen_addr);
69
70    let server_task: JoinHandle<Result<(), BoxError>> = tokio::spawn(async move {
71        Server::builder()
72            .max_concurrent_streams(Some(MAX_CONCURRENT_STREAMS))
73            .http2_keepalive_interval(Some(HTTP2_KEEPALIVE_INTERVAL))
74            .http2_keepalive_timeout(Some(HTTP2_KEEPALIVE_TIMEOUT))
75            .add_service(reflection_service)
76            .add_service(IndexerServer::new(indexer_service))
77            .serve_with_incoming(TcpIncoming::from(tcp_listener))
78            .await?;
79
80        Ok(())
81    });
82
83    Ok((server_task, listen_addr))
84}
zebra_rpc/indexer/server.rs

zebra_rpc/indexer/
server.rs
