ckb-chain 1.2.2

CKB chain service
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138

use crate::utils::orphan_block_pool::EXPIRED_EPOCH;
use crate::{ChainController, LonelyBlock};
use ckb_constant::sync::BLOCK_DOWNLOAD_WINDOW;
use ckb_db::{Direction, IteratorMode};
use ckb_db_schema::COLUMN_NUMBER_HASH;
use ckb_logger::{error, info};
use ckb_shared::Shared;
use ckb_stop_handler::has_received_stop_signal;
use ckb_store::ChainStore;
use ckb_types::core::{BlockNumber, BlockView};
use ckb_types::packed;
use ckb_types::prelude::{Entity, FromSliceShouldBeOk, Reader};
use std::cmp;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;

pub(crate) struct InitLoadUnverified {
    shared: Shared,
    chain_controller: ChainController,
    is_verifying_unverified_blocks_on_startup: Arc<AtomicBool>,
}

impl InitLoadUnverified {
    pub(crate) fn new(
        shared: Shared,
        chain_controller: ChainController,
        is_verifying_unverified_blocks_on_startup: Arc<AtomicBool>,
    ) -> Self {
        InitLoadUnverified {
            shared,
            chain_controller,
            is_verifying_unverified_blocks_on_startup,
        }
    }

    fn find_unverified_block_hashes(&self, check_unverified_number: u64) -> Vec<packed::Byte32> {
        let pack_number: packed::Uint64 = check_unverified_number.into();
        let prefix = pack_number.as_slice();

        // If a block has `COLUMN_NUMBER_HASH` but not `BlockExt`,
        // it indicates an unverified block inserted during the last shutdown.
        let unverified_hashes: Vec<packed::Byte32> = self
            .shared
            .store()
            .get_iter(
                COLUMN_NUMBER_HASH,
                IteratorMode::From(prefix, Direction::Forward),
            )
            .take_while(|(key, _)| key.starts_with(prefix))
            .map(|(key_number_hash, _v)| {
                let reader =
                    packed::NumberHashReader::from_slice_should_be_ok(key_number_hash.as_ref());

                reader.block_hash().to_entity()
            })
            .filter(|hash| self.shared.store().get_block_ext(hash).is_none())
            .collect::<Vec<packed::Byte32>>();
        unverified_hashes
    }

    pub(crate) fn start(&self) {
        info!(
            "finding unverified blocks, current tip: {}-{}",
            self.shared.snapshot().tip_number(),
            self.shared.snapshot().tip_hash()
        );

        self.find_and_verify_unverified_blocks();

        self.is_verifying_unverified_blocks_on_startup
            .store(false, std::sync::atomic::Ordering::Release);
        info!("find unverified blocks finished");
    }

    fn find_unverified_blocks<F>(&self, f: F)
    where
        F: Fn(&packed::Byte32),
    {
        let tip_number: BlockNumber = self.shared.snapshot().tip_number();
        let start_check_number = cmp::max(
            1,
            tip_number.saturating_sub(EXPIRED_EPOCH * self.shared.consensus().max_epoch_length()),
        );
        let Some(end_check_number) = tip_number.checked_add(BLOCK_DOWNLOAD_WINDOW * 10) else {
            error!(
                "unverified block scan end overflows: tip_number {}, window {}",
                tip_number,
                BLOCK_DOWNLOAD_WINDOW * 10
            );
            return;
        };

        for check_unverified_number in start_check_number..=end_check_number {
            if has_received_stop_signal() {
                info!("init_unverified_blocks thread received exit signal, exit now");
                return;
            }

            // start checking `check_unverified_number` have COLUMN_NUMBER_HASH value in db?
            let unverified_hashes: Vec<packed::Byte32> =
                self.find_unverified_block_hashes(check_unverified_number);

            if check_unverified_number > tip_number && unverified_hashes.is_empty() {
                info!(
                    "no unverified blocks found after tip, current tip: {}-{}",
                    tip_number,
                    self.shared.snapshot().tip_hash()
                );
                return;
            }

            for unverified_hash in unverified_hashes {
                f(&unverified_hash);
            }
        }
    }

    fn find_and_verify_unverified_blocks(&self) {
        self.find_unverified_blocks(|unverified_hash| {
            let unverified_block: BlockView = self
                .shared
                .store()
                .get_block(unverified_hash)
                .expect("unverified block must be in db");

            if has_received_stop_signal() {
                return;
            }

            self.chain_controller
                .asynchronous_process_lonely_block(LonelyBlock {
                    block: Arc::new(unverified_block),
                    switch: None,
                    verify_callback: None,
                });
        });
    }
}