yellowstone_block_machine/
wrapper.rs

1use {
2    crate::{
3        event::{BlockMetaEvInfo, EntryEvInfo, GeyserEventInfo, SlotStatusKind, SlotUpdateEvInfo},
4        forks::Forks,
5        state_machine::{
6            BlockStateMachineOutput, BlockSummary, BlocksStateMachine, DeadletterEvent, EntryInfo,
7            SlotCommitmentStatusUpdate, SlotLifecycle, SlotLifecycleUpdate, UntrackedSlot,
8        },
9    },
10    solana_clock::Slot,
11    solana_commitment_config::CommitmentLevel,
12    solana_hash::Hash,
13};
14
15const STATE_MACHINE_GC_EVERY_COMPLETED_SLOTS: usize = 10;
16
17///
18/// The core state machine that processes incoming Geyser events and produces block machine outputs.
19///
20/// Mainly a Wrapper to translate Grpc events to State Machine events.
21///
22#[derive(Debug, Default)]
23pub struct BlocksStateMachineWrapper {
24    pub sm: BlocksStateMachine,
25    completed_slots_since_last_gc: usize,
26    slot_gc_tracer: Option<Vec<Slot>>,
27}
28
29impl From<EntryEvInfo> for EntryInfo {
30    fn from(value: EntryEvInfo) -> Self {
31        Self {
32            entry_hash: Hash::new_from_array(value.hash),
33            slot: value.slot,
34            entry_index: value.index,
35            starting_txn_index: value.starting_transaction_index,
36            executed_txn_count: value.executed_transaction_count,
37        }
38    }
39}
40
41impl BlocksStateMachineWrapper {
42    pub fn new() -> Self {
43        Self {
44            sm: BlocksStateMachine::default(),
45            completed_slots_since_last_gc: 0,
46            slot_gc_tracer: None,
47        }
48    }
49
50    pub fn new_with_slot_gc_tracing() -> Self {
51        Self {
52            sm: BlocksStateMachine::default(),
53            completed_slots_since_last_gc: 0,
54            slot_gc_tracer: Some(Vec::with_capacity(10)),
55        }
56    }
57
58    ///
59    /// Pops the next slot that has been garbage collected by the state machine, if slot GC tracing is enabled.
60    ///
61    #[inline]
62    pub fn pop_slot_gc_trace(&mut self) -> Option<Slot> {
63        let tracer = self.slot_gc_tracer.as_mut()?;
64        tracer.pop()
65    }
66
67    fn maybe_run_gc_after_completed_slot(&mut self) {
68        self.completed_slots_since_last_gc += 1;
69        if self.completed_slots_since_last_gc >= STATE_MACHINE_GC_EVERY_COMPLETED_SLOTS {
70            self.sm.gc(self.slot_gc_tracer.as_mut());
71            self.completed_slots_since_last_gc = 0;
72        }
73    }
74
75    pub fn handle_block_entry(&mut self, entry: EntryEvInfo) -> Result<(), UntrackedSlot> {
76        let entry_info: EntryInfo = entry.into();
77        self.sm.process_replay_event(entry_info.into())
78    }
79
80    #[allow(clippy::collapsible_else_if)]
81    pub fn handle_slot_update(
82        &mut self,
83        slot_update: SlotUpdateEvInfo,
84    ) -> Result<(), UntrackedSlot> {
85        const LIFE_CYCLE_STATUS: [SlotStatusKind; 4] = [
86            SlotStatusKind::FirstShredReceived,
87            SlotStatusKind::Completed,
88            SlotStatusKind::CreatedBank,
89            SlotStatusKind::Dead,
90        ];
91
92        if LIFE_CYCLE_STATUS.contains(&slot_update.status) {
93            let lifecycle_update = SlotLifecycleUpdate {
94                slot: slot_update.slot,
95                parent_slot: slot_update.parent,
96                stage: match slot_update.status {
97                    SlotStatusKind::FirstShredReceived => SlotLifecycle::FirstShredReceived,
98                    SlotStatusKind::Completed => SlotLifecycle::Completed,
99                    SlotStatusKind::CreatedBank => SlotLifecycle::CreatedBank,
100                    SlotStatusKind::Dead => SlotLifecycle::Dead,
101                    _ => unreachable!(),
102                },
103            };
104            self.sm.process_replay_event(lifecycle_update.into())?;
105        } else {
106            if slot_update.dead_error {
107                // Downgrade to lifecycle update
108                let lifecycle_update = SlotLifecycleUpdate {
109                    slot: slot_update.slot,
110                    parent_slot: slot_update.parent,
111                    stage: SlotLifecycle::Dead,
112                };
113                self.sm.process_replay_event(lifecycle_update.into())?;
114            } else {
115                let commitment_level_update = SlotCommitmentStatusUpdate {
116                    parent_slot: slot_update.parent,
117                    slot: slot_update.slot,
118                    commitment: match slot_update.status {
119                        SlotStatusKind::Processed => CommitmentLevel::Processed,
120                        SlotStatusKind::Confirmed => CommitmentLevel::Confirmed,
121                        SlotStatusKind::Finalized => CommitmentLevel::Finalized,
122                        _ => unreachable!(),
123                    },
124                };
125
126                self.sm
127                    .process_consensus_event(commitment_level_update.into());
128            }
129        }
130        Ok(())
131    }
132
133    pub fn handle_block_meta(&mut self, block_meta: BlockMetaEvInfo) -> Result<(), UntrackedSlot> {
134        let block_summary = BlockSummary {
135            slot: block_meta.slot,
136            entry_count: block_meta.entries_count,
137            parent_slot: block_meta.parent_slot,
138            executed_transaction_count: block_meta.executed_transaction_count,
139            blockhash: Hash::new_from_array(block_meta.blockhash),
140        };
141        self.sm.process_replay_event(block_summary.into())
142        // Currently not used in block reconstruction
143    }
144
145    pub fn pop_next_state_machine_output(&mut self) -> Option<BlockStateMachineOutput> {
146        let output = self.sm.pop_next_unprocess_blockstore_update()?;
147        if matches!(output, BlockStateMachineOutput::FrozenBlock(_)) {
148            self.maybe_run_gc_after_completed_slot();
149        }
150        Some(output)
151    }
152
153    pub fn fork_graph(&self) -> &Forks<Slot> {
154        &self.sm.forks
155    }
156
157    #[inline]
158    pub fn pop_next_dlq(&mut self) -> Option<DeadletterEvent> {
159        self.sm.pop_next_dlq()
160    }
161
162    pub fn handle_new_geyser_event(&mut self, event: GeyserEventInfo) -> Result<(), UntrackedSlot> {
163        match event {
164            GeyserEventInfo::Slot(slot_update) => self.handle_slot_update(slot_update),
165            GeyserEventInfo::BlockMeta(block_meta) => self.handle_block_meta(block_meta),
166            GeyserEventInfo::Entry(entry) => self.handle_block_entry(entry),
167            GeyserEventInfo::Transaction { slot }
168            | GeyserEventInfo::Account { slot }
169            | GeyserEventInfo::Other { slot } => {
170                if !self.sm.is_slot_tracked(slot) {
171                    return Err(UntrackedSlot);
172                }
173                Ok(())
174            }
175        }
176    }
177}
yellowstone_block_machine/wrapper.rs

yellowstone_block_machine/
wrapper.rs
