Skip to main content

zebrad/commands/
start.rs

1//! `start` subcommand - entry point for starting a zebra node
2//!
3//! ## Application Structure
4//!
5//! A zebra node consists of the following major services and tasks:
6//!
7//! Peers:
8//!  * Peer Connection Pool Service
9//!    * primary external interface for outbound requests from this node to remote peers
10//!    * accepts requests from services and tasks in this node, and sends them to remote peers
11//!  * Peer Discovery Service
12//!    * maintains a list of peer addresses, and connection priority metadata
13//!    * discovers new peer addresses from existing peer connections
14//!    * initiates new outbound peer connections in response to demand from tasks within this node
15//!  * Peer Cache Service
16//!    * Reads previous peer cache on startup, and adds it to the configured DNS seed peers
17//!    * Periodically updates the peer cache on disk from the latest address book state
18//!
19//! Blocks & Mempool Transactions:
20//!  * Consensus Service
21//!    * handles all validation logic for the node
22//!    * verifies blocks using zebra-chain, then stores verified blocks in zebra-state
23//!    * verifies mempool and block transactions using zebra-chain and zebra-script,
24//!      and returns verified mempool transactions for mempool storage
25//!  * Inbound Service
26//!    * primary external interface for inbound peer requests to this node
27//!    * handles requests from peers for network data, chain data, and mempool transactions
28//!    * spawns download and verify tasks for each gossiped block
29//!    * sends gossiped transactions to the mempool service
30//!
31//! Blocks:
32//!  * Sync Task
33//!    * runs in the background and continuously queries the network for
34//!      new blocks to be verified and added to the local state
35//!    * spawns download and verify tasks for each crawled block
36//!  * State Service
37//!    * contextually verifies blocks
38//!    * handles in-memory storage of multiple non-finalized chains
39//!    * handles permanent storage of the best finalized chain
40//!  * Old State Version Cleanup Task
41//!    * deletes outdated state versions
42//!  * Block Gossip Task
43//!    * runs in the background and continuously queries the state for
44//!      newly committed blocks to be gossiped to peers
45//!  * Block Notify Task
46//!    * if the user has configured a `notify.block_notify_command`, runs that command
47//!      whenever the best chain tip changes (Zebra's equivalent of zcashd's `-blocknotify`)
48//!  * Progress Task
49//!    * logs progress towards the chain tip
50//!
51//! Block Mining:
52//!  * Internal Miner Task
53//!    * if the user has configured Zebra to mine blocks, spawns tasks to generate new blocks,
54//!      and submits them for verification. This automatically shares these new blocks with peers.
55//!
56//! Mempool Transactions:
57//!  * Mempool Service
58//!    * activates when the syncer is near the chain tip
59//!    * spawns download and verify tasks for each crawled or gossiped transaction
60//!    * handles in-memory storage of unmined transactions
61//!  * Queue Checker Task
62//!    * runs in the background, polling the mempool to store newly verified transactions
63//!  * Transaction Gossip Task
64//!    * runs in the background and gossips newly added mempool transactions
65//!      to peers
66//!
67//! Remote Procedure Calls:
68//!  * JSON-RPC Service
69//!    * answers RPC client requests using the State Service and Mempool Service
70//!    * submits client transactions to the node's mempool
71//!
72//! Zebra also has diagnostic support:
73//! * [metrics](https://github.com/ZcashFoundation/zebra/blob/main/book/src/user/metrics.md)
74//! * [tracing](https://github.com/ZcashFoundation/zebra/blob/main/book/src/user/tracing.md)
75//! * [progress-bar](https://docs.rs/howudoin/0.1.1/howudoin)
76//!
77//! Some of the diagnostic features are optional, and need to be enabled at compile-time.
78
79use std::{
80    net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr},
81    path::Path,
82    sync::Arc,
83};
84
85use abscissa_core::{config, Command, FrameworkError};
86use color_eyre::eyre::{eyre, Report};
87use futures::FutureExt;
88use tokio::{
89    pin, select,
90    sync::{oneshot, watch},
91};
92use tower::{builder::ServiceBuilder, util::BoxService, ServiceExt};
93use tracing_futures::Instrument;
94
95use zebra_chain::block::genesis::regtest_genesis_block;
96use zebra_consensus::router::BackgroundTaskHandles;
97use zebra_rpc::{methods::RpcImpl, server::RpcServer, SubmitBlockChannel};
98
99use crate::{
100    application::{build_version, user_agent, LAST_WARN_ERROR_LOG_SENDER},
101    components::{
102        health,
103        inbound::{self, InboundSetupData, MAX_INBOUND_RESPONSE_TIME},
104        mempool::{self, Mempool},
105        notify::{self, BlockNotifyError},
106        sync::{self, show_block_chain_progress, VERIFICATION_PIPELINE_SCALING_MULTIPLIER},
107        tokio::{RuntimeRun, TokioComponent},
108        zcashd_compat, ChainSync, Inbound,
109    },
110    config::ZebradConfig,
111    prelude::*,
112};
113
114#[cfg(feature = "internal-miner")]
115use crate::components;
116
117/// Start the application (default command)
118#[derive(Command, Debug, Default, clap::Parser)]
119pub struct StartCmd {
120    /// Filter strings which override the config file and defaults
121    #[clap(help = "tracing filters which override the zebrad.toml config")]
122    filters: Vec<String>,
123
124    /// Enable zcashd-compat mode.
125    #[clap(long)]
126    zcashd_compat: bool,
127
128    /// Continue startup even when zcashd-compat preflight detects minimum hardware shortfalls.
129    #[clap(long = "unsafe-low-specs")]
130    unsafe_low_specs: bool,
131}
132
133/// Warns if Linux TCP slow-start-after-idle is enabled, which significantly
134/// reduces single-peer throughput for block propagation.
135///
136/// See `book/src/user/troubleshooting.md`.
137#[cfg(target_os = "linux")]
138fn check_tcp_slow_start_after_idle() {
139    const PATH: &str = "/proc/sys/net/ipv4/tcp_slow_start_after_idle";
140
141    let raw = match std::fs::read_to_string(PATH) {
142        Ok(raw) => raw,
143        Err(error) => {
144            debug!(
145                ?error,
146                path = PATH,
147                "could not read TCP sysctl, skipping check"
148            );
149            return;
150        }
151    };
152
153    if raw.trim() == "0" {
154        return;
155    }
156
157    warn!(
158        setting = "net.ipv4.tcp_slow_start_after_idle",
159        "TCP slow-start-after-idle is enabled, which resets TCP's congestion window \
160         between block requests and significantly reduces single-peer throughput for \
161         block propagation. \
162         Hint: set `net.ipv4.tcp_slow_start_after_idle=0` via sysctl. \
163         See https://zebra.zfnd.org/user/troubleshooting.html#linux-tcp-tuning-for-block-propagation"
164    );
165}
166
167#[cfg(not(target_os = "linux"))]
168fn check_tcp_slow_start_after_idle() {}
169
170impl StartCmd {
171    /// Extra time Zebra waits for the zcashd-compat supervisor task beyond the
172    /// child's `shutdown_grace_period`. The supervisor's `terminate_child` waits
173    /// the full grace period before its SIGKILL last resort, so the outer wait
174    /// must be strictly longer or aborting the task races the graceful path.
175    const ZCASHD_COMPAT_SHUTDOWN_TIMEOUT_MARGIN: std::time::Duration =
176        std::time::Duration::from_secs(30);
177
178    /// Returns the Zebra P2P address supervised zcashd should `-connect` to.
179    ///
180    /// Uses `zcashd_compat.p2p_connect_addr` when set, otherwise Zebra's bound
181    /// P2P listener, substituting loopback for unspecified addresses so
182    /// zcashd gets a dialable target on the same host.
183    fn zcashd_compat_p2p_connect_addr(
184        config: &ZebradConfig,
185        local_listener: SocketAddr,
186    ) -> SocketAddr {
187        if let Some(addr) = config.zcashd_compat.p2p_connect_addr {
188            return addr;
189        }
190
191        if local_listener.ip().is_unspecified() {
192            // Substitute the loopback address of the same IP family: an
193            // IPv6-only listener is not reachable via 127.0.0.1.
194            match local_listener.ip() {
195                IpAddr::V4(_) => SocketAddr::from(([127, 0, 0, 1], local_listener.port())),
196                IpAddr::V6(_) => {
197                    SocketAddr::new(IpAddr::V6(Ipv6Addr::LOCALHOST), local_listener.port())
198                }
199            }
200        } else {
201            local_listener
202        }
203    }
204
205    /// Returns the default inbound peer IPs that always receive block gossip in
206    /// zcashd-compat mode.
207    fn zcashd_compat_default_block_gossip_peer_ips() -> Vec<IpAddr> {
208        vec![
209            IpAddr::V4(Ipv4Addr::LOCALHOST),
210            IpAddr::V6(Ipv6Addr::LOCALHOST),
211        ]
212    }
213
214    /// Returns the supervisor shutdown timeout when zcashd-compat `zcashd` supervision is active.
215    ///
216    /// This is the configured `shutdown_grace_period` plus a fixed margin, so the
217    /// supervisor task always gets to finish its own SIGTERM → grace → SIGKILL
218    /// sequence before Zebra gives up on the task.
219    fn zcashd_compat_supervisor_shutdown_timeout(
220        config: &ZebradConfig,
221    ) -> Option<std::time::Duration> {
222        (config.zcashd_compat.enabled && config.zcashd_compat.manage_zcashd).then_some(
223            config
224                .zcashd_compat
225                .shutdown_grace_period
226                .saturating_add(Self::ZCASHD_COMPAT_SHUTDOWN_TIMEOUT_MARGIN),
227        )
228    }
229
230    /// Returns `false` so Zebra keeps running if zcashd-compat supervision exits unexpectedly.
231    fn zcashd_compat_supervisor_should_exit(
232        zcashd_compat_result: Result<Result<(), Report>, tokio::task::JoinError>,
233    ) -> bool {
234        zcashd_compat::set_supervision_unexpectedly_disabled_metrics();
235
236        match zcashd_compat_result {
237            Ok(Ok(())) => {
238                warn!(
239                    "zcashd-compat supervisor task exited unexpectedly in supervision mode; \
240                     continuing without zcashd supervision"
241                );
242            }
243            Ok(Err(err)) => {
244                warn!(
245                    ?err,
246                    "zcashd-compat supervisor task failed in supervision mode; \
247                     continuing without zcashd supervision"
248                );
249            }
250            Err(join_err) => {
251                warn!(
252                    ?join_err,
253                    "zcashd-compat supervisor task panicked in supervision mode; \
254                     continuing without zcashd supervision"
255                );
256            }
257        }
258
259        false
260    }
261
262    async fn start(&self) -> Result<(), Report> {
263        check_tcp_slow_start_after_idle();
264
265        let config = APPLICATION.config();
266        let is_regtest = config.network.network.is_regtest();
267
268        let config = if is_regtest {
269            Arc::new(ZebradConfig {
270                mempool: mempool::Config {
271                    debug_enable_at_height: Some(0),
272                    ..config.mempool
273                },
274                ..Arc::unwrap_or_clone(config)
275            })
276        } else {
277            config
278        };
279
280        let zcashd_compat_block_gossip_peer_ips = if config.zcashd_compat.enabled {
281            if config.zcashd_compat.block_gossip_peer_ips.is_empty() {
282                // The sidecar privileges (pinned gossip, reserved slot, stall
283                // exemption) match on the sidecar's *source* IP. In
284                // cross-container/cross-host topologies that source is not
285                // loopback, so the default list would silently strip the
286                // sidecar of everything this mode provides.
287                if config
288                    .zcashd_compat
289                    .p2p_connect_addr
290                    .is_some_and(|addr| !addr.ip().is_loopback())
291                {
292                    warn!(
293                        p2p_connect_addr = ?config.zcashd_compat.p2p_connect_addr,
294                        "zcashd_compat.p2p_connect_addr is not loopback, but \
295                         zcashd_compat.block_gossip_peer_ips defaults to loopback only; \
296                         if the sidecar connects from a non-loopback IP, set \
297                         block_gossip_peer_ips to that IP or it will not receive \
298                         pinned block gossip"
299                    );
300                }
301
302                Self::zcashd_compat_default_block_gossip_peer_ips()
303            } else {
304                config.zcashd_compat.block_gossip_peer_ips.clone()
305            }
306        } else {
307            Vec::new()
308        };
309
310        if config.zcashd_compat.enabled {
311            // Preflight does blocking filesystem and /proc reads, and can hash
312            // the cached zcashd binary, so keep it off the async runtime.
313            let preflight_config = config.clone();
314            let unsafe_low_specs = self.unsafe_low_specs;
315            tokio::task::spawn_blocking(move || {
316                zcashd_compat::run_preflight(&preflight_config, unsafe_low_specs)
317            })
318            .await
319            .map_err(|err| eyre!("failed to join zcashd-compat preflight task: {err}"))??;
320        }
321
322        let resolved_zcashd_path = if config.zcashd_compat.enabled
323            && config.zcashd_compat.manage_zcashd
324        {
325            let zcashd_compat_config = config.zcashd_compat.clone();
326            let state_cache_dir = config.state.cache_dir.clone();
327            Some(
328                tokio::task::spawn_blocking(move || {
329                    zcashd_compat::resolve_zcashd_binary_path(
330                        &zcashd_compat_config,
331                        &state_cache_dir,
332                    )
333                })
334                .await
335                .map_err(|err| eyre!("failed to join managed zcashd binary resolver: {err}"))??,
336            )
337        } else {
338            None
339        };
340
341        info!("initializing node state");
342        let (_, max_checkpoint_height) = zebra_consensus::router::init_checkpoint_list(
343            config.consensus.clone(),
344            &config.network.network,
345        );
346
347        info!("opening database, this may take a few minutes");
348
349        let (state_service, read_only_state_service, latest_chain_tip, chain_tip_change) =
350            zebra_state::init(
351                config.state.clone(),
352                &config.network.network,
353                max_checkpoint_height,
354                config.sync.checkpoint_verify_concurrency_limit
355                    * (VERIFICATION_PIPELINE_SCALING_MULTIPLIER + 1),
356            )
357            .await;
358
359        info!("logging database metrics on startup");
360        read_only_state_service.log_db_metrics();
361
362        let state = ServiceBuilder::new()
363            .buffer(Self::state_buffer_bound())
364            .service(state_service);
365
366        info!("initializing network");
367        // The service that our node uses to respond to requests by peers. The
368        // load_shed middleware ensures that we reduce the size of the peer set
369        // in response to excess load.
370        //
371        // # Security
372        //
373        // This layer stack is security-sensitive, modifying it can cause hangs,
374        // or enable denial of service attacks.
375        //
376        // See `zebra_network::Connection::drive_peer_request()` for details.
377        let (setup_tx, setup_rx) = oneshot::channel();
378        let inbound = ServiceBuilder::new()
379            .load_shed()
380            .buffer(inbound::downloads::MAX_INBOUND_CONCURRENCY)
381            .timeout(MAX_INBOUND_RESPONSE_TIME)
382            .service(Inbound::new(
383                config.sync.full_verify_concurrency_limit,
384                setup_rx,
385            ));
386
387        let (peer_set, address_book, misbehavior_sender) =
388            zebra_network::init_with_block_gossip_peer_ips(
389                config.network.clone(),
390                inbound,
391                latest_chain_tip.clone(),
392                user_agent(),
393                zcashd_compat_block_gossip_peer_ips,
394            )
395            .await;
396
397        // Start health server if configured (after sync_status is available)
398
399        info!("initializing verifiers");
400        let (tx_verifier_setup_tx, tx_verifier_setup_rx) = oneshot::channel();
401        let (block_verifier_router, tx_verifier, consensus_task_handles, max_checkpoint_height) =
402            zebra_consensus::router::init(
403                config.consensus.clone(),
404                &config.network.network,
405                state.clone(),
406                tx_verifier_setup_rx,
407            )
408            .await;
409
410        info!("initializing syncer");
411        let (mut syncer, sync_status) = ChainSync::new(
412            &config,
413            max_checkpoint_height,
414            peer_set.clone(),
415            block_verifier_router.clone(),
416            state.clone(),
417            read_only_state_service.clone(),
418            latest_chain_tip.clone(),
419            misbehavior_sender.clone(),
420        );
421
422        info!("initializing mempool");
423        let (mempool, mempool_transaction_subscriber) = Mempool::new(
424            &config.network.network,
425            &config.mempool,
426            peer_set.background_service(),
427            state.clone(),
428            tx_verifier,
429            sync_status.clone(),
430            latest_chain_tip.clone(),
431            chain_tip_change.clone(),
432            misbehavior_sender.clone(),
433        );
434
435        // Subscribe as soon as possible to not miss any events
436        let mempool_change_receiver = mempool_transaction_subscriber.subscribe();
437
438        let mempool = BoxService::new(mempool);
439        let mempool = ServiceBuilder::new()
440            .buffer(mempool::downloads::MAX_INBOUND_CONCURRENCY)
441            .service(mempool);
442
443        if tx_verifier_setup_tx.send(mempool.clone()).is_err() {
444            warn!("error setting up the transaction verifier with a handle to the mempool service");
445        };
446
447        info!("fully initializing inbound peer request handler");
448        // Fully start the inbound service as soon as possible
449        let setup_data = InboundSetupData {
450            address_book: address_book.clone(),
451            block_download_peer_set: peer_set.background_service(),
452            block_verifier: block_verifier_router.clone(),
453            mempool: mempool.clone(),
454            state: state.clone(),
455            latest_chain_tip: latest_chain_tip.clone(),
456            misbehavior_sender,
457        };
458        setup_tx
459            .send(setup_data)
460            .map_err(|_| eyre!("could not send setup data to inbound service"))?;
461        // And give it time to clear its queue
462        tokio::task::yield_now().await;
463
464        // Create a channel to send mined blocks to the gossip task
465        let submit_block_channel = SubmitBlockChannel::new();
466
467        // Launch RPC server
468        let (rpc_impl, mut rpc_tx_queue_handle) = RpcImpl::new(
469            config.network.network.clone(),
470            config.mining.clone(),
471            config.rpc.debug_force_finished_sync,
472            build_version(),
473            user_agent(),
474            mempool.clone(),
475            state.clone(),
476            read_only_state_service.clone(),
477            block_verifier_router.clone(),
478            sync_status.clone(),
479            latest_chain_tip.clone(),
480            address_book.clone(),
481            LAST_WARN_ERROR_LOG_SENDER.subscribe(),
482            Some(submit_block_channel.sender()),
483        );
484        let rpc_impl = rpc_impl.with_end_of_support_height(
485            sync::end_of_support::end_of_support_height(&config.network.network),
486        );
487
488        // Keep a block template ready for the `getblocktemplate` RPC, if this node is configured
489        // for mining, and something can actually ask for a template. Without the RPC server and
490        // without the internal miner, nothing can call `getblocktemplate`, so precomputing
491        // templates would build coinbase transactions that no one reads.
492        #[cfg(feature = "internal-miner")]
493        let is_internal_miner_enabled = config.mining.is_internal_miner_enabled();
494        #[cfg(not(feature = "internal-miner"))]
495        let is_internal_miner_enabled = false;
496
497        let block_template_task_handle =
498            if config.rpc.listen_addr.is_some() || is_internal_miner_enabled {
499                rpc_impl
500                    .spawn_block_template_updater()
501                    .inspect(|_| info!("spawned block template updater task"))
502            } else {
503                None
504            };
505
506        // Supervise the updater like every other ongoing task: if it exits or panics, the RPC
507        // keeps serving the last template it published, and pays the new-tip timeout on every call
508        // after the next tip change, so a silent exit has to be visible.
509        let block_template_task_handle: tokio::task::JoinHandle<()> = block_template_task_handle
510            .unwrap_or_else(|| tokio::spawn(std::future::pending().in_current_span()));
511
512        let rpc_task_handle = if config.rpc.listen_addr.is_some() {
513            RpcServer::start(rpc_impl.clone(), config.rpc.clone())
514                .await
515                .expect("server should start")
516        } else {
517            tokio::spawn(std::future::pending().in_current_span())
518        };
519
520        let zcashd_compat_shutdown_timeout =
521            Self::zcashd_compat_supervisor_shutdown_timeout(&config);
522        let (zcashd_compat_shutdown_tx, zcashd_compat_shutdown_rx) = watch::channel(false);
523        let mut zcashd_compat_task_handle = if let Some(resolved_zcashd_path) = resolved_zcashd_path
524        {
525            let local_listener = address_book
526                .lock()
527                .expect("unexpected panic in address book mutex guard")
528                .local_listener_socket_addr();
529            let supervisor_config = zcashd_compat::SupervisorConfig::new(
530                &config.zcashd_compat,
531                resolved_zcashd_path,
532                &config.state.cache_dir,
533                config.network.network.kind(),
534                Self::zcashd_compat_p2p_connect_addr(&config, local_listener),
535            );
536
537            info!(
538                connect = %supervisor_config.zebra_p2p_addr,
539                "zcashd-compat mode enabled"
540            );
541
542            tokio::spawn(
543                zcashd_compat::run_supervisor(supervisor_config, zcashd_compat_shutdown_rx)
544                    .in_current_span(),
545            )
546        } else {
547            if config.zcashd_compat.enabled {
548                zcashd_compat::set_supervision_config_disabled_metrics();
549                info!("zcashd-compat mode enabled: zcashd supervision disabled");
550            }
551
552            tokio::spawn(std::future::pending().in_current_span())
553        };
554
555        // TODO: Add a shutdown signal and start the server with `serve_with_incoming_shutdown()` if
556        //       any related unit tests sometimes crash with memory errors
557        let indexer_rpc_task_handle = {
558            if let Some(indexer_listen_addr) = config.rpc.indexer_listen_addr {
559                info!("spawning indexer RPC server");
560                let (indexer_rpc_task_handle, _listen_addr) = zebra_rpc::indexer::server::init(
561                    indexer_listen_addr,
562                    read_only_state_service.clone(),
563                    latest_chain_tip.clone(),
564                    mempool_transaction_subscriber.clone(),
565                )
566                .await
567                .map_err(|err| eyre!(err))?;
568
569                indexer_rpc_task_handle
570            } else {
571                warn!("configure an indexer_listen_addr to start the indexer RPC server");
572                tokio::spawn(std::future::pending().in_current_span())
573            }
574        };
575
576        let lightwalletd_rpc_task_handle = {
577            if let Some(lightwalletd_listen_addr) = config.rpc.lightwalletd_listen_addr {
578                info!("spawning lightwalletd gRPC server");
579                let (lightwalletd_rpc_task_handle, _listen_addr) =
580                    zebra_rpc::lightwalletd::server::init(
581                        lightwalletd_listen_addr,
582                        rpc_impl.clone(),
583                        read_only_state_service.clone(),
584                        mempool.clone(),
585                        latest_chain_tip.clone(),
586                        mempool_transaction_subscriber.clone(),
587                        config.network.network.clone(),
588                    )
589                    .await
590                    .map_err(|err| eyre!(err))?;
591
592                lightwalletd_rpc_task_handle
593            } else {
594                tokio::spawn(std::future::pending().in_current_span())
595            }
596        };
597
598        // Start concurrent tasks which don't add load to other tasks
599        info!("spawning block gossip task");
600        let block_gossip_task_handle = tokio::spawn(
601            sync::gossip_best_tip_block_hashes(
602                sync_status.clone(),
603                chain_tip_change.clone(),
604                peer_set.clone(),
605                Some(submit_block_channel.receiver()),
606            )
607            .in_current_span(),
608        );
609
610        info!("spawning block notify task");
611        let block_notify_task_handle: tokio::task::JoinHandle<Result<(), BlockNotifyError>> =
612            if let Some(command) = config.notify.block_notify_command.clone() {
613                tokio::spawn(
614                    notify::run_block_notify(
615                        command,
616                        sync_status.clone(),
617                        chain_tip_change.clone(),
618                    )
619                    .in_current_span(),
620                )
621            } else {
622                tokio::spawn(std::future::pending().in_current_span())
623            };
624
625        info!("spawning mempool queue checker task");
626        let mempool_queue_checker_task_handle = mempool::QueueChecker::spawn(mempool.clone());
627
628        info!("spawning mempool transaction gossip task");
629        let tx_gossip_task_handle = tokio::spawn(
630            mempool::gossip_mempool_transaction_id(mempool_change_receiver, peer_set.clone())
631                .in_current_span(),
632        );
633
634        info!("spawning delete old databases task");
635        let mut old_databases_task_handle = zebra_state::check_and_delete_old_state_databases(
636            &config.state,
637            &config.network.network,
638        );
639
640        info!("spawning progress logging task");
641        let (chain_tip_metrics_sender, chain_tip_metrics_receiver) =
642            health::ChainTipMetrics::channel();
643        let progress_task_handle = tokio::spawn(
644            show_block_chain_progress(
645                config.network.network.clone(),
646                latest_chain_tip.clone(),
647                sync_status.clone(),
648                chain_tip_metrics_sender,
649            )
650            .in_current_span(),
651        );
652
653        // Start health server if configured
654        info!("initializing health endpoints");
655        let (health_task_handle, _) = health::init(
656            config.health.clone(),
657            config.network.network.clone(),
658            chain_tip_metrics_receiver,
659            sync_status.clone(),
660            address_book.clone(),
661        )
662        .await;
663
664        // Spawn never ending end of support task.
665        info!("spawning end of support checking task");
666        let end_of_support_task_handle = tokio::spawn(
667            sync::end_of_support::start(config.network.network.clone(), latest_chain_tip.clone())
668                .in_current_span(),
669        );
670
671        // Give the inbound service more time to clear its queue,
672        // then start concurrent tasks that can add load to the inbound service
673        // (by opening more peer connections, so those peers send us requests)
674        tokio::task::yield_now().await;
675
676        // The crawler only activates immediately in tests that use mempool debug mode
677        info!("spawning mempool crawler task");
678        let mempool_crawler_task_handle = mempool::Crawler::spawn(
679            &config.mempool,
680            peer_set,
681            mempool.clone(),
682            sync_status.clone(),
683            chain_tip_change.clone(),
684        );
685
686        info!("spawning syncer task");
687        // In regtest, commit the genesis block directly (bypassing the syncer's genesis
688        // download, which requires a connected peer). Then run the syncer normally so
689        // that multi-hop block propagation works: gossiped blocks that arrive out of
690        // order (e.g. only the latest tip hash was gossiped) will be recovered by the
691        // syncer using block locators within REGTEST_SYNC_RESTART_DELAY (2 seconds).
692        if is_regtest
693            && !syncer
694                .state_contains(config.network.network.genesis_hash())
695                .await?
696        {
697            let genesis_hash = block_verifier_router
698                .clone()
699                .oneshot(zebra_consensus::Request::Commit(regtest_genesis_block()))
700                .await
701                .expect("should validate Regtest genesis block");
702
703            assert_eq!(
704                genesis_hash,
705                config.network.network.genesis_hash(),
706                "validated block hash should match network genesis hash"
707            )
708        }
709        let syncer_task_handle = tokio::spawn(syncer.sync().in_current_span());
710
711        // And finally, spawn the internal Zcash miner, if it is enabled.
712        //
713        // TODO: add a config to enable the miner rather than a feature.
714        #[cfg(feature = "internal-miner")]
715        let miner_task_handle = if config.mining.is_internal_miner_enabled() {
716            info!("spawning Zcash miner");
717            components::miner::spawn_init(&config.metrics, rpc_impl)
718        } else {
719            tokio::spawn(std::future::pending().in_current_span())
720        };
721
722        #[cfg(not(feature = "internal-miner"))]
723        // Spawn a dummy miner task which doesn't do anything and never finishes.
724        let miner_task_handle: tokio::task::JoinHandle<Result<(), Report>> =
725            tokio::spawn(std::future::pending().in_current_span());
726
727        info!("spawned initial Zebra tasks");
728
729        // TODO: put tasks into an ongoing FuturesUnordered and a startup FuturesUnordered?
730
731        // ongoing tasks
732        pin!(rpc_task_handle);
733        pin!(indexer_rpc_task_handle);
734        pin!(lightwalletd_rpc_task_handle);
735        pin!(syncer_task_handle);
736        pin!(block_gossip_task_handle);
737        pin!(block_notify_task_handle);
738        pin!(mempool_crawler_task_handle);
739        pin!(mempool_queue_checker_task_handle);
740        pin!(tx_gossip_task_handle);
741        pin!(progress_task_handle);
742        pin!(end_of_support_task_handle);
743        pin!(miner_task_handle);
744        pin!(block_template_task_handle);
745
746        // startup tasks
747        let BackgroundTaskHandles {
748            mut state_checkpoint_verify_handle,
749        } = consensus_task_handles;
750
751        let state_checkpoint_verify_handle_fused = (&mut state_checkpoint_verify_handle).fuse();
752        pin!(state_checkpoint_verify_handle_fused);
753
754        let old_databases_task_handle_fused = (&mut old_databases_task_handle).fuse();
755        pin!(old_databases_task_handle_fused);
756
757        // The zcashd-compat supervisor exits when supervision is disabled or fails,
758        // but Zebra keeps running, so its handle must be fused.
759        let mut zcashd_compat_task_finished = false;
760        let zcashd_compat_task_handle_fused = (&mut zcashd_compat_task_handle).fuse();
761        pin!(zcashd_compat_task_handle_fused);
762
763        // Wait for tasks to finish
764        let exit_status = loop {
765            let mut exit_when_task_finishes = true;
766
767            let result = select! {
768                rpc_join_result = &mut rpc_task_handle => {
769                    let rpc_server_result = rpc_join_result
770                        .expect("unexpected panic in the rpc task");
771                    info!(?rpc_server_result, "rpc task exited");
772                    Ok(())
773                }
774
775                rpc_tx_queue_result = &mut rpc_tx_queue_handle => {
776                    rpc_tx_queue_result
777                        .expect("unexpected panic in the rpc transaction queue task");
778                    info!("rpc transaction queue task exited");
779                    Ok(())
780                }
781
782                indexer_rpc_join_result = &mut indexer_rpc_task_handle => {
783                    let indexer_rpc_server_result = indexer_rpc_join_result
784                        .expect("unexpected panic in the indexer task");
785                    info!(?indexer_rpc_server_result, "indexer rpc task exited");
786                    Ok(())
787                }
788
789                lightwalletd_rpc_join_result = &mut lightwalletd_rpc_task_handle => {
790                    let lightwalletd_rpc_server_result = lightwalletd_rpc_join_result
791                        .expect("unexpected panic in the lightwalletd gRPC task");
792                    info!(?lightwalletd_rpc_server_result, "lightwalletd gRPC task exited");
793                    Ok(())
794                }
795
796                sync_result = &mut syncer_task_handle => sync_result
797                    .expect("unexpected panic in the syncer task")
798                    .map(|_| info!("syncer task exited")),
799
800                block_gossip_result = &mut block_gossip_task_handle => block_gossip_result
801                    .expect("unexpected panic in the chain tip block gossip task")
802                    .map(|_| info!("chain tip block gossip task exited"))
803                    .map_err(|e| eyre!(e)),
804
805                block_notify_result = &mut block_notify_task_handle => block_notify_result
806                    .expect("unexpected panic in the block notify task")
807                    .map(|_| info!("block notify task exited"))
808                    .map_err(|e| eyre!(e)),
809
810                mempool_crawl_result = &mut mempool_crawler_task_handle => mempool_crawl_result
811                    .expect("unexpected panic in the mempool crawler")
812                    .map(|_| info!("mempool crawler task exited"))
813                    .map_err(|e| eyre!(e)),
814
815                mempool_queue_result = &mut mempool_queue_checker_task_handle => mempool_queue_result
816                    .expect("unexpected panic in the mempool queue checker")
817                    .map(|_| info!("mempool queue checker task exited"))
818                    .map_err(|e| eyre!(e)),
819
820                tx_gossip_result = &mut tx_gossip_task_handle => tx_gossip_result
821                    .expect("unexpected panic in the transaction gossip task")
822                    .map(|_| info!("transaction gossip task exited"))
823                    .map_err(|e| eyre!(e)),
824
825                // The progress task runs forever, unless it panics.
826                // So we don't need to provide an exit status for it.
827                progress_result = &mut progress_task_handle => {
828                    info!("chain progress task exited");
829                    progress_result
830                        .expect("unexpected panic in the chain progress task");
831                }
832
833                end_of_support_result = &mut end_of_support_task_handle => end_of_support_result
834                    .expect("unexpected panic in the end of support task")
835                    .map(|_| info!("end of support task exited")),
836
837                // We also expect the state checkpoint verify task to finish.
838                state_checkpoint_verify_result = &mut state_checkpoint_verify_handle_fused => {
839                    state_checkpoint_verify_result
840                        .unwrap_or_else(|_| panic!(
841                            "unexpected panic checking previous state followed the best chain"));
842
843                    exit_when_task_finishes = false;
844                    Ok(())
845                }
846
847                // And the old databases task should finish while Zebra is running.
848                old_databases_result = &mut old_databases_task_handle_fused => {
849                    old_databases_result
850                        .unwrap_or_else(|_| panic!(
851                            "unexpected panic deleting old database directories"));
852
853                    exit_when_task_finishes = false;
854                    Ok(())
855                }
856
857                block_template_result = &mut block_template_task_handle => {
858                    block_template_result
859                        .expect("unexpected panic in the block template updater task");
860                    info!("block template updater task exited");
861
862                    Ok(())
863                }
864
865                miner_result = &mut miner_task_handle => miner_result
866                    .expect("unexpected panic in the miner task")
867                    .map(|_| info!("miner task exited")),
868
869                zcashd_compat_result = &mut zcashd_compat_task_handle_fused => {
870                    zcashd_compat_task_finished = true;
871                    exit_when_task_finishes =
872                        Self::zcashd_compat_supervisor_should_exit(zcashd_compat_result);
873                    Ok(())
874                },
875            };
876
877            // Stop Zebra if a task finished and returned an error,
878            // or if an ongoing task exited.
879            if let Err(err) = result {
880                break Err(err);
881            }
882
883            if exit_when_task_finishes {
884                break Ok(());
885            }
886        };
887
888        info!("exiting Zebra because an ongoing task exited: asking other tasks to stop");
889
890        // ongoing tasks
891        rpc_task_handle.abort();
892        rpc_tx_queue_handle.abort();
893        block_template_task_handle.abort();
894        health_task_handle.abort();
895        syncer_task_handle.abort();
896        block_gossip_task_handle.abort();
897        block_notify_task_handle.abort();
898        mempool_crawler_task_handle.abort();
899        mempool_queue_checker_task_handle.abort();
900        tx_gossip_task_handle.abort();
901        progress_task_handle.abort();
902        end_of_support_task_handle.abort();
903        miner_task_handle.abort();
904        if zcashd_compat_task_finished {
905            debug!("zcashd-compat supervisor task already exited before shutdown");
906        } else if let Some(zcashd_compat_shutdown_timeout) = zcashd_compat_shutdown_timeout {
907            info!(
908                ?zcashd_compat_shutdown_timeout,
909                "requesting zcashd-compat supervisor shutdown"
910            );
911            if zcashd_compat_shutdown_tx.send(true).is_err() {
912                warn!("zcashd-compat supervisor shutdown request was not delivered");
913            }
914            if tokio::time::timeout(
915                zcashd_compat_shutdown_timeout,
916                &mut zcashd_compat_task_handle,
917            )
918            .await
919            .is_err()
920            {
921                warn!(
922                    ?zcashd_compat_shutdown_timeout,
923                    "zcashd-compat supervisor did not finish before shutdown timeout; \
924                     abandoning child process handle"
925                );
926                // The supervisor spawns zcashd without kill_on_drop, so this
927                // abort abandons an already-signalled child rather than
928                // SIGKILLing it mid-flush.
929                zcashd_compat_task_handle.abort();
930            }
931        } else {
932            debug!("aborting zcashd-compat supervisor task without managed child shutdown");
933            zcashd_compat_task_handle.abort();
934        }
935
936        // startup tasks
937        state_checkpoint_verify_handle.abort();
938        old_databases_task_handle.abort();
939
940        info!(
941            "exiting Zebra: all tasks have been asked to stop, waiting for remaining tasks to finish"
942        );
943
944        exit_status
945    }
946
947    /// Returns the bound for the state service buffer,
948    /// based on the configurations of the services that use the state concurrently.
949    fn state_buffer_bound() -> usize {
950        let config = APPLICATION.config();
951
952        // Ignore the checkpoint verify limit, because it is very large.
953        //
954        // TODO: do we also need to account for concurrent use across services?
955        //       we could multiply the maximum by 3/2, or add a fixed constant
956        [
957            config.sync.download_concurrency_limit,
958            config.sync.full_verify_concurrency_limit,
959            inbound::downloads::MAX_INBOUND_CONCURRENCY,
960            mempool::downloads::MAX_INBOUND_CONCURRENCY,
961        ]
962        .into_iter()
963        .max()
964        .unwrap()
965    }
966}
967
968impl Runnable for StartCmd {
969    /// Start the application.
970    fn run(&self) {
971        info!("Starting zebrad");
972        let rt = APPLICATION
973            .state()
974            .components_mut()
975            .get_downcast_mut::<TokioComponent>()
976            .expect("TokioComponent should be available")
977            .rt
978            .take();
979
980        rt.expect("runtime should not already be taken")
981            .run(self.start());
982
983        info!("stopping zebrad");
984    }
985}
986
987impl config::Override<ZebradConfig> for StartCmd {
988    // Process the given command line options, overriding settings from
989    // a configuration file using explicit flags taken from command-line
990    // arguments.
991    fn override_config(&self, mut config: ZebradConfig) -> Result<ZebradConfig, FrameworkError> {
992        if !self.filters.is_empty() {
993            config.tracing.filter = Some(self.filters.join(","));
994        }
995
996        // `--zcashd-compat` is a one-way override that enables zcashd-compat mode.
997        // The actual zcashd-compat guardrails are applied below using
998        // `config.zcashd_compat.enabled` so CLI and config-file activation share one path.
999        if self.zcashd_compat {
1000            config.zcashd_compat.enabled = true;
1001        }
1002
1003        if !config.zcashd_compat.enabled && !config.zcashd_compat.block_gossip_peer_ips.is_empty() {
1004            return Err(std::io::Error::other(
1005                "zcashd_compat.block_gossip_peer_ips requires zcashd_compat.enabled = true",
1006            )
1007            .into());
1008        }
1009
1010        if config.zcashd_compat.enabled && config.zcashd_compat.manage_zcashd {
1011            zcashd_compat::reject_peer_selection_extra_args(
1012                &config.zcashd_compat.zcashd_extra_args,
1013            )
1014            .map_err(|err| std::io::Error::other(err.to_string()))?;
1015
1016            match zcashd_compat::effective_zcashd_source(&config.zcashd_compat) {
1017                Ok(zcashd_compat::ZcashdBinarySource::Path(path))
1018                    if !zcashd_compat::is_command_resolvable(Path::new(&path)) =>
1019                {
1020                    return Err(std::io::Error::other(format!(
1021                        "zcashd-compat mode could not resolve zcashd_path={}",
1022                        path.display()
1023                    ))
1024                    .into());
1025                }
1026                Ok(_) => {}
1027                Err(err) => return Err(std::io::Error::other(err.to_string()).into()),
1028            }
1029        }
1030
1031        Ok(config)
1032    }
1033}
1034
1035#[cfg(test)]
1036mod tests {
1037    use abscissa_core::config::Override;
1038    use color_eyre::eyre::eyre;
1039
1040    use super::StartCmd;
1041    use crate::components::zcashd_compat;
1042    use crate::config::ZebradConfig;
1043
1044    #[test]
1045    fn zcashd_compat_flag_enables_mode() {
1046        let cmd = StartCmd {
1047            filters: Vec::new(),
1048            zcashd_compat: true,
1049            unsafe_low_specs: false,
1050        };
1051        let mut config = ZebradConfig::default();
1052        config.zcashd_compat.manage_zcashd = false;
1053
1054        let config = cmd
1055            .override_config(config)
1056            .expect("zcashd-compat override config should succeed");
1057
1058        assert!(config.zcashd_compat.enabled);
1059    }
1060
1061    #[test]
1062    fn zcashd_compat_config_enables_mode() {
1063        let cmd = StartCmd {
1064            filters: Vec::new(),
1065            zcashd_compat: false,
1066            unsafe_low_specs: false,
1067        };
1068        let mut config = ZebradConfig::default();
1069        config.zcashd_compat.enabled = true;
1070        config.zcashd_compat.manage_zcashd = false;
1071
1072        let config = cmd
1073            .override_config(config)
1074            .expect("zcashd-compat override config should succeed");
1075
1076        assert!(config.zcashd_compat.enabled);
1077    }
1078
1079    #[test]
1080    fn block_gossip_peer_ips_require_zcashd_compat() {
1081        let cmd = StartCmd {
1082            filters: Vec::new(),
1083            zcashd_compat: false,
1084            unsafe_low_specs: false,
1085        };
1086        let mut config = ZebradConfig::default();
1087        config.zcashd_compat.block_gossip_peer_ips =
1088            vec![std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST)];
1089
1090        let error = cmd
1091            .override_config(config)
1092            .expect_err("block gossip peers should require zcashd-compat");
1093
1094        assert!(
1095            error
1096                .to_string()
1097                .contains("zcashd_compat.block_gossip_peer_ips requires"),
1098            "error should explain the zcashd-compat requirement: {error}"
1099        );
1100    }
1101
1102    #[test]
1103    fn zcashd_compat_config_rejects_peer_selection_extra_args() {
1104        let cmd = StartCmd {
1105            filters: Vec::new(),
1106            zcashd_compat: false,
1107            unsafe_low_specs: false,
1108        };
1109        let mut config = ZebradConfig::default();
1110        config.zcashd_compat.enabled = true;
1111        config.zcashd_compat.manage_zcashd = true;
1112        config.zcashd_compat.zcashd_source = zcashd_compat::ConfigZcashdBinarySource::Embedded;
1113        config.zcashd_compat.zcashd_extra_args = vec!["-addnode=1.2.3.4".to_string()];
1114
1115        let error = cmd
1116            .override_config(config)
1117            .expect_err("peer-selection extra args should be rejected");
1118        assert!(
1119            error.to_string().contains("peer-selection"),
1120            "unexpected error: {error}"
1121        );
1122    }
1123
1124    #[test]
1125    fn zcashd_compat_manage_zcashd_requires_resolvable_path() {
1126        let cmd = StartCmd {
1127            filters: Vec::new(),
1128            zcashd_compat: true,
1129            unsafe_low_specs: false,
1130        };
1131        let mut config = ZebradConfig::default();
1132        config.zcashd_compat.manage_zcashd = true;
1133        config.zcashd_compat.zcashd_path = Some("/definitely/missing/zcashd-compat".into());
1134
1135        let error = cmd
1136            .override_config(config)
1137            .expect_err("zcashd-compat override should fail for an unresolvable zcashd path");
1138
1139        assert!(
1140            error
1141                .to_string()
1142                .contains("zcashd-compat mode could not resolve zcashd_path"),
1143            "unexpected error: {error}"
1144        );
1145    }
1146
1147    #[test]
1148    fn zcashd_compat_path_source_requires_explicit_path() {
1149        let cmd = StartCmd {
1150            filters: Vec::new(),
1151            zcashd_compat: true,
1152            unsafe_low_specs: false,
1153        };
1154        let mut config = ZebradConfig::default();
1155        config.zcashd_compat.manage_zcashd = true;
1156        config.zcashd_compat.zcashd_source = zcashd_compat::ConfigZcashdBinarySource::Path;
1157        config.zcashd_compat.zcashd_path = None;
1158
1159        let error = cmd
1160            .override_config(config)
1161            .expect_err("path source should require explicit zcashd_path");
1162        assert!(
1163            error.to_string().contains("zcashd_source=path"),
1164            "unexpected error: {error}"
1165        );
1166    }
1167
1168    #[test]
1169    fn zcashd_compat_embedded_source_allows_missing_local_path() {
1170        let cmd = StartCmd {
1171            filters: Vec::new(),
1172            zcashd_compat: true,
1173            unsafe_low_specs: false,
1174        };
1175        let mut config = ZebradConfig::default();
1176        config.zcashd_compat.manage_zcashd = true;
1177        config.zcashd_compat.zcashd_source = zcashd_compat::ConfigZcashdBinarySource::Embedded;
1178        config.zcashd_compat.zcashd_path = None;
1179
1180        cmd.override_config(config)
1181            .expect("embedded source should be validated at runtime, not override-time");
1182    }
1183
1184    #[test]
1185    fn zcashd_compat_config_manage_zcashd_requires_resolvable_path() {
1186        let cmd = StartCmd {
1187            filters: Vec::new(),
1188            zcashd_compat: false,
1189            unsafe_low_specs: false,
1190        };
1191        let mut config = ZebradConfig::default();
1192        config.zcashd_compat.enabled = true;
1193        config.zcashd_compat.manage_zcashd = true;
1194        config.zcashd_compat.zcashd_path = Some("/definitely/missing/zcashd-compat".into());
1195
1196        let error = cmd
1197            .override_config(config)
1198            .expect_err("zcashd-compat config should fail for an unresolvable zcashd path");
1199
1200        assert!(
1201            error
1202                .to_string()
1203                .contains("zcashd-compat mode could not resolve zcashd_path"),
1204            "unexpected error: {error}"
1205        );
1206    }
1207
1208    #[test]
1209    fn zcashd_compat_supervisor_shutdown_timeout_matches_config() {
1210        let mut config = ZebradConfig::default();
1211
1212        config.zcashd_compat.enabled = true;
1213        config.zcashd_compat.manage_zcashd = true;
1214        config.zcashd_compat.shutdown_grace_period = std::time::Duration::from_secs(42);
1215        assert_eq!(
1216            StartCmd::zcashd_compat_supervisor_shutdown_timeout(&config),
1217            Some(
1218                std::time::Duration::from_secs(42)
1219                    + StartCmd::ZCASHD_COMPAT_SHUTDOWN_TIMEOUT_MARGIN
1220            ),
1221            "outer supervisor wait must exceed the child grace period so task \
1222             abort cannot preempt graceful termination",
1223        );
1224
1225        config.zcashd_compat.manage_zcashd = false;
1226        assert_eq!(
1227            StartCmd::zcashd_compat_supervisor_shutdown_timeout(&config),
1228            None
1229        );
1230
1231        config.zcashd_compat.enabled = false;
1232        config.zcashd_compat.manage_zcashd = true;
1233        assert_eq!(
1234            StartCmd::zcashd_compat_supervisor_shutdown_timeout(&config),
1235            None
1236        );
1237    }
1238
1239    #[test]
1240    fn zcashd_compat_supervisor_ok_exit_does_not_exit_zebra() {
1241        assert!(!StartCmd::zcashd_compat_supervisor_should_exit(Ok(Ok(()))));
1242    }
1243
1244    #[test]
1245    fn zcashd_compat_supervisor_error_does_not_exit_zebra() {
1246        assert!(!StartCmd::zcashd_compat_supervisor_should_exit(Ok(Err(
1247            eyre!("simulated zcashd supervisor runtime failure"),
1248        ))));
1249    }
1250
1251    #[tokio::test]
1252    async fn zcashd_compat_supervisor_panic_does_not_exit_zebra() {
1253        let join_err = tokio::spawn(async {
1254            panic!("simulated zcashd supervisor panic");
1255        })
1256        .await
1257        .expect_err("task should panic");
1258
1259        assert!(!StartCmd::zcashd_compat_supervisor_should_exit(Err(
1260            join_err
1261        )));
1262    }
1263}