Skip to main content

zebrad/components/
sync.rs

1//! The syncer downloads and verifies large numbers of blocks from peers to Zebra.
2//!
3//! It is used when Zebra is a long way behind the current chain tip.
4
5use std::{
6    cmp::max,
7    collections::{HashMap, HashSet},
8    convert,
9    pin::Pin,
10    task::Poll,
11    time::Duration,
12};
13
14use color_eyre::eyre::{eyre, Report};
15use futures::stream::{FuturesUnordered, StreamExt};
16use indexmap::IndexSet;
17use serde::{Deserialize, Serialize};
18use tokio::{
19    sync::{mpsc, watch},
20    task::JoinError,
21    time::{sleep, timeout},
22};
23use tower::{
24    builder::ServiceBuilder, hedge::Hedge, limit::ConcurrencyLimit, retry::Retry, timeout::Timeout,
25    Service, ServiceExt,
26};
27
28use zebra_chain::{
29    block::{self, Height, HeightDiff},
30    chain_tip::ChainTip,
31};
32use zebra_network::{self as zn, PeerSocketAddr};
33use zebra_state as zs;
34
35use crate::{
36    components::sync::downloads::BlockDownloadVerifyError, config::ZebradConfig, BoxError,
37};
38
39mod downloads;
40pub mod end_of_support;
41mod gossip;
42mod progress;
43mod recent_sync_lengths;
44mod status;
45
46#[cfg(test)]
47mod tests;
48
49use downloads::{AlwaysHedge, Downloads};
50
51pub use downloads::VERIFICATION_PIPELINE_SCALING_MULTIPLIER;
52pub use gossip::{gossip_best_tip_block_hashes, BlockGossipError};
53pub use progress::show_block_chain_progress;
54pub use recent_sync_lengths::RecentSyncLengths;
55pub use status::SyncStatus;
56
57/// Controls the number of peers used for each ObtainTips and ExtendTips request.
58const FANOUT: usize = 3;
59
60/// Controls how many times we will retry each block download.
61///
62/// Failing block downloads is important because it defends against peers who
63/// feed us bad hashes. But spurious failures of valid blocks cause the syncer to
64/// restart from the previous checkpoint, potentially re-downloading blocks.
65///
66/// We also hedge requests, so we may retry up to twice this many times. Hedged
67/// retries may be concurrent, inner retries are sequential.
68const BLOCK_DOWNLOAD_RETRY_LIMIT: usize = 3;
69
70/// Controls how many times the syncer will re-request a block whose download
71/// failed because no peer delivered it (a `NotFound`), before giving up and
72/// letting the normal tip re-walk handle it.
73///
74/// Without this re-request, a single missing block at the checkpoint frontier
75/// is dropped and never re-fetched, wedging the whole verify pipeline until the
76/// 8-minute `BLOCK_VERIFY_TIMEOUT` fires (#5709). Each attempt already goes
77/// through the tower-level `BLOCK_DOWNLOAD_RETRY_LIMIT` (and hedging), so this
78/// is a coarse, hash-scoped retry on top of an exhausted per-request retry.
79const MAX_BLOCK_REOBTAIN_RETRIES: u8 = 3;
80
81/// A lower bound on the user-specified checkpoint verification concurrency limit.
82///
83/// Set to the maximum checkpoint interval, so the pipeline holds around a checkpoint's
84/// worth of blocks.
85///
86/// ## Security
87///
88/// If a malicious node is chosen for an ObtainTips or ExtendTips request, it can
89/// provide up to 500 malicious block hashes. These block hashes will be
90/// distributed across all available peers. Assuming there are around 50 connected
91/// peers, the malicious node will receive approximately 10 of those block requests.
92///
93/// Malicious deserialized blocks can take up a large amount of RAM, see
94/// [`super::inbound::downloads::MAX_INBOUND_CONCURRENCY`] and #1880 for details.
95/// So we want to keep the lookahead limit reasonably small.
96///
97/// Once these malicious blocks start failing validation, the syncer will cancel all
98/// the pending download and verify tasks, drop all the blocks, and start a new
99/// ObtainTips with a new set of peers.
100pub const MIN_CHECKPOINT_CONCURRENCY_LIMIT: usize = zebra_consensus::MAX_CHECKPOINT_HEIGHT_GAP;
101
102/// The default for the user-specified lookahead limit.
103///
104/// See [`MIN_CHECKPOINT_CONCURRENCY_LIMIT`] for details.
105pub const DEFAULT_CHECKPOINT_CONCURRENCY_LIMIT: usize = MAX_TIPS_RESPONSE_HASH_COUNT * 2;
106
107/// A lower bound on the user-specified concurrency limit.
108///
109/// If the concurrency limit is 0, Zebra can't download or verify any blocks.
110pub const MIN_CONCURRENCY_LIMIT: usize = 1;
111
112/// The expected maximum number of hashes in an ObtainTips or ExtendTips response.
113///
114/// This is used to allow block heights that are slightly beyond the lookahead limit,
115/// but still limit the number of blocks in the pipeline between the downloader and
116/// the state.
117///
118/// See [`MIN_CHECKPOINT_CONCURRENCY_LIMIT`] for details.
119pub const MAX_TIPS_RESPONSE_HASH_COUNT: usize = 500;
120
121/// Controls how long we wait for a tips response to return.
122///
123/// ## Correctness
124///
125/// If this timeout is removed (or set too high), the syncer will sometimes hang.
126///
127/// If this timeout is set too low, the syncer will sometimes get stuck in a
128/// failure loop.
129pub const TIPS_RESPONSE_TIMEOUT: Duration = Duration::from_secs(6);
130
131/// Controls how long we wait between gossiping successive blocks or transactions.
132///
133/// ## Correctness
134///
135/// If this timeout is set too high, blocks and transactions won't propagate through
136/// the network efficiently.
137///
138/// If this timeout is set too low, the peer set and remote peers can get overloaded.
139pub const PEER_GOSSIP_DELAY: Duration = Duration::from_secs(7);
140
141/// Controls how long we wait for a block download request to complete.
142///
143/// This timeout makes sure that the syncer doesn't hang when:
144///   - the lookahead queue is full, and
145///   - we are waiting for a request that is stuck.
146///
147/// See [`BLOCK_VERIFY_TIMEOUT`] for details.
148///
149/// ## Correctness
150///
151/// If this timeout is removed (or set too high), the syncer will sometimes hang.
152///
153/// If this timeout is set too low, the syncer will sometimes get stuck in a
154/// failure loop.
155///
156/// We set the timeout so that it requires under 1 Mbps bandwidth for a full 2 MB block.
157pub(super) const BLOCK_DOWNLOAD_TIMEOUT: Duration = Duration::from_secs(20);
158
159/// Controls how long we wait for a block verify request to complete.
160///
161/// This timeout makes sure that the syncer doesn't hang when:
162///  - the lookahead queue is full, and
163///  - all pending verifications:
164///    - are waiting on a missing download request,
165///    - are waiting on a download or verify request that has failed, but we have
166///      deliberately ignored the error,
167///    - are for blocks a long way ahead of the current tip, or
168///    - are for invalid blocks which will never verify, because they depend on
169///      missing blocks or transactions.
170///
171/// These conditions can happen during normal operation - they are not bugs.
172///
173/// This timeout also mitigates or hides the following kinds of bugs:
174///  - all pending verifications:
175///    - are waiting on a download or verify request that has failed, but we have
176///      accidentally dropped the error,
177///    - are waiting on a download request that has hung inside Zebra,
178///    - are on tokio threads that are waiting for blocked operations.
179///
180/// ## Correctness
181///
182/// If this timeout is removed (or set too high), the syncer will sometimes hang.
183///
184/// If this timeout is set too low, the syncer will sometimes get stuck in a
185/// failure loop.
186///
187/// We've observed spurious 15 minute timeouts when a lot of blocks are being committed to
188/// the state. But there are also some blocks that seem to hang entirely, and never return.
189///
190/// So we allow about half the spurious timeout, which might cause some re-downloads.
191pub(super) const BLOCK_VERIFY_TIMEOUT: Duration = Duration::from_secs(8 * 60);
192
193/// A shorter timeout used for the first few blocks after the final checkpoint.
194///
195/// This is a workaround for bug #5125, where the first fully validated blocks
196/// after the final checkpoint fail with a timeout, due to a UTXO race condition.
197const FINAL_CHECKPOINT_BLOCK_VERIFY_TIMEOUT: Duration = Duration::from_secs(2 * 60);
198
199/// The number of blocks after the final checkpoint that get the shorter timeout.
200///
201/// We've only seen this error on the first few blocks after the final checkpoint.
202const FINAL_CHECKPOINT_BLOCK_VERIFY_TIMEOUT_LIMIT: HeightDiff = 100;
203
204/// Controls how long we wait to restart syncing after finishing a sync run.
205///
206/// This delay should be long enough to:
207///   - allow zcashd peers to process pending requests. If the node only has a
208///     few peers, we want to clear as much peer state as possible. In
209///     particular, zcashd sends "next block range" hints, based on zcashd's
210///     internal model of our sync progress. But we want to discard these hints,
211///     so they don't get confused with ObtainTips and ExtendTips responses, and
212///   - allow in-progress downloads to time out.
213///
214/// This delay is particularly important on instances with slow or unreliable
215/// networks, and on testnet, which has a small number of slow peers.
216///
217/// Using a prime number makes sure that syncer fanouts don't synchronise with other crawls.
218///
219/// ## Correctness
220///
221/// If this delay is removed (or set too low), the syncer will
222/// sometimes get stuck in a failure loop, due to leftover downloads from
223/// previous sync runs.
224const SYNC_RESTART_DELAY: Duration = Duration::from_secs(67);
225
226/// In regtest, use a much shorter restart delay so that downstream nodes pick up
227/// newly-mined blocks quickly (e.g. after `generate(N)` in integration tests).
228/// The default 67-second delay exceeds the typical `sync_all` timeout of 60 seconds.
229const REGTEST_SYNC_RESTART_DELAY: Duration = Duration::from_secs(2);
230
231/// Controls how long we wait to retry a failed attempt to download
232/// and verify the genesis block.
233///
234/// This timeout gives the crawler time to find better peers.
235///
236/// ## Security
237///
238/// If this timeout is removed (or set too low), Zebra will immediately retry
239/// to download and verify the genesis block from its peers. This can cause
240/// a denial of service on those peers.
241///
242/// If this timeout is too short, old or buggy nodes will keep making useless
243/// network requests. If there are a lot of them, it could overwhelm the network.
244const GENESIS_TIMEOUT_RETRY: Duration = Duration::from_secs(10);
245
246/// Sync configuration section.
247#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)]
248#[serde(deny_unknown_fields, default)]
249pub struct Config {
250    /// The number of parallel block download requests.
251    ///
252    /// This is set to a low value by default, to avoid task and
253    /// network contention. Increasing this value may improve
254    /// performance on machines with a fast network connection.
255    #[serde(alias = "max_concurrent_block_requests")]
256    pub download_concurrency_limit: usize,
257
258    /// The number of blocks submitted in parallel to the checkpoint verifier.
259    ///
260    /// Increasing this limit increases the buffer size, so it reduces
261    /// the impact of an individual block request failing. However, it
262    /// also increases memory and CPU usage if block validation stalls,
263    /// or there are some large blocks in the pipeline.
264    ///
265    /// The block size limit is 2MB, so in theory, this could represent multiple
266    /// gigabytes of data, if we downloaded arbitrary blocks. However,
267    /// because we randomly load balance outbound requests, and separate
268    /// block download from obtaining block hashes, an adversary would
269    /// have to control a significant fraction of our peers to lead us
270    /// astray.
271    ///
272    /// For reliable checkpoint syncing, Zebra enforces a
273    /// [`MIN_CHECKPOINT_CONCURRENCY_LIMIT`].
274    ///
275    /// This is set to a high value by default, to avoid verification pipeline stalls.
276    /// Decreasing this value reduces RAM usage.
277    #[serde(alias = "lookahead_limit")]
278    pub checkpoint_verify_concurrency_limit: usize,
279
280    /// The number of blocks submitted in parallel to the full verifier.
281    ///
282    /// This is set to a low value by default, to avoid verification timeouts on large blocks.
283    /// Increasing this value may improve performance on machines with many cores.
284    pub full_verify_concurrency_limit: usize,
285
286    /// The number of threads used to verify signatures, proofs, and other CPU-intensive code.
287    ///
288    /// If the number of threads is not configured or zero, Zebra uses the number of logical cores.
289    /// If the number of logical cores can't be detected, Zebra uses one thread.
290    /// For details, see [the `rayon` documentation](https://docs.rs/rayon/latest/rayon/struct.ThreadPoolBuilder.html#method.num_threads).
291    pub parallel_cpu_threads: usize,
292}
293
294impl Default for Config {
295    fn default() -> Self {
296        Self {
297            // 2/3 of the default outbound peer limit.
298            download_concurrency_limit: 50,
299
300            // A few max-length checkpoints.
301            checkpoint_verify_concurrency_limit: DEFAULT_CHECKPOINT_CONCURRENCY_LIMIT,
302
303            // This default is deliberately very low, so Zebra can verify a few large blocks in under 60 seconds,
304            // even on machines with only a few cores.
305            //
306            // This lets users see the committed block height changing in every progress log,
307            // and avoids hangs due to out-of-order verifications flooding the CPUs.
308            //
309            // TODO:
310            // - limit full verification concurrency based on block transaction counts?
311            // - move more disk work to blocking tokio threads,
312            //   and CPU work to the rayon thread pool inside blocking tokio threads
313            full_verify_concurrency_limit: 20,
314
315            // Use one thread per CPU.
316            //
317            // If this causes tokio executor starvation, move CPU-intensive tasks to rayon threads,
318            // or reserve a few cores for tokio threads, based on `num_cpus()`.
319            parallel_cpu_threads: 0,
320        }
321    }
322}
323
324/// Helps work around defects in the bitcoin protocol by checking whether
325/// the returned hashes actually extend a chain tip.
326#[derive(Copy, Clone, Debug, Hash, PartialEq, Eq)]
327struct CheckedTip {
328    tip: block::Hash,
329    expected_next: block::Hash,
330}
331
332pub struct ChainSync<ZN, ZS, ZV, ZSTip>
333where
334    ZN: Service<zn::Request, Response = zn::Response, Error = BoxError>
335        + Send
336        + Sync
337        + Clone
338        + 'static,
339    ZN::Future: Send,
340    ZS: Service<zs::Request, Response = zs::Response, Error = BoxError>
341        + Send
342        + Sync
343        + Clone
344        + 'static,
345    ZS::Future: Send,
346    ZV: Service<zebra_consensus::Request, Response = block::Hash, Error = BoxError>
347        + Send
348        + Sync
349        + Clone
350        + 'static,
351    ZV::Future: Send,
352    ZSTip: ChainTip + Clone + Send + 'static,
353{
354    // Configuration
355    //
356    /// The genesis hash for the configured network
357    genesis_hash: block::Hash,
358
359    /// The largest block height for the checkpoint verifier, based on the current config.
360    max_checkpoint_height: Height,
361
362    /// The configured checkpoint verification concurrency limit, after applying the minimum limit.
363    checkpoint_verify_concurrency_limit: usize,
364
365    /// The configured full verification concurrency limit, after applying the minimum limit.
366    full_verify_concurrency_limit: usize,
367
368    /// Whether the node is running on regtest. Used to apply a shorter sync restart delay.
369    is_regtest: bool,
370
371    // Services
372    //
373    /// A network service which is used to perform ObtainTips and ExtendTips
374    /// requests.
375    ///
376    /// Has no retry logic, because failover is handled using fanout.
377    tip_network: Timeout<ZN>,
378
379    /// A service which downloads and verifies blocks, using the provided
380    /// network and verifier services.
381    downloads: Pin<
382        Box<
383            Downloads<
384                Hedge<ConcurrencyLimit<Retry<zn::RetryLimit, Timeout<ZN>>>, AlwaysHedge>,
385                Timeout<ZV>,
386                ZSTip,
387            >,
388        >,
389    >,
390
391    /// The cached block chain state.
392    state: ZS,
393
394    /// Allows efficient access to the best tip of the blockchain.
395    latest_chain_tip: ZSTip,
396
397    // Internal sync state
398    //
399    /// The tips that the syncer is currently following.
400    prospective_tips: HashSet<CheckedTip>,
401
402    /// The lengths of recent sync responses.
403    recent_syncs: RecentSyncLengths,
404
405    /// Receiver that is `true` when the downloader is past the lookahead limit.
406    /// This is based on the downloaded block height and the state tip height.
407    past_lookahead_limit_receiver: zs::WatchReceiver<bool>,
408
409    /// Sender for reporting peer addresses that advertised unexpectedly invalid transactions.
410    misbehavior_sender: mpsc::Sender<(PeerSocketAddr, u32)>,
411
412    /// Blocks whose download failed with `NotFound` and should be re-requested on
413    /// the next sync round, instead of being silently dropped (#5709).
414    reobtain_hashes: IndexSet<block::Hash>,
415
416    /// Per-hash count of how many times a `NotFound` block has been re-requested,
417    /// bounded by [`MAX_BLOCK_REOBTAIN_RETRIES`].
418    block_reobtain_retries: HashMap<block::Hash, u8>,
419}
420
421/// Polls the network to determine whether further blocks are available and
422/// downloads them.
423///
424/// This component is used for initial block sync, but the `Inbound` service is
425/// responsible for participating in the gossip protocols used for block
426/// diffusion.
427impl<ZN, ZS, ZV, ZSTip> ChainSync<ZN, ZS, ZV, ZSTip>
428where
429    ZN: Service<zn::Request, Response = zn::Response, Error = BoxError>
430        + Send
431        + Sync
432        + Clone
433        + 'static,
434    ZN::Future: Send,
435    ZS: Service<zs::Request, Response = zs::Response, Error = BoxError>
436        + Send
437        + Sync
438        + Clone
439        + 'static,
440    ZS::Future: Send,
441    ZV: Service<zebra_consensus::Request, Response = block::Hash, Error = BoxError>
442        + Send
443        + Sync
444        + Clone
445        + 'static,
446    ZV::Future: Send,
447    ZSTip: ChainTip + Clone + Send + 'static,
448{
449    /// Returns a new syncer instance, using:
450    ///  - chain: the zebra-chain `Network` to download (Mainnet or Testnet)
451    ///  - peers: the zebra-network peers to contact for downloads
452    ///  - verifier: the zebra-consensus verifier that checks the chain
453    ///  - state: the zebra-state that stores the chain
454    ///  - latest_chain_tip: the latest chain tip from `state`
455    ///
456    /// Also returns a [`SyncStatus`] to check if the syncer has likely reached the chain tip.
457    pub fn new(
458        config: &ZebradConfig,
459        max_checkpoint_height: Height,
460        peers: ZN,
461        verifier: ZV,
462        state: ZS,
463        latest_chain_tip: ZSTip,
464        misbehavior_sender: mpsc::Sender<(PeerSocketAddr, u32)>,
465    ) -> (Self, SyncStatus) {
466        let mut download_concurrency_limit = config.sync.download_concurrency_limit;
467        let mut checkpoint_verify_concurrency_limit =
468            config.sync.checkpoint_verify_concurrency_limit;
469        let mut full_verify_concurrency_limit = config.sync.full_verify_concurrency_limit;
470
471        if download_concurrency_limit < MIN_CONCURRENCY_LIMIT {
472            warn!(
473                "configured download concurrency limit {} too low, increasing to {}",
474                config.sync.download_concurrency_limit, MIN_CONCURRENCY_LIMIT,
475            );
476
477            download_concurrency_limit = MIN_CONCURRENCY_LIMIT;
478        }
479
480        if checkpoint_verify_concurrency_limit < MIN_CHECKPOINT_CONCURRENCY_LIMIT {
481            warn!(
482                "configured checkpoint verify concurrency limit {} too low, increasing to {}",
483                config.sync.checkpoint_verify_concurrency_limit, MIN_CHECKPOINT_CONCURRENCY_LIMIT,
484            );
485
486            checkpoint_verify_concurrency_limit = MIN_CHECKPOINT_CONCURRENCY_LIMIT;
487        }
488
489        if full_verify_concurrency_limit < MIN_CONCURRENCY_LIMIT {
490            warn!(
491                "configured full verify concurrency limit {} too low, increasing to {}",
492                config.sync.full_verify_concurrency_limit, MIN_CONCURRENCY_LIMIT,
493            );
494
495            full_verify_concurrency_limit = MIN_CONCURRENCY_LIMIT;
496        }
497
498        let tip_network = Timeout::new(peers.clone(), TIPS_RESPONSE_TIMEOUT);
499
500        // The Hedge middleware is the outermost layer, hedging requests
501        // between two retry-wrapped networks.  The innermost timeout
502        // layer is relatively unimportant, because slow requests will
503        // probably be preemptively hedged.
504        //
505        // The Hedge goes outside the Retry, because the Retry layer
506        // abstracts away spurious failures from individual peers
507        // making a less-fallible network service, and the Hedge layer
508        // tries to reduce latency of that less-fallible service.
509        let block_network = Hedge::new(
510            ServiceBuilder::new()
511                .concurrency_limit(download_concurrency_limit)
512                .retry(zn::RetryLimit::new(BLOCK_DOWNLOAD_RETRY_LIMIT))
513                .timeout(BLOCK_DOWNLOAD_TIMEOUT)
514                .service(peers),
515            AlwaysHedge,
516            20,
517            0.95,
518            2 * SYNC_RESTART_DELAY,
519        );
520
521        // We apply a timeout to the verifier to avoid hangs due to missing earlier blocks.
522        let verifier = Timeout::new(verifier, BLOCK_VERIFY_TIMEOUT);
523
524        let (sync_status, recent_syncs) = SyncStatus::new_for_network(&config.network.network);
525
526        let (past_lookahead_limit_sender, past_lookahead_limit_receiver) = watch::channel(false);
527        let past_lookahead_limit_receiver = zs::WatchReceiver::new(past_lookahead_limit_receiver);
528
529        let downloads = Box::pin(Downloads::new(
530            block_network,
531            verifier,
532            latest_chain_tip.clone(),
533            past_lookahead_limit_sender,
534            max(
535                checkpoint_verify_concurrency_limit,
536                full_verify_concurrency_limit,
537            ),
538            max_checkpoint_height,
539        ));
540
541        let new_syncer = Self {
542            genesis_hash: config.network.network.genesis_hash(),
543            max_checkpoint_height,
544            checkpoint_verify_concurrency_limit,
545            full_verify_concurrency_limit,
546            is_regtest: config.network.network.is_regtest(),
547            tip_network,
548            downloads,
549            state,
550            latest_chain_tip,
551            prospective_tips: HashSet::new(),
552            recent_syncs,
553            past_lookahead_limit_receiver,
554            misbehavior_sender,
555            reobtain_hashes: IndexSet::new(),
556            block_reobtain_retries: HashMap::new(),
557        };
558
559        (new_syncer, sync_status)
560    }
561
562    /// Runs the syncer to synchronize the chain and keep it synchronized.
563    #[instrument(skip(self))]
564    pub async fn sync(mut self) -> Result<(), Report> {
565        // We can't download the genesis block using our normal algorithm,
566        // due to protocol limitations
567        self.request_genesis().await?;
568
569        loop {
570            if self.try_to_sync().await.is_err() {
571                self.downloads.cancel_all();
572            }
573
574            self.update_metrics();
575
576            let restart_delay = if self.is_regtest {
577                REGTEST_SYNC_RESTART_DELAY
578            } else {
579                SYNC_RESTART_DELAY
580            };
581            info!(
582                timeout = ?restart_delay,
583                state_tip = ?self.latest_chain_tip.best_tip_height(),
584                "waiting to restart sync"
585            );
586            sleep(restart_delay).await;
587        }
588    }
589
590    /// Tries to synchronize the chain as far as it can.
591    ///
592    /// Obtains some prospective tips and iteratively tries to extend them and download the missing
593    /// blocks.
594    ///
595    /// Returns `Ok` if it was able to synchronize as much of the chain as it could, and then ran
596    /// out of prospective tips. This happens when synchronization finishes or if Zebra ended up
597    /// following a fork. Either way, Zebra should attempt to obtain some more tips.
598    ///
599    /// Returns `Err` if there was an unrecoverable error and restarting the synchronization is
600    /// necessary. This includes outer timeouts, where an entire syncing step takes an extremely
601    /// long time. (These usually indicate hangs.)
602    #[instrument(skip(self))]
603    async fn try_to_sync(&mut self) -> Result<(), Report> {
604        self.prospective_tips = HashSet::new();
605
606        self.reobtain_hashes.clear();
607        self.block_reobtain_retries.clear();
608
609        info!(
610            state_tip = ?self.latest_chain_tip.best_tip_height(),
611            "starting sync, obtaining new tips"
612        );
613        let mut extra_hashes = timeout(SYNC_RESTART_DELAY, self.obtain_tips())
614            .await
615            .map_err(Into::into)
616            // TODO: replace with flatten() when it stabilises (#70142)
617            .and_then(convert::identity)
618            .map_err(|e| {
619                info!("temporary error obtaining tips: {:#}", e);
620                e
621            })?;
622        self.update_metrics();
623
624        while !self.prospective_tips.is_empty() || !extra_hashes.is_empty() {
625            // Avoid hangs due to service readiness or other internal operations
626            extra_hashes = timeout(BLOCK_VERIFY_TIMEOUT, self.try_to_sync_once(extra_hashes))
627                .await
628                .map_err(Into::into)
629                // TODO: replace with flatten() when it stabilises (#70142)
630                .and_then(convert::identity)?;
631        }
632
633        info!("exhausted prospective tip set");
634
635        Ok(())
636    }
637
638    /// Tries to synchronize the chain once, using the existing `extra_hashes`.
639    ///
640    /// Tries to extend the existing tips and download the missing blocks.
641    ///
642    /// Returns `Ok(extra_hashes)` if it was able to extend once and synchronize sone of the chain.
643    /// Returns `Err` if there was an unrecoverable error and restarting the synchronization is
644    /// necessary.
645    #[instrument(skip(self, extra_hashes))]
646    async fn try_to_sync_once(
647        &mut self,
648        mut extra_hashes: IndexSet<block::Hash>,
649    ) -> Result<IndexSet<block::Hash>, Report> {
650        // Check whether any block tasks are currently ready.
651        while let Poll::Ready(Some(rsp)) = futures::poll!(self.downloads.next()) {
652            // Some temporary errors are ignored, and syncing continues with other blocks.
653            // If it turns out they were actually important, syncing will run out of blocks, and
654            // the syncer will reset itself.
655            self.handle_block_response(rsp)?;
656        }
657        // Re-request any blocks that just failed with `NotFound`, before pausing
658        // on the lookahead limit (#5709).
659        self.reobtain_missing_blocks().await;
660        self.update_metrics();
661
662        // Pause new downloads while the syncer or downloader are past their lookahead limits.
663        //
664        // To avoid a deadlock or long waits for blocks to expire, we ignore the download
665        // lookahead limit when there are only a small number of blocks waiting.
666        while self.downloads.in_flight() >= self.lookahead_limit(extra_hashes.len())
667            || (self.downloads.in_flight() >= self.lookahead_limit(extra_hashes.len()) / 2
668                && self.past_lookahead_limit_receiver.cloned_watch_data())
669        {
670            trace!(
671                tips.len = self.prospective_tips.len(),
672                in_flight = self.downloads.in_flight(),
673                extra_hashes = extra_hashes.len(),
674                lookahead_limit = self.lookahead_limit(extra_hashes.len()),
675                state_tip = ?self.latest_chain_tip.best_tip_height(),
676                "waiting for pending blocks",
677            );
678
679            let response = self.downloads.next().await.expect("downloads is nonempty");
680
681            self.handle_block_response(response)?;
682            // A block that just failed with `NotFound` is what unblocks the
683            // verifier, so re-request it now rather than waiting for the pause
684            // loop to clear — which it cannot until this block arrives (#5709).
685            self.reobtain_missing_blocks().await;
686            self.update_metrics();
687        }
688
689        // Once we're below the lookahead limit, we can request more blocks or hashes.
690        if !extra_hashes.is_empty() {
691            debug!(
692                tips.len = self.prospective_tips.len(),
693                in_flight = self.downloads.in_flight(),
694                extra_hashes = extra_hashes.len(),
695                lookahead_limit = self.lookahead_limit(extra_hashes.len()),
696                state_tip = ?self.latest_chain_tip.best_tip_height(),
697                "requesting more blocks",
698            );
699
700            let response = self.request_blocks(extra_hashes).await;
701            extra_hashes = Self::handle_hash_response(response)?;
702        } else {
703            info!(
704                tips.len = self.prospective_tips.len(),
705                in_flight = self.downloads.in_flight(),
706                extra_hashes = extra_hashes.len(),
707                lookahead_limit = self.lookahead_limit(extra_hashes.len()),
708                state_tip = ?self.latest_chain_tip.best_tip_height(),
709                "extending tips",
710            );
711
712            extra_hashes = self.extend_tips().await.map_err(|e| {
713                info!("temporary error extending tips: {:#}", e);
714                e
715            })?;
716        }
717        self.update_metrics();
718
719        Ok(extra_hashes)
720    }
721
722    /// Re-issues downloads for blocks that failed with `NotFound` (#5709).
723    ///
724    /// These are re-requested even while the download pipeline is past its
725    /// lookahead limit, because a missing low block is exactly what stops the
726    /// checkpoint verifier from advancing. Waiting for the lookahead pause to
727    /// clear would deadlock — the pause cannot clear until this block arrives.
728    /// The per-hash retry count is bounded by [`MAX_BLOCK_REOBTAIN_RETRIES`].
729    async fn reobtain_missing_blocks(&mut self) {
730        if self.reobtain_hashes.is_empty() {
731            return;
732        }
733
734        for hash in std::mem::take(&mut self.reobtain_hashes) {
735            // The block was removed from the in-flight set when its download
736            // failed, so this re-queues it. A residual duplicate/queue error is
737            // benign — it means the block is already being handled.
738            if let Err(error) = self.downloads.download_and_verify(hash).await {
739                trace!(?hash, ?error, "re-download of missing block not queued");
740            }
741        }
742    }
743
744    /// Given a block_locator list fan out request for subsequent hashes to
745    /// multiple peers
746    #[instrument(skip(self))]
747    async fn obtain_tips(&mut self) -> Result<IndexSet<block::Hash>, Report> {
748        let stage_start = std::time::Instant::now();
749
750        let block_locator = self
751            .state
752            .ready()
753            .await
754            .map_err(|e| eyre!(e))?
755            .call(zebra_state::Request::BlockLocator)
756            .await
757            .map(|response| match response {
758                zebra_state::Response::BlockLocator(block_locator) => block_locator,
759                _ => unreachable!(
760                    "GetBlockLocator request can only result in Response::BlockLocator"
761                ),
762            })
763            .map_err(|e| eyre!(e))?;
764
765        debug!(
766            tip = ?block_locator.first().expect("we have at least one block locator object"),
767            ?block_locator,
768            "got block locator and trying to obtain new chain tips"
769        );
770
771        let mut requests = FuturesUnordered::new();
772        for attempt in 0..FANOUT {
773            if attempt > 0 {
774                // Let other tasks run, so we're more likely to choose a different peer.
775                //
776                // TODO: move fanouts into the PeerSet, so we always choose different peers (#2214)
777                tokio::task::yield_now().await;
778            }
779
780            let ready_tip_network = self.tip_network.ready().await;
781            requests.push(tokio::spawn(ready_tip_network.map_err(|e| eyre!(e))?.call(
782                zn::Request::FindBlocks {
783                    known_blocks: block_locator.clone(),
784                    stop: None,
785                },
786            )));
787        }
788
789        let mut download_set = IndexSet::new();
790        while let Some(res) = requests.next().await {
791            match res
792                .unwrap_or_else(|e @ JoinError { .. }| {
793                    if e.is_panic() {
794                        panic!("panic in obtain tips task: {e:?}");
795                    } else {
796                        info!(
797                            "task error during obtain tips task: {e:?},\
798                     is Zebra shutting down?"
799                        );
800                        Err(e.into())
801                    }
802                })
803                .map_err::<Report, _>(|e| eyre!(e))
804            {
805                Ok(zn::Response::BlockHashes(hashes)) => {
806                    trace!(?hashes);
807
808                    // zcashd sometimes appends an unrelated hash at the start
809                    // or end of its response.
810                    //
811                    // We can't discard the first hash, because it might be a
812                    // block we want to download. So we just accept any
813                    // out-of-order first hashes.
814
815                    // We use the last hash for the tip, and we want to avoid bad
816                    // tips from zcashd's quirk of appending an unrelated hash.
817                    // So we discard the last hash on mainnet/testnet.
818                    // (We don't need to worry about missed downloads, because we
819                    // will pick them up again in ExtendTips.)
820                    //
821                    // In regtest we only connect to Zebra nodes, not zcashd,
822                    // so we trust all hashes in the response and keep them all.
823                    // This is necessary when there are only a small number of
824                    // blocks to sync (e.g. 2 new blocks), where stripping the
825                    // last hash leaves only 1 unknown hash and rchunks_exact(2)
826                    // would discard the entire response.
827                    let hashes = if self.is_regtest {
828                        hashes.as_slice()
829                    } else {
830                        match hashes.as_slice() {
831                            [] => continue,
832                            [rest @ .., _last] => rest,
833                        }
834                    };
835                    if hashes.is_empty() {
836                        continue;
837                    }
838
839                    let mut first_unknown = None;
840                    for (i, &hash) in hashes.iter().enumerate() {
841                        if !self.state_contains(hash).await? {
842                            first_unknown = Some(i);
843                            break;
844                        }
845                    }
846
847                    debug!(hashes.len = ?hashes.len(), ?first_unknown);
848
849                    let unknown_hashes = if let Some(index) = first_unknown {
850                        &hashes[index..]
851                    } else {
852                        continue;
853                    };
854
855                    trace!(?unknown_hashes);
856
857                    let new_tip = if let Some(end) = unknown_hashes.rchunks_exact(2).next() {
858                        CheckedTip {
859                            tip: end[0],
860                            expected_next: end[1],
861                        }
862                    } else {
863                        debug!("discarding response that extends only one block");
864                        continue;
865                    };
866
867                    // Make sure we get the same tips, regardless of the
868                    // order of peer responses
869                    if !download_set.contains(&new_tip.expected_next) {
870                        debug!(?new_tip,
871                                        "adding new prospective tip, and removing existing tips in the new block hash list");
872                        self.prospective_tips
873                            .retain(|t| !unknown_hashes.contains(&t.expected_next));
874                        self.prospective_tips.insert(new_tip);
875                    } else {
876                        debug!(
877                            ?new_tip,
878                            "discarding prospective tip: already in download set"
879                        );
880                    }
881
882                    // security: the first response determines our download order
883                    //
884                    // TODO: can we make the download order independent of response order?
885                    let prev_download_len = download_set.len();
886                    download_set.extend(unknown_hashes);
887                    let new_download_len = download_set.len();
888                    let new_hashes = new_download_len - prev_download_len;
889                    debug!(new_hashes, "added hashes to download set");
890                    metrics::histogram!("sync.obtain.response.hash.count")
891                        .record(new_hashes as f64);
892                }
893                Ok(_) => unreachable!("network returned wrong response"),
894                // We ignore this error because we made multiple fanout requests.
895                Err(e) => debug!(?e),
896            }
897        }
898
899        debug!(?self.prospective_tips);
900
901        // Check that the new tips we got are actually unknown.
902        for hash in &download_set {
903            debug!(?hash, "checking if state contains hash");
904            if self.state_contains(*hash).await? {
905                return Err(eyre!("queued download of hash behind our chain tip"));
906            }
907        }
908
909        let new_downloads = download_set.len();
910        debug!(new_downloads, "queueing new downloads");
911        metrics::gauge!("sync.obtain.queued.hash.count").set(new_downloads as f64);
912
913        // security: use the actual number of new downloads from all peers,
914        // so the last peer to respond can't toggle our mempool
915        self.recent_syncs.push_obtain_tips_length(new_downloads);
916
917        let response = self.request_blocks(download_set).await;
918
919        metrics::histogram!("sync.stage.duration_seconds", "stage" => "obtain_tips")
920            .record(stage_start.elapsed().as_secs_f64());
921
922        Self::handle_hash_response(response).map_err(Into::into)
923    }
924
925    #[instrument(skip(self))]
926    async fn extend_tips(&mut self) -> Result<IndexSet<block::Hash>, Report> {
927        let stage_start = std::time::Instant::now();
928
929        let tips = std::mem::take(&mut self.prospective_tips);
930
931        let mut download_set = IndexSet::new();
932        debug!(tips = ?tips.len(), "trying to extend chain tips");
933        for tip in tips {
934            debug!(?tip, "asking peers to extend chain tip");
935            let mut responses = FuturesUnordered::new();
936            for attempt in 0..FANOUT {
937                if attempt > 0 {
938                    // Let other tasks run, so we're more likely to choose a different peer.
939                    //
940                    // TODO: move fanouts into the PeerSet, so we always choose different peers (#2214)
941                    tokio::task::yield_now().await;
942                }
943
944                let ready_tip_network = self.tip_network.ready().await;
945                responses.push(tokio::spawn(ready_tip_network.map_err(|e| eyre!(e))?.call(
946                    zn::Request::FindBlocks {
947                        known_blocks: vec![tip.tip],
948                        stop: None,
949                    },
950                )));
951            }
952            while let Some(res) = responses.next().await {
953                match res
954                    .expect("panic in spawned extend tips request")
955                    .map_err::<Report, _>(|e| eyre!(e))
956                {
957                    Ok(zn::Response::BlockHashes(hashes)) => {
958                        debug!(first = ?hashes.first(), len = ?hashes.len());
959                        trace!(?hashes);
960
961                        // zcashd sometimes appends an unrelated hash at the
962                        // start or end of its response. Check the first hash
963                        // against the previous response, and discard mismatches.
964                        let unknown_hashes = match hashes.as_slice() {
965                            [expected_hash, rest @ ..] if expected_hash == &tip.expected_next => {
966                                rest
967                            }
968                            // If the first hash doesn't match, retry with the second.
969                            [first_hash, expected_hash, rest @ ..]
970                                if expected_hash == &tip.expected_next =>
971                            {
972                                debug!(?first_hash,
973                                                ?tip.expected_next,
974                                                ?tip.tip,
975                                                "unexpected first hash, but the second matches: using the hashes after the match");
976                                rest
977                            }
978                            // We ignore these responses
979                            [] => continue,
980                            [single_hash] => {
981                                debug!(?single_hash,
982                                                ?tip.expected_next,
983                                                ?tip.tip,
984                                                "discarding response containing a single unexpected hash");
985                                continue;
986                            }
987                            [first_hash, second_hash, rest @ ..] => {
988                                debug!(?first_hash,
989                                                ?second_hash,
990                                                rest_len = ?rest.len(),
991                                                ?tip.expected_next,
992                                                ?tip.tip,
993                                                "discarding response that starts with two unexpected hashes");
994                                continue;
995                            }
996                        };
997
998                        // We use the last hash for the tip, and we want to avoid
999                        // bad tips. So we discard the last hash. (We don't need
1000                        // to worry about missed downloads, because we will pick
1001                        // them up again in the next ExtendTips.)
1002                        let unknown_hashes = match unknown_hashes {
1003                            [] => continue,
1004                            [rest @ .., _last] => rest,
1005                        };
1006
1007                        let new_tip = if let Some(end) = unknown_hashes.rchunks_exact(2).next() {
1008                            CheckedTip {
1009                                tip: end[0],
1010                                expected_next: end[1],
1011                            }
1012                        } else {
1013                            debug!("discarding response that extends only one block");
1014                            continue;
1015                        };
1016
1017                        trace!(?unknown_hashes);
1018
1019                        // Make sure we get the same tips, regardless of the
1020                        // order of peer responses
1021                        if !download_set.contains(&new_tip.expected_next) {
1022                            debug!(?new_tip,
1023                                            "adding new prospective tip, and removing any existing tips in the new block hash list");
1024                            self.prospective_tips
1025                                .retain(|t| !unknown_hashes.contains(&t.expected_next));
1026                            self.prospective_tips.insert(new_tip);
1027                        } else {
1028                            debug!(
1029                                ?new_tip,
1030                                "discarding prospective tip: already in download set"
1031                            );
1032                        }
1033
1034                        // security: the first response determines our download order
1035                        //
1036                        // TODO: can we make the download order independent of response order?
1037                        let prev_download_len = download_set.len();
1038                        download_set.extend(unknown_hashes);
1039                        let new_download_len = download_set.len();
1040                        let new_hashes = new_download_len - prev_download_len;
1041                        debug!(new_hashes, "added hashes to download set");
1042                        metrics::histogram!("sync.extend.response.hash.count")
1043                            .record(new_hashes as f64);
1044                    }
1045                    Ok(_) => unreachable!("network returned wrong response"),
1046                    // We ignore this error because we made multiple fanout requests.
1047                    Err(e) => debug!(?e),
1048                }
1049            }
1050        }
1051
1052        let new_downloads = download_set.len();
1053        debug!(new_downloads, "queueing new downloads");
1054        metrics::gauge!("sync.extend.queued.hash.count").set(new_downloads as f64);
1055
1056        // security: use the actual number of new downloads from all peers,
1057        // so the last peer to respond can't toggle our mempool
1058        self.recent_syncs.push_extend_tips_length(new_downloads);
1059
1060        let response = self.request_blocks(download_set).await;
1061
1062        metrics::histogram!("sync.stage.duration_seconds", "stage" => "extend_tips")
1063            .record(stage_start.elapsed().as_secs_f64());
1064
1065        Self::handle_hash_response(response).map_err(Into::into)
1066    }
1067
1068    /// Download and verify the genesis block, if it isn't currently known to
1069    /// our node.
1070    async fn request_genesis(&mut self) -> Result<(), Report> {
1071        // Due to Bitcoin protocol limitations, we can't request the genesis
1072        // block using our standard tip-following algorithm:
1073        //  - getblocks requires at least one hash
1074        //  - responses start with the block *after* the requested block, and
1075        //  - the genesis hash is used as a placeholder for "no matches".
1076        //
1077        // So we just download and verify the genesis block here.
1078        while !self.state_contains(self.genesis_hash).await? {
1079            info!("starting genesis block download and verify");
1080
1081            let response = timeout(SYNC_RESTART_DELAY, self.request_genesis_once())
1082                .await
1083                .map_err(Into::into);
1084
1085            // 3 layers of results is not ideal, but we need the timeout on the outside.
1086            match response {
1087                Ok(Ok(Ok(response))) => self
1088                    .handle_block_response(Ok(response))
1089                    .expect("never returns Err for Ok"),
1090                // Handle fatal errors
1091                Ok(Err(fatal_error)) => Err(fatal_error)?,
1092                // Handle timeouts and block errors
1093                Err(error) | Ok(Ok(Err(error))) => {
1094                    // TODO: exit syncer on permanent service errors (NetworkError, VerifierError)
1095                    if Self::should_restart_sync(&error) {
1096                        warn!(
1097                            ?error,
1098                            "could not download or verify genesis block, retrying"
1099                        );
1100                    } else {
1101                        info!(
1102                            ?error,
1103                            "temporary error downloading or verifying genesis block, retrying"
1104                        );
1105                    }
1106
1107                    tokio::time::sleep(GENESIS_TIMEOUT_RETRY).await;
1108                }
1109            }
1110        }
1111
1112        Ok(())
1113    }
1114
1115    /// Try to download and verify the genesis block once.
1116    ///
1117    /// Fatal errors are returned in the outer result, temporary errors in the inner one.
1118    async fn request_genesis_once(
1119        &mut self,
1120    ) -> Result<Result<(Height, block::Hash), BlockDownloadVerifyError>, Report> {
1121        let response = self.downloads.download_and_verify(self.genesis_hash).await;
1122        Self::handle_response(response).map_err(|e| eyre!(e))?;
1123
1124        let response = self.downloads.next().await.expect("downloads is nonempty");
1125
1126        Ok(response)
1127    }
1128
1129    /// Queue download and verify tasks for each block that isn't currently known to our node.
1130    ///
1131    /// TODO: turn obtain and extend tips into a separate task, which sends hashes via a channel?
1132    async fn request_blocks(
1133        &mut self,
1134        mut hashes: IndexSet<block::Hash>,
1135    ) -> Result<IndexSet<block::Hash>, BlockDownloadVerifyError> {
1136        let lookahead_limit = self.lookahead_limit(hashes.len());
1137
1138        debug!(
1139            hashes.len = hashes.len(),
1140            ?lookahead_limit,
1141            "requesting blocks",
1142        );
1143
1144        let extra_hashes = if hashes.len() > lookahead_limit {
1145            hashes.split_off(lookahead_limit)
1146        } else {
1147            IndexSet::new()
1148        };
1149
1150        // Dispatch blocks with duplicate-tolerant error handling.
1151        // DuplicateBlockQueuedForDownload is caught and skipped instead of
1152        // propagating — this prevents dropping unprocessed hashes from the
1153        // batch, which would create frontier gaps and stalls (#5709).
1154        for hash in hashes.into_iter() {
1155            match self.downloads.download_and_verify(hash).await {
1156                Ok(()) => {}
1157                Err(BlockDownloadVerifyError::DuplicateBlockQueuedForDownload { .. }) => {
1158                    debug!("block request was already queued, continuing");
1159                }
1160                Err(error) => return Err(error),
1161            }
1162        }
1163
1164        Ok(extra_hashes)
1165    }
1166
1167    /// The configured lookahead limit, based on the currently verified height,
1168    /// and the number of hashes we haven't queued yet.
1169    fn lookahead_limit(&self, new_hashes: usize) -> usize {
1170        let max_checkpoint_height: usize = self
1171            .max_checkpoint_height
1172            .0
1173            .try_into()
1174            .expect("fits in usize");
1175
1176        // When the state is empty, we want to verify using checkpoints
1177        let verified_height: usize = self
1178            .latest_chain_tip
1179            .best_tip_height()
1180            .unwrap_or(Height(0))
1181            .0
1182            .try_into()
1183            .expect("fits in usize");
1184
1185        if verified_height >= max_checkpoint_height {
1186            self.full_verify_concurrency_limit
1187        } else if (verified_height + new_hashes) >= max_checkpoint_height {
1188            // If we're just about to start full verification, allow enough for the remaining checkpoint,
1189            // and also enough for a separate full verification lookahead.
1190            let checkpoint_hashes = verified_height + new_hashes - max_checkpoint_height;
1191
1192            self.full_verify_concurrency_limit + checkpoint_hashes
1193        } else {
1194            self.checkpoint_verify_concurrency_limit
1195        }
1196    }
1197
1198    /// Handles a response for a requested block.
1199    ///
1200    /// See [`Self::handle_response`] for more details.
1201    #[allow(unknown_lints)]
1202    fn handle_block_response(
1203        &mut self,
1204        response: Result<(Height, block::Hash), BlockDownloadVerifyError>,
1205    ) -> Result<(), BlockDownloadVerifyError> {
1206        match response {
1207            Ok((height, hash)) => {
1208                trace!(?height, ?hash, "verified and committed block to state");
1209
1210                // The block arrived, so forget any re-request bookkeeping for it.
1211                self.block_reobtain_retries.remove(&hash);
1212
1213                return Ok(());
1214            }
1215
1216            Err(BlockDownloadVerifyError::Invalid {
1217                ref error,
1218                advertiser_addr: Some(advertiser_addr),
1219                ..
1220            }) if error.misbehavior_score() != 0 => {
1221                let _ = self
1222                    .misbehavior_sender
1223                    .try_send((advertiser_addr, error.misbehavior_score()));
1224            }
1225
1226            Err(BlockDownloadVerifyError::AboveLookaheadHeightLimit {
1227                advertiser_addr: Some(advertiser_addr),
1228                ..
1229            }) => {
1230                let _ = self.misbehavior_sender.try_send((advertiser_addr, 100));
1231            }
1232
1233            Err(BlockDownloadVerifyError::InvalidHeight {
1234                advertiser_addr: Some(advertiser_addr),
1235                ..
1236            }) => {
1237                let _ = self.misbehavior_sender.try_send((advertiser_addr, 100));
1238            }
1239
1240            Err(_) => {}
1241        };
1242
1243        // A block whose download failed because no peer delivered it (`NotFound`)
1244        // is otherwise dropped here and never re-requested, which wedges the
1245        // checkpoint frontier until the verify timeout (#5709). Re-queue it for
1246        // the next sync round, bounded by `MAX_BLOCK_REOBTAIN_RETRIES`. Consensus
1247        // failures (`Invalid`/`ValidationRequestError`) are deliberately excluded —
1248        // re-downloading a block the network already rejected is pointless.
1249        if let Err(BlockDownloadVerifyError::DownloadFailed { error, hash }) = &response {
1250            if format!("{error:?}").contains("NotFound") {
1251                let attempts = self.block_reobtain_retries.entry(*hash).or_insert(0);
1252                if *attempts < MAX_BLOCK_REOBTAIN_RETRIES {
1253                    *attempts += 1;
1254                    self.reobtain_hashes.insert(*hash);
1255                    debug!(
1256                        ?hash,
1257                        attempts = *attempts,
1258                        "re-queueing missing block for re-download"
1259                    );
1260                } else {
1261                    debug!(
1262                        ?hash,
1263                        "missing block exceeded re-download retries, dropping"
1264                    );
1265                    self.block_reobtain_retries.remove(hash);
1266                }
1267            }
1268        }
1269
1270        Self::handle_response(response)
1271    }
1272
1273    /// Handles a response to block hash submission, passing through any extra hashes.
1274    ///
1275    /// See [`Self::handle_response`] for more details.
1276    #[allow(unknown_lints)]
1277    fn handle_hash_response(
1278        response: Result<IndexSet<block::Hash>, BlockDownloadVerifyError>,
1279    ) -> Result<IndexSet<block::Hash>, BlockDownloadVerifyError> {
1280        match response {
1281            Ok(extra_hashes) => Ok(extra_hashes),
1282            Err(_) => Self::handle_response(response).map(|()| IndexSet::new()),
1283        }
1284    }
1285
1286    /// Handles a response to a syncer request.
1287    ///
1288    /// Returns `Ok` if the request was successful, or if an expected error occurred,
1289    /// so that the synchronization can continue normally.
1290    ///
1291    /// Returns `Err` if an unexpected error occurred, to force the synchronizer to restart.
1292    #[allow(unknown_lints)]
1293    fn handle_response<T>(
1294        response: Result<T, BlockDownloadVerifyError>,
1295    ) -> Result<(), BlockDownloadVerifyError> {
1296        match response {
1297            Ok(_t) => Ok(()),
1298            Err(error) => {
1299                // TODO: exit syncer on permanent service errors (NetworkError, VerifierError)
1300                if Self::should_restart_sync(&error) {
1301                    Err(error)
1302                } else {
1303                    Ok(())
1304                }
1305            }
1306        }
1307    }
1308
1309    /// Returns `true` if the hash is present in the state, and `false`
1310    /// if the hash is not present in the state.
1311    pub(crate) async fn state_contains(&mut self, hash: block::Hash) -> Result<bool, Report> {
1312        match self
1313            .state
1314            .ready()
1315            .await
1316            .map_err(|e| eyre!(e))?
1317            .call(zebra_state::Request::KnownBlock(hash))
1318            .await
1319            .map_err(|e| eyre!(e))?
1320        {
1321            zs::Response::KnownBlock(loc) => Ok(loc.is_some()),
1322            _ => unreachable!("wrong response to known block request"),
1323        }
1324    }
1325
1326    fn update_metrics(&mut self) {
1327        metrics::gauge!("sync.prospective_tips.len",).set(self.prospective_tips.len() as f64);
1328        metrics::gauge!("sync.downloads.in_flight",).set(self.downloads.in_flight() as f64);
1329    }
1330
1331    /// Return if the sync should be restarted based on the given error
1332    /// from the block downloader and verifier stream.
1333    fn should_restart_sync(e: &BlockDownloadVerifyError) -> bool {
1334        match e {
1335            // Structural matches: downcasts
1336            BlockDownloadVerifyError::Invalid { error, .. } if error.is_duplicate_request() => {
1337                debug!(error = ?e, "block was already verified or committed, possibly from a previous sync run, continuing");
1338                false
1339            }
1340
1341            // Structural matches: direct
1342            BlockDownloadVerifyError::CancelledDuringDownload { .. }
1343            | BlockDownloadVerifyError::CancelledDuringVerification { .. } => {
1344                debug!(error = ?e, "block verification was cancelled, continuing");
1345                false
1346            }
1347            BlockDownloadVerifyError::BehindTipHeightLimit { .. } => {
1348                debug!(
1349                    error = ?e,
1350                    "block height is behind the current state tip, \
1351                     assuming the syncer will eventually catch up to the state, continuing"
1352                );
1353                false
1354            }
1355            BlockDownloadVerifyError::AboveLookaheadHeightLimit { .. } => {
1356                debug!(
1357                    error = ?e,
1358                    "block height is above the lookahead limit, \
1359                     dropping the block and continuing sync"
1360                );
1361                false
1362            }
1363            BlockDownloadVerifyError::InvalidHeight { .. } => {
1364                debug!(
1365                    error = ?e,
1366                    "block has no valid height, \
1367                     dropping the block and continuing sync"
1368                );
1369                false
1370            }
1371            BlockDownloadVerifyError::DuplicateBlockQueuedForDownload { .. } => {
1372                debug!(
1373                    error = ?e,
1374                    "queued duplicate block hash for download, \
1375                     assuming the syncer will eventually resolve duplicates, continuing"
1376                );
1377                false
1378            }
1379
1380            BlockDownloadVerifyError::DownloadFailed { ref error, .. }
1381                if format!("{error:?}").contains("NotFound") =>
1382            {
1383                // Covers these errors:
1384                // - NotFoundResponse
1385                // - NotFoundRegistry
1386                //
1387                // TODO: improve this by checking the type (#2908)
1388                //       restart after a certain number of NotFound errors?
1389                debug!(error = ?e, "block was not found, possibly from a peer that doesn't have the block yet, continuing");
1390                false
1391            }
1392
1393            _ => {
1394                // download_and_verify downcasts errors from the block verifier
1395                // into VerifyChainError, and puts the result inside one of the
1396                // BlockDownloadVerifyError enumerations. This downcast could
1397                // become incorrect e.g. after some refactoring, and it is difficult
1398                // to write a test to check it. The test below is a best-effort
1399                // attempt to catch if that happens and log it.
1400                //
1401                // TODO: add a proper test and remove this
1402                // https://github.com/ZcashFoundation/zebra/issues/2909
1403                let err_str = format!("{e:?}");
1404                if err_str.contains("NotFound") {
1405                    error!(?e,
1406                        "a BlockDownloadVerifyError that should have been filtered out was detected, \
1407                        which possibly indicates a programming error in the downcast inside \
1408                        zebrad::components::sync::downloads::Downloads::download_and_verify"
1409                    )
1410                }
1411
1412                warn!(?e, "error downloading and verifying block");
1413                true
1414            }
1415        }
1416    }
1417}