clone_solana_rpc/
block_meta_service.rs

1//! The `BlockMetaService` is responsible for persisting block metadata from
2//! banks into the `Blockstore`
3
4pub use clone_solana_ledger::blockstore_processor::BlockMetaSender;
5use {
6    clone_solana_ledger::blockstore::{Blockstore, BlockstoreError},
7    clone_solana_runtime::bank::{Bank, KeyedRewardsAndNumPartitions},
8    clone_solana_transaction_status::{Reward, RewardsAndNumPartitions},
9    crossbeam_channel::{Receiver, RecvTimeoutError},
10    std::{
11        sync::{
12            atomic::{AtomicBool, AtomicU64, Ordering},
13            Arc,
14        },
15        thread::{self, Builder, JoinHandle},
16        time::Duration,
17    },
18};
19
20pub type BlockMetaReceiver = Receiver<Arc<Bank>>;
21
22pub struct BlockMetaService {
23    thread_hdl: JoinHandle<()>,
24}
25
26impl BlockMetaService {
27    pub fn new(
28        block_meta_receiver: BlockMetaReceiver,
29        blockstore: Arc<Blockstore>,
30        max_complete_rewards_slot: Arc<AtomicU64>,
31        exit: Arc<AtomicBool>,
32    ) -> Self {
33        let thread_hdl = Builder::new()
34            .name("solBlockMeta".to_string())
35            .spawn(move || {
36                info!("BlockMetaService has started");
37                loop {
38                    if exit.load(Ordering::Relaxed) {
39                        break;
40                    }
41
42                    let bank = match block_meta_receiver.recv_timeout(Duration::from_secs(1)) {
43                        Ok(bank) => bank,
44                        Err(RecvTimeoutError::Timeout) => continue,
45                        Err(err @ RecvTimeoutError::Disconnected) => {
46                            info!("BlockMetaService is stopping because: {err}");
47                            break;
48                        }
49                    };
50
51                    if let Err(err) =
52                        Self::write_block_meta(&bank, &blockstore, &max_complete_rewards_slot)
53                    {
54                        error!("BlockMetaService is stopping because: {err}");
55                        // Set the exit flag to allow other services to gracefully stop
56                        exit.store(true, Ordering::Relaxed);
57                        break;
58                    }
59                }
60                info!("BlockMetaService has stopped");
61            })
62            .unwrap();
63        Self { thread_hdl }
64    }
65
66    fn write_block_meta(
67        bank: &Bank,
68        blockstore: &Blockstore,
69        max_complete_rewards_slot: &Arc<AtomicU64>,
70    ) -> Result<(), BlockstoreError> {
71        let slot = bank.slot();
72
73        blockstore.set_block_time(slot, bank.clock().unix_timestamp)?;
74        blockstore.set_block_height(slot, bank.block_height())?;
75
76        let rewards = bank.get_rewards_and_num_partitions();
77        if rewards.should_record() {
78            let KeyedRewardsAndNumPartitions {
79                keyed_rewards,
80                num_partitions,
81            } = rewards;
82            let rewards = keyed_rewards
83                .into_iter()
84                .map(|(pubkey, reward_info)| Reward {
85                    pubkey: pubkey.to_string(),
86                    lamports: reward_info.lamports,
87                    post_balance: reward_info.post_balance,
88                    reward_type: Some(reward_info.reward_type),
89                    commission: reward_info.commission,
90                })
91                .collect();
92            let blockstore_rewards = RewardsAndNumPartitions {
93                rewards,
94                num_partitions,
95            };
96
97            blockstore.write_rewards(slot, blockstore_rewards)?;
98        }
99        max_complete_rewards_slot.fetch_max(slot, Ordering::SeqCst);
100
101        Ok(())
102    }
103
104    pub fn join(self) -> thread::Result<()> {
105        self.thread_hdl.join()
106    }
107}
clone_solana_rpc/block_meta_service.rs

clone_solana_rpc/
block_meta_service.rs
