yellowstone_block_machine/
state_machine.rs

1use {
2    crate::forks::{Forks, ForksMutationTracer},
3    derive_more::From,
4    rustc_hash::{FxHashMap, FxHashSet},
5    serde::{Deserialize, Serialize},
6    solana_clock::Slot,
7    solana_commitment_config::CommitmentLevel,
8    solana_hash::Hash,
9    std::{
10        collections::VecDeque,
11        time::{Duration, Instant},
12    },
13};
14
15#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
16pub enum SlotLifecycle {
17    FirstShredReceived,
18    CreatedBank,
19    Completed,
20    Dead,
21}
22
23#[derive(Debug, Clone)]
24pub struct SlotCommitmentStatusUpdate {
25    pub parent_slot: Option<Slot>,
26    pub slot: Slot,
27    pub commitment: CommitmentLevel,
28}
29
30#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
31pub struct SlotLifecycleUpdate {
32    pub parent_slot: Option<Slot>,
33    pub slot: Slot,
34    pub stage: SlotLifecycle,
35}
36
37pub struct BlockstorePublisherConfig {
38    pub linger: u64,
39    pub max_batch_size: bytesize::ByteSize,
40}
41
42#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
43pub struct BlockSummary {
44    pub slot: Slot,
45    pub entry_count: u64,
46    pub parent_slot: Slot,
47    pub executed_transaction_count: u64,
48    pub blockhash: Hash,
49}
50
51// #[derive(Debug, From)]
52// pub enum BlockstoreInputEvent {
53//     Entry(EntryInfo),
54//     SlotCommitmentStatus(SlotCommitmentStatusUpdate),
55//     SlotLifecycleStatus(SlotLifecycleUpdate),
56//     BlockSummary(BlockSummary),
57// }
58
59#[derive(Debug, Clone, From)]
60pub enum BlockReplayEvent {
61    SlotLifecycleStatus(SlotLifecycleUpdate),
62    Entry(EntryInfo),
63    BlockSummary(BlockSummary),
64}
65
66#[derive(Debug, Clone, From)]
67pub enum ConsensusUpdate {
68    SlotCommitmentStatus(SlotCommitmentStatusUpdate),
69}
70
71pub type InnerBlockSequence = i64;
72
73#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
74pub struct EntryInfo {
75    pub slot: Slot,
76    pub entry_index: u64,
77    pub starting_txn_index: u64,
78    pub entry_hash: Hash,
79    pub executed_txn_count: u64,
80}
81
82///
83/// Block under construction
84///
85#[derive(Debug)]
86pub struct Block {
87    pub slot: Slot,
88    entries: FxHashMap<u64, EntryInfo>,
89    entry_cnt: u64,
90    tick_entry_cnt: u64,
91    created_at: std::time::Instant,
92}
93
94#[derive(Clone, Serialize, Deserialize, Debug, PartialEq)]
95pub struct FrozenBlock {
96    pub slot: Slot,
97    pub parent_slot: Slot,
98    pub entries: Vec<EntryInfo>,
99    pub blockhash: Hash,
100}
101
102// Avg 2k tx + 2k account update
103pub const AVG_BLOCK_LEN: usize = 4000;
104// Avg transaction per block ~ 2000;
105pub const AVG_TPB: usize = 2000;
106
107///
108///
109/// State machine for buffering blockstore events based on linger and other buffering limits.
110///
111/// Implements a sans-IO API for better easier testability and better separation of concerns.
112///
113/// If you want to use this in a real-world scenario, you need to implement your own sync/async driver
114/// to progress in the same machine.
115///
116impl Block {
117    pub fn new_with_clock(slot: Slot, clock: Instant) -> Self {
118        Self {
119            slot,
120            entries: Default::default(),
121            created_at: clock,
122            entry_cnt: 0,
123            tick_entry_cnt: 0,
124        }
125    }
126
127    pub fn new(slot: Slot) -> Self {
128        Self::new_with_clock(slot, Instant::now())
129    }
130
131    fn last_entry_hash(&self) -> Option<Hash> {
132        self.entries
133            .get(&(self.entry_cnt - 1))
134            .map(|entry| entry.entry_hash)
135    }
136
137    fn freeze(self, summary: &BlockSummary) -> FrozenBlock {
138        FrozenBlock {
139            slot: self.slot,
140            entries: self.entries.values().cloned().collect(),
141            blockhash: summary.blockhash,
142            parent_slot: summary.parent_slot,
143        }
144    }
145
146    fn can_be_optimistic_frozen(&self) -> bool {
147        if self.entry_cnt == 0 {
148            return false;
149        }
150
151        (0..self.entry_cnt).all(|idx| self.entries.contains_key(&idx))
152    }
153
154    fn forge_optimistic_block_summary(&self, parent_slot: Slot) -> BlockSummary {
155        BlockSummary {
156            slot: self.slot,
157            parent_slot,
158            entry_count: self.entry_cnt,
159            executed_transaction_count: self.entries.values().map(|e| e.executed_txn_count).sum(),
160            blockhash: self.last_entry_hash().expect("last entry hash"),
161        }
162    }
163
164    fn insert_entry(&mut self, block_entry: EntryInfo) {
165        let entry_idx = block_entry.entry_index;
166        let tx_count = block_entry.executed_txn_count;
167        if self.entries.insert(entry_idx, block_entry).is_none() {
168            self.entry_cnt += 1;
169
170            if tx_count == 0 {
171                self.tick_entry_cnt += 1;
172            }
173        }
174    }
175}
176
177type Revision = usize;
178
179///
180/// Block State machine
181///
182/// This state machine is responsible for buffering blockstore events and provide a way to consume them in a controlled manner.
183///
184/// The entire blockstore state is logic is SANS-IO and must stay like this to ensure that we can test it properly.
185///
186///
187/// IMPORTANT: SANS-IO means there is no IO operations in this state machine or async function that include channel communication.
188///
189/// Is to the developer to implement the IO part of the state machine by implementing your own "driver".
190///
191#[derive(Debug)]
192pub struct BlocksStateMachine {
193    /// Holds block under construction, not yet frozen
194    block_buffer_map: FxHashMap<Slot, Block>,
195
196    /// Holds blocks that are frozen
197    frozen_block_index: FxHashMap<Slot, FxHashSet<CommitmentLevel>>,
198
199    /// Maps when it is safe to "deregister" slot.
200    /// Deregistering a slot is removing all index data about this slot as it cannot be reference by downstream consumer.
201    /// We handle finalized block separately since they are the easiest to deregister.
202    deregister_finalized_slot_schedule: FxHashMap<Revision, Vec<Slot>>,
203
204    /// Data from deregister_slot_schedule goes into a queue since we may want to process it in the future.
205    pending_finalized_slot_deregister: VecDeque<Slot>,
206
207    /// We queue slot status update if the slot is not yet in the frozen block index.
208    /// This is to provide nice guarantee such that you will read the entire block
209    /// before any slot status update.
210    pending_slot_status_update: FxHashMap<Slot, VecDeque<SlotCommitmentStatusUpdate>>,
211
212    /// Represents the passage of time when blockstore update occurs.
213    /// this can be useful to track how things are old or schedule things in future revision.
214    revision: Revision,
215
216    /// Keep track of the minimum revision [`blockstore_update_queue`].
217    min_history_revision_in_queue: Option<usize>,
218
219    /// Update queue for blockstore events.
220    blockstore_update_queue: VecDeque<(usize, BlockStateMachineOutput)>,
221
222    /// Maintain forks history of the blockchain.    
223    pub forks: Forks<Slot>,
224    forks_history: FxHashSet<Slot>,
225
226    ///
227    /// Keep track of the slots that were detected as forks in the current tick.
228    ///
229    forks_detected_in_current_tick: FxHashSet<Slot>,
230
231    /// Holds deadletter queue message for blockstore events that cannot be processed.
232    dlq: VecDeque<DeadletterEvent>,
233
234    /// Keep track of the age of the slot (when the [`Block`] was first created in the event history).
235    slot_age: FxHashMap<Slot, Revision>,
236
237    /// Keep track of the maximum revision event referenced by slot.
238    slot_max_version_referenced: FxHashMap<Slot, Revision>,
239
240    /// Dead blocks are blocks who will never be frozen.
241    /// This can happen when we boot fumarole initially : some block we receive slot status update before the block data.
242    dead_blocks_queue: VecDeque<Slot>,
243
244    ///
245    /// Buffers slots that were retroactively rooted by a slot status update.
246    ///
247    retroactively_rooted_slots: FxHashSet<Slot>,
248
249    ///
250    /// Keep track of slots that need optimistic freeze because their child slot was frozen before them.
251    /// When this happen, it probably means that either agave has a bug or the geyser plugin is buggy.
252    ///
253    need_optimistic_freeze: FxHashSet<Slot>,
254}
255
256#[derive(Debug)]
257pub enum DeadletterEvent {
258    Incomplete(Slot),
259}
260
261///
262/// Dead blocks will never be frozen
263///
264pub struct InvalidBlock {
265    pub slot: Slot,
266}
267
268#[derive(Debug, Clone)]
269pub struct ForkDetected {
270    pub slot: Slot,
271}
272
273#[derive(Debug, Clone)]
274pub struct DeadBlockDetected {
275    pub slot: Slot,
276}
277
278#[derive(Debug)]
279pub enum BlockStateMachineOutput {
280    FrozenBlock(FrozenBlock),
281    SlotStatus(SlotCommitmentStatusUpdate),
282    ForksDetected(ForkDetected),
283    DeadSlotDetected(DeadBlockDetected),
284    BankCreated(Slot),
285    /// Given that Anza did not expose any information about the duplicate-unconfirmed-slot scenario, we have to assume that it can happen and
286    /// the only way to detect it is via a second BankCreated event for the same slot number.
287    BankReset(Slot),
288}
289
290impl BlockStateMachineOutput {
291    pub fn slot(&self) -> Slot {
292        match self {
293            Self::DeadSlotDetected(blk) => blk.slot,
294            Self::FrozenBlock(blk) => blk.slot,
295            Self::SlotStatus(update) => update.slot,
296            Self::ForksDetected(info) => info.slot,
297            Self::BankCreated(slot) => *slot,
298            Self::BankReset(slot) => *slot,
299        }
300    }
301}
302
303///
304/// Stats produce during [`BlockstoreSM::gc`] operation.
305///
306#[derive(Debug, Default)]
307pub struct BlockstoreGCStats {
308    // Number of slot that were purged
309    pub slot_purge_count: usize,
310    // Number of slots could potentially be purge but that were blocked by some condition.
311    pub slot_blocked_count: usize,
312}
313
314#[derive(Debug, Default, Clone)]
315pub struct BlockstoreStats {
316    pub block_buffer_len: usize,
317    pub forks_map_len: usize,
318    pub dead_block_queue_len: usize,
319    pub blockstore_update_queue_len: usize,
320}
321
322fn cmp_commitment_level(a: &CommitmentLevel, b: &CommitmentLevel) -> std::cmp::Ordering {
323    match (a, b) {
324        (CommitmentLevel::Processed, CommitmentLevel::Processed) => std::cmp::Ordering::Equal,
325        (CommitmentLevel::Finalized, CommitmentLevel::Finalized) => std::cmp::Ordering::Equal,
326        (CommitmentLevel::Confirmed, CommitmentLevel::Confirmed) => std::cmp::Ordering::Equal,
327        (CommitmentLevel::Processed, _) => std::cmp::Ordering::Less,
328        (CommitmentLevel::Finalized, _) => std::cmp::Ordering::Greater,
329        (CommitmentLevel::Confirmed, CommitmentLevel::Finalized) => std::cmp::Ordering::Less,
330        (CommitmentLevel::Confirmed, CommitmentLevel::Processed) => std::cmp::Ordering::Greater,
331    }
332}
333
334fn iter_to_commitment(cl: &CommitmentLevel) -> impl Iterator<Item = CommitmentLevel> {
335    match cl {
336        CommitmentLevel::Processed => vec![CommitmentLevel::Processed].into_iter(),
337        CommitmentLevel::Confirmed => {
338            vec![CommitmentLevel::Processed, CommitmentLevel::Confirmed].into_iter()
339        }
340        CommitmentLevel::Finalized => vec![
341            CommitmentLevel::Processed,
342            CommitmentLevel::Confirmed,
343            CommitmentLevel::Finalized,
344        ]
345        .into_iter(),
346    }
347}
348
349#[derive(Debug)]
350pub struct OldestBufferedBlockInfo {
351    pub slot: Slot,
352    pub age: Duration,
353    pub parent_slot: Option<Slot>,
354    pub pending_slot_status: usize,
355}
356
357impl Default for BlocksStateMachine {
358    fn default() -> Self {
359        Self::new()
360    }
361}
362
363#[derive(Debug)]
364pub struct LongShortForksMutationTracer<'a> {
365    long: &'a mut FxHashSet<Slot>,
366    short: &'a mut FxHashSet<Slot>,
367}
368
369impl ForksMutationTracer<Slot> for LongShortForksMutationTracer<'_> {
370    fn insert(&mut self, slot: Slot) {
371        // We only insert into short if slot not already present in long.
372        if self.long.insert(slot) {
373            self.short.insert(slot);
374        }
375    }
376}
377
378///
379/// Occurred when a replay event is rejected by the state machine because it relates to a slot that cannot be tracked by the state machine.
380///
381#[derive(Debug, thiserror::Error)]
382#[error("replay event rejected")]
383pub struct UntrackedSlot;
384
385impl BlocksStateMachine {
386    ///
387    /// Creates a new blockstore instance
388    ///
389    pub fn new() -> Self {
390        Self {
391            block_buffer_map: Default::default(),
392            frozen_block_index: Default::default(),
393            pending_slot_status_update: Default::default(),
394            blockstore_update_queue: VecDeque::with_capacity(1000),
395            revision: 0,
396            min_history_revision_in_queue: Default::default(),
397            dlq: Default::default(),
398            slot_age: Default::default(),
399            slot_max_version_referenced: Default::default(),
400            deregister_finalized_slot_schedule: Default::default(),
401            pending_finalized_slot_deregister: Default::default(),
402            forks: Default::default(),
403            forks_history: Default::default(),
404            dead_blocks_queue: Default::default(),
405            retroactively_rooted_slots: Default::default(),
406            forks_detected_in_current_tick: Default::default(),
407            need_optimistic_freeze: Default::default(),
408        }
409    }
410
411    pub fn stats(&self) -> BlockstoreStats {
412        BlockstoreStats {
413            block_buffer_len: self.block_buffer_map.len(),
414            forks_map_len: self.forks.len(),
415            dead_block_queue_len: self.dead_blocks_queue.len(),
416            blockstore_update_queue_len: self.blockstore_update_queue.len(),
417        }
418    }
419
420    fn next_history_revision(&mut self) -> usize {
421        let temp = self.revision;
422        self.revision += 1;
423        temp
424    }
425
426    fn push_new_update(&mut self, update: BlockStateMachineOutput) -> Revision {
427        let new_revision = self.next_history_revision();
428        let slot = update.slot();
429        let max_revision = self
430            .slot_max_version_referenced
431            .entry(slot)
432            .or_insert(new_revision);
433        *max_revision = std::cmp::max(*max_revision, new_revision);
434        self.blockstore_update_queue
435            .push_back((new_revision, update));
436        new_revision
437    }
438
439    /// Pending slot update are pending because the slot is not yet in the frozen block index.
440    fn flush_pending_slot_status_update(&mut self, slot: Slot) {
441        if let Some(updates) = self.pending_slot_status_update.remove(&slot) {
442            for mut slot_status in updates {
443                if slot_status.parent_slot.is_none() {
444                    if let Some(parent) = self.forks.get_parent(&slot_status.slot) {
445                        slot_status.parent_slot = Some(parent);
446                    }
447                }
448                self.handle_slot_commitment_status_update(slot_status);
449            }
450        }
451    }
452
453    pub fn is_slot_tracked(&self, slot: Slot) -> bool {
454        self.frozen_block_index.contains_key(&slot) || self.block_buffer_map.contains_key(&slot)
455    }
456
457    fn handle_slot_lifecyle_status(
458        &mut self,
459        slot_lifecycle_status: SlotLifecycleUpdate,
460    ) -> Result<(), UntrackedSlot> {
461        let slot = slot_lifecycle_status.slot;
462
463        if let Some(parent) = slot_lifecycle_status.parent_slot {
464            let mut multiset = LongShortForksMutationTracer {
465                long: &mut self.forks_history,
466                short: &mut self.forks_detected_in_current_tick,
467            };
468            self.forks.add_slot_with_parent_with_rooted_trace(
469                slot,
470                parent,
471                &mut multiset,
472                &mut self.retroactively_rooted_slots,
473            );
474        }
475
476        match slot_lifecycle_status.stage {
477            SlotLifecycle::FirstShredReceived => {
478                tracing::trace!("First shred received for slot {}", slot);
479                match self.block_buffer_map.entry(slot) {
480                    std::collections::hash_map::Entry::Vacant(vacant_entry) => {
481                        let block = Block::new(slot);
482                        vacant_entry.insert(block);
483                    }
484                    _ => {
485                        // Maybe duplicate events
486                    }
487                }
488            }
489            SlotLifecycle::CreatedBank => {
490                tracing::trace!("Bank created for slot {}", slot);
491                // In case of duplicate unconfirmed slot (replay of old slot), we may receive multiple "CreatedBank" event for the same slot.
492                if self
493                    .block_buffer_map
494                    .insert(slot, Block::new(slot))
495                    .is_none()
496                {
497                    // New bank created
498                    if self.frozen_block_index.contains_key(&slot) {
499                        self.push_new_update(BlockStateMachineOutput::BankReset(slot));
500                    } else {
501                        self.push_new_update(BlockStateMachineOutput::BankCreated(slot));
502                    }
503                } else {
504                    self.push_new_update(BlockStateMachineOutput::BankReset(slot));
505                }
506                if let Some(pending) = self.pending_slot_status_update.get_mut(&slot) {
507                    pending
508                        .retain(|slot_status| slot_status.commitment != CommitmentLevel::Processed);
509                }
510                if let Some(visited_commitment) = self.frozen_block_index.get_mut(&slot) {
511                    visited_commitment.remove(&CommitmentLevel::Processed);
512                }
513            }
514            SlotLifecycle::Completed => {
515                if !self.block_buffer_map.contains_key(&slot) {
516                    tracing::trace!("Slot {} is not in the block buffer map, skipping", slot);
517                    return Err(UntrackedSlot);
518                }
519            }
520            SlotLifecycle::Dead => {
521                self.mark_block_as_dead(slot);
522            }
523        }
524        Ok(())
525    }
526
527    fn handle_slot_commitment_status_update(
528        &mut self,
529        mut slot_status: SlotCommitmentStatusUpdate,
530    ) {
531        let slot = slot_status.slot;
532
533        if let Some(parent) = slot_status.parent_slot {
534            let mut multitrace = LongShortForksMutationTracer {
535                long: &mut self.forks_history,
536                short: &mut self.forks_detected_in_current_tick,
537            };
538            self.forks.add_slot_with_parent_with_rooted_trace(
539                slot,
540                parent,
541                &mut multitrace,
542                &mut self.retroactively_rooted_slots,
543            );
544        } else {
545            // If for some reason we already know the parent, just inject it into the slot status.
546            if let Some(parent) = self.forks.get_parent(&slot_status.slot) {
547                slot_status.parent_slot = Some(parent);
548            }
549        }
550
551        if !self.frozen_block_index.contains_key(&slot)
552            && !self.block_buffer_map.contains_key(&slot)
553        {
554            return;
555        }
556
557        match self.frozen_block_index.get_mut(&slot_status.slot) {
558            Some(visited_commitment) => {
559                let commitment = slot_status.commitment;
560                let mut slot_status_to_push = vec![];
561                for commitment2 in iter_to_commitment(&commitment) {
562                    if visited_commitment.insert(commitment2) {
563                        slot_status_to_push.push(SlotCommitmentStatusUpdate {
564                            parent_slot: slot_status.parent_slot,
565                            slot,
566                            commitment: commitment2,
567                        });
568                    }
569                }
570                // This code handle a tricky part where slot status update of higher level commitment level may arrive
571                // before lower level commitment.
572                // By iterating from lower to higher level commitment, we can ensure that we don't miss any slot status update.
573                // Why? Dragonsmouth is already doing something similar, I do it here to ensure that we don't miss any slot status update
574                // if we change data source that may not guarantee the order of slot status update.
575                for slot_status2 in slot_status_to_push {
576                    let revision =
577                        self.push_new_update(BlockStateMachineOutput::SlotStatus(slot_status2));
578                    tracing::debug!(
579                        "Slot status update for slot {} at revision {}",
580                        slot,
581                        revision
582                    );
583                    if commitment == CommitmentLevel::Finalized {
584                        let mut multiset = LongShortForksMutationTracer {
585                            long: &mut self.forks_history,
586                            short: &mut self.forks_detected_in_current_tick,
587                        };
588
589                        self.forks.make_slot_rooted_with_rooted_trace(
590                            slot,
591                            &mut multiset,
592                            &mut self.retroactively_rooted_slots,
593                        );
594                        self.deregister_finalized_slot_schedule
595                            .entry(revision)
596                            .or_default()
597                            .push(slot);
598                    }
599                }
600            }
601            _ => {
602                if self.block_buffer_map.contains_key(&slot) {
603                    self.pending_slot_status_update
604                        .entry(slot_status.slot)
605                        .or_default()
606                        .push_back(slot_status);
607                } else {
608                    unreachable!("checks at the beginning of the function should prevent this");
609                }
610            }
611        }
612    }
613
614    ///
615    /// Mark a block as dead.
616    ///
617    fn mark_block_as_dead(&mut self, slot: Slot) {
618        let mut multitrace = LongShortForksMutationTracer {
619            long: &mut self.forks_history,
620            short: &mut self.forks_detected_in_current_tick,
621        };
622        self.forks.mark_slot_as_forked(slot, &mut multitrace);
623        self.remove_slot_references_in_state(slot);
624    }
625
626    fn handle_block_entry_insert(&mut self, data: EntryInfo) -> Result<(), UntrackedSlot> {
627        let slot = data.slot;
628        let Some(buffer) = self.block_buffer_map.get_mut(&slot) else {
629            // If the block container has not been created yet, it means we never received FIRST_SHRED.
630            // Therefore we cannot insert the block data.
631            return Err(UntrackedSlot);
632        };
633        buffer.insert_entry(data);
634        Ok(())
635    }
636
637    #[inline]
638    fn push_to_dlq(&mut self, msg: DeadletterEvent) {
639        self.dlq.push_back(msg);
640    }
641
642    fn process_retroactively_rooted_slots(&mut self) {
643        if self.retroactively_rooted_slots.is_empty() {
644            return;
645        }
646        let retroactively_rooted_slots = std::mem::take(&mut self.retroactively_rooted_slots);
647        for slot in retroactively_rooted_slots {
648            tracing::trace!("Retroactively rooting slot {}", slot);
649            self.handle_slot_commitment_status_update(SlotCommitmentStatusUpdate {
650                slot,
651                parent_slot: self.forks.get_parent(&slot),
652                commitment: CommitmentLevel::Finalized,
653            });
654        }
655    }
656
657    fn flush_forks_detected_in_current_tick(&mut self) {
658        if self.forks_detected_in_current_tick.is_empty() {
659            return;
660        }
661        let forks_detected = std::mem::take(&mut self.forks_detected_in_current_tick);
662        for slot in forks_detected {
663            tracing::warn!("Forks detected for slot {}", slot);
664            self.push_new_update(BlockStateMachineOutput::ForksDetected(ForkDetected {
665                slot,
666            }));
667        }
668    }
669
670    ///
671    /// Freeze block when we receive block summary.
672    ///
673    fn handle_block_summary(&mut self, block_summary: BlockSummary) -> Result<(), UntrackedSlot> {
674        let slot = block_summary.slot;
675        let Some(block) = self.block_buffer_map.remove(&slot) else {
676            tracing::debug!("Block summary for slot {slot} but no block data found",);
677            return Err(UntrackedSlot);
678        };
679
680        let frozen_block = block.freeze(&block_summary);
681        assert_eq!(slot, frozen_block.slot);
682
683        if let Some(parent) = self.forks.get_parent(&slot) {
684            if self.block_buffer_map.contains_key(&parent) {
685                tracing::warn!(
686                    "Freezing block for slot {} whose parent slot {} is still in the block buffer map",
687                    slot,
688                    parent
689                );
690                // This should never happen, but in case it does we will try to optimistically freeze the parent block.
691                self.need_optimistic_freeze.insert(parent);
692            }
693        }
694
695        tracing::debug!("Block frozen for slot {}", slot);
696        self.frozen_block_index.entry(slot).or_default();
697        self.push_new_update(BlockStateMachineOutput::FrozenBlock(frozen_block));
698
699        if let Some(max_pending_commitment_level) = self
700            .pending_slot_status_update
701            .get(&slot)
702            .iter()
703            .flat_map(|update| update.iter())
704            .filter(|update| {
705                cmp_commitment_level(&update.commitment, &CommitmentLevel::Processed).is_gt()
706            })
707            .max_by(|x, y| cmp_commitment_level(&x.commitment, &y.commitment))
708        {
709            // Sometime we may have a "processed" commitment level happens before slot is frozen,
710            // this is because "frozen" state depends on block meta which may be sent after "Processed" slot status.
711            // Technically, it should be impossible to have a commitment level higher than Processed and the slot not frozen already.
712            // We should probably panic here, but we will just log an error.
713
714            // Why we may received "Processed" slot status before the block meta?
715            // In agave code base, as slot commitment level are updated, they directly notify loaded geyser plugins, blocking the replay stage
716            // until they all acknowledge the slot status update.
717            // On the other hand, block meta message are sent of a crossbeam channel that is eventually process by a background thread, which add a bit of latency
718            // During replay, block meta is sent than right after it sends the slot status update.
719            // Most of the time, we receive the block meta before the slot status update, but in some cases, we may receive the slot status update first.
720
721            tracing::warn!(
722                "Slot {slot} froze after receiving slot status update higher than Processed: {}",
723                max_pending_commitment_level.commitment
724            );
725        }
726        self.flush_pending_slot_status_update(slot);
727        Ok(())
728    }
729
730    ///
731    /// Main entry point to process blockstore event and progress the state-machine.
732    ///
733    /// Registers the event into the state machine and process any side effect that may arise from it.
734    ///
735    pub fn process_replay_event(&mut self, event: BlockReplayEvent) -> Result<(), UntrackedSlot> {
736        match event {
737            BlockReplayEvent::BlockSummary(bs) => {
738                tracing::trace!("Inserting block summary for slot {}", bs.slot);
739                self.handle_block_summary(bs)?;
740            }
741            BlockReplayEvent::Entry(data) => {
742                self.handle_block_entry_insert(data)?;
743            }
744            BlockReplayEvent::SlotLifecycleStatus(slot_lifecycle_status) => {
745                tracing::trace!(
746                    "Inserting slot lifecycle status for slot {}",
747                    slot_lifecycle_status.slot
748                );
749                self.handle_slot_lifecyle_status(slot_lifecycle_status)?;
750            }
751        }
752
753        self.process_retroactively_rooted_slots();
754        self.flush_forks_detected_in_current_tick();
755        self.execute_optimistic_freeze_for_needed_slots();
756        Ok(())
757    }
758
759    fn execute_optimistic_freeze_for_needed_slots(&mut self) {
760        if self.need_optimistic_freeze.is_empty() {
761            return;
762        }
763        let slots_to_freeze = std::mem::take(&mut self.need_optimistic_freeze);
764        for slot in slots_to_freeze {
765            // Check if we can freeze the block : we must have some entry to compute the block hash.
766            if let Some(block) = self.block_buffer_map.get(&slot) {
767                let parent_slot = self.forks.get_parent(&slot);
768
769                match (block.can_be_optimistic_frozen(), parent_slot) {
770                    (true, Some(parent_slot)) => {
771                        let forged_block_summary =
772                            block.forge_optimistic_block_summary(parent_slot);
773                        tracing::warn!(
774                            "Recoverd block summary for slot {}: {:?}",
775                            slot,
776                            forged_block_summary
777                        );
778                        self.handle_block_summary(forged_block_summary)
779                            .expect("untracked");
780                    }
781                    _ => {
782                        tracing::error!(
783                            "Cannot optimistically freeze slot {} because it has no entries",
784                            slot
785                        );
786                        self.remove_slot_references_in_state(slot);
787                        self.push_to_dlq(DeadletterEvent::Incomplete(slot));
788                    }
789                }
790            }
791        }
792    }
793
794    pub fn process_consensus_event(&mut self, event: ConsensusUpdate) {
795        match event {
796            ConsensusUpdate::SlotCommitmentStatus(slot_status) => {
797                self.handle_slot_commitment_status_update(slot_status);
798            }
799        }
800        self.process_retroactively_rooted_slots();
801        self.flush_forks_detected_in_current_tick();
802    }
803
804    ///
805    /// Make sure the remove any reference to `slot` in all the [`BlockstoreSM`] indexes.
806    ///
807    fn remove_slot_references_in_state(&mut self, slot: Slot) {
808        self.block_buffer_map.remove(&slot);
809        self.frozen_block_index.remove(&slot);
810        self.pending_slot_status_update.remove(&slot);
811        self.slot_max_version_referenced.remove(&slot);
812        self.slot_age.remove(&slot);
813    }
814
815    ///
816    /// Returns the oldest block in the buffer.
817    ///
818    pub fn oldest_block_in_buffer(&self) -> Option<OldestBufferedBlockInfo> {
819        self.block_buffer_map
820            .values()
821            .max_by_key(|block| block.created_at.elapsed())
822            .map(|block| OldestBufferedBlockInfo {
823                slot: block.slot,
824                age: block.created_at.elapsed(),
825                parent_slot: self.forks.get_parent(&block.slot),
826                pending_slot_status: self
827                    .pending_slot_status_update
828                    .get(&block.slot)
829                    .map(|queue| queue.len())
830                    .unwrap_or_default(),
831            })
832    }
833
834    ///
835    /// This function will clean out deprecated slot index information from the state machine making sure it does not grow forever.
836    ///
837    /// Stuff to "deindex" are :
838    /// metadata about slot that are finalized -- Since they are finalized we don't need to keep them around anymore.
839    /// Forked Slot -- Slot that are forked and we know we will never reach finalized status for them.
840    ///
841    pub fn gc(&mut self, mut deleted: Option<&mut Vec<Slot>>) -> BlockstoreGCStats {
842        self.process_deregister_finalized_block_queue();
843        let mut stats = BlockstoreGCStats::default();
844        let mut elligible_for_deletion = Vec::with_capacity(self.forks_history.len());
845        let mut forks_to_remove = FxHashSet::default();
846        self.forks
847            .truncate_excess_rooted_slots(&mut forks_to_remove);
848        let oldest_rooted_slot = self.forks.oldest_rooted_slot().unwrap_or(0);
849        for slot in self.forks_history.iter() {
850            // If the oldest rooted slot that we have is bigger than current slot that we are processing than we should
851            // have received Finalized status for this slot by now.
852            // In other word this slot will never finish.
853            if *slot < oldest_rooted_slot {
854                elligible_for_deletion.push(*slot);
855            } else {
856                tracing::debug!(
857                    "Slot {} cannot be safely evicted from index because it is still part of the fork index memory",
858                    slot
859                );
860                stats.slot_blocked_count += 1;
861                continue;
862            }
863
864            if let Some(queue) = self.pending_slot_status_update.get(slot) {
865                if queue
866                    .iter()
867                    .any(|s| s.commitment == CommitmentLevel::Processed)
868                    && !forks_to_remove.contains(slot)
869                {
870                    tracing::debug!(
871                        "Slot {} cannot be safely evicted from index because pending Processed slot status",
872                        slot
873                    );
874                    stats.slot_blocked_count += 1;
875                    continue;
876                }
877            }
878            elligible_for_deletion.push(*slot);
879        }
880        stats.slot_purge_count = elligible_for_deletion.len();
881        for slot in elligible_for_deletion {
882            self.forks_history.remove(&slot);
883            self.remove_slot_references_in_state(slot);
884            if let Some(trace) = deleted.as_mut() {
885                trace.push(slot);
886            }
887        }
888        stats
889    }
890
891    ///
892    /// Process the deregistered slot from the queue.
893    ///
894    pub fn process_deregister_finalized_block_queue(&mut self) {
895        while let Some(slot) = self.pending_finalized_slot_deregister.pop_front() {
896            self.remove_slot_references_in_state(slot);
897        }
898    }
899
900    ///
901    /// Pops the next unprocessed blockstore update from the queue.
902    ///
903    pub fn pop_next_unprocess_blockstore_update(&mut self) -> Option<BlockStateMachineOutput> {
904        let (revision, data) = self.blockstore_update_queue.pop_front()?;
905        self.min_history_revision_in_queue = Some(revision + 1);
906        // Check if we need to schedule deregister process.
907        if let Some(slots) = self.deregister_finalized_slot_schedule.remove(&revision) {
908            self.pending_finalized_slot_deregister.extend(slots);
909        }
910        Some(data)
911    }
912
913    pub fn unprocess_blockstore_update_queue_len(&self) -> usize {
914        self.blockstore_update_queue.len()
915    }
916
917    pub fn pop_next_dlq(&mut self) -> Option<DeadletterEvent> {
918        self.dlq.pop_front()
919    }
920}
921
922pub fn module_path_for_test() -> &'static str {
923    module_path!()
924}
925
926#[cfg(test)]
927mod tests {
928    use {
929        crate::state_machine::{
930            BlockStateMachineOutput, BlockSummary, EntryInfo, SlotCommitmentStatusUpdate,
931            SlotLifecycle, SlotLifecycleUpdate, iter_to_commitment,
932        },
933        solana_clock::{DEFAULT_TICKS_PER_SLOT, Slot},
934        solana_commitment_config::CommitmentLevel,
935        solana_hash::Hash,
936    };
937
938    fn generate_entries(slot: Slot, num_data_entries: u64, tx_per_entry: u64) -> Vec<EntryInfo> {
939        assert!(num_data_entries >= DEFAULT_TICKS_PER_SLOT);
940        let mut entries = Vec::with_capacity((num_data_entries + DEFAULT_TICKS_PER_SLOT) as usize);
941        let tick_entry_module = num_data_entries / DEFAULT_TICKS_PER_SLOT;
942        let mut tick_entry_remain = DEFAULT_TICKS_PER_SLOT as usize;
943        for i in 0..num_data_entries {
944            let start_txn_index = i * tx_per_entry;
945            let entry = EntryInfo {
946                slot,
947                entry_index: i,
948                starting_txn_index: start_txn_index,
949                entry_hash: Hash::new_unique(),
950                executed_txn_count: tx_per_entry,
951            };
952            entries.push(entry);
953
954            if i % tick_entry_module == 0 {
955                // Add a tick entry
956                entries.push(EntryInfo {
957                    slot,
958                    entry_index: i + DEFAULT_TICKS_PER_SLOT,
959                    starting_txn_index: start_txn_index + tx_per_entry,
960                    entry_hash: Hash::new_unique(),
961                    executed_txn_count: 0, // Tick entry has no transactions
962                });
963                tick_entry_remain -= 1;
964            }
965        }
966        for _ in 0..tick_entry_remain {
967            // Add remaining tick entries
968            entries.push(EntryInfo {
969                slot,
970                entry_index: num_data_entries + DEFAULT_TICKS_PER_SLOT,
971                starting_txn_index: num_data_entries * tx_per_entry,
972                entry_hash: Hash::new_unique(),
973                executed_txn_count: 0, // Tick entry has no transactions
974            });
975        }
976        entries
977    }
978
979    #[test]
980    pub fn it_should_handle_all_lifecycle_transition_and_produce_frozen_block() {
981        let mut blockstore = super::BlocksStateMachine::default();
982
983        let first_shred_recv = SlotLifecycleUpdate {
984            slot: 1,
985            parent_slot: None,
986            stage: SlotLifecycle::FirstShredReceived,
987        };
988
989        let completed_block = SlotLifecycleUpdate {
990            slot: 1,
991            parent_slot: None,
992            stage: SlotLifecycle::Completed,
993        };
994
995        let slot_status_update = SlotCommitmentStatusUpdate {
996            slot: 1,
997            parent_slot: None,
998            commitment: CommitmentLevel::Processed,
999        };
1000
1001        const NUM_DATA_ENTRIES: u64 = 64;
1002        let entries = generate_entries(1, NUM_DATA_ENTRIES, 10);
1003        let last_entry_hash = entries.last().unwrap().entry_hash;
1004        let summary = BlockSummary {
1005            slot: 1,
1006            parent_slot: 0,
1007            entry_count: NUM_DATA_ENTRIES + DEFAULT_TICKS_PER_SLOT,
1008            executed_transaction_count: NUM_DATA_ENTRIES * 10,
1009            blockhash: last_entry_hash,
1010        };
1011
1012        // Whatever the order of insertion it should to notify the sealed block before slot status
1013        blockstore
1014            .process_replay_event(first_shred_recv.into())
1015            .unwrap();
1016        blockstore
1017            .process_replay_event(completed_block.into())
1018            .unwrap();
1019        for e in entries {
1020            blockstore.process_replay_event(e.into()).unwrap();
1021        }
1022        blockstore.process_replay_event(summary.into()).unwrap();
1023        blockstore.process_consensus_event(slot_status_update.into());
1024
1025        let actual = blockstore.pop_next_unprocess_blockstore_update();
1026        assert!(matches!(
1027            actual,
1028            Some(super::BlockStateMachineOutput::FrozenBlock(_))
1029        ));
1030        let actual = blockstore.pop_next_unprocess_blockstore_update();
1031        assert!(matches!(
1032            actual,
1033            Some(super::BlockStateMachineOutput::SlotStatus(_))
1034        ));
1035        let actual = blockstore.pop_next_unprocess_blockstore_update();
1036        assert!(actual.is_none());
1037    }
1038
1039    #[test]
1040    pub fn it_should_mark_slot_as_dead_if_not_received_first_shred() {
1041        let mut blockstore = super::BlocksStateMachine::default();
1042
1043        let completed_block = SlotLifecycleUpdate {
1044            slot: 1,
1045            parent_slot: None,
1046            stage: SlotLifecycle::Completed,
1047        };
1048
1049        // Send completed block without first shred received
1050        assert!(
1051            blockstore
1052                .process_replay_event(completed_block.into())
1053                .is_err()
1054        );
1055    }
1056
1057    #[test]
1058    pub fn blockstore_gc_should_work_even_when_empty() {
1059        let mut blockstore = super::BlocksStateMachine::default();
1060        let mut gc_trace = Vec::new();
1061        let actual = blockstore.gc(Some(&mut gc_trace));
1062        assert_eq!(actual.slot_purge_count, 0);
1063        assert_eq!(actual.slot_blocked_count, 0);
1064        assert!(gc_trace.is_empty());
1065    }
1066
1067    #[test]
1068    pub fn blockstore_should_correct_missing_processed_slot_status() {
1069        let mut blockstore = super::BlocksStateMachine::default();
1070        let slot_confirmed = SlotCommitmentStatusUpdate {
1071            parent_slot: None,
1072            slot: 1,
1073            commitment: CommitmentLevel::Confirmed,
1074        };
1075
1076        let first_shred_recv = SlotLifecycleUpdate {
1077            slot: 1,
1078            parent_slot: None,
1079            stage: SlotLifecycle::FirstShredReceived,
1080        };
1081
1082        let completed_block = SlotLifecycleUpdate {
1083            slot: 1,
1084            parent_slot: None,
1085            stage: SlotLifecycle::Completed,
1086        };
1087
1088        const NUM_DATA_ENTRIES: u64 = 64;
1089        let entries = generate_entries(1, 64, 10);
1090        let last_entry_hash = entries.last().unwrap().entry_hash;
1091        let summary = BlockSummary {
1092            slot: 1,
1093            parent_slot: 0,
1094            entry_count: NUM_DATA_ENTRIES + DEFAULT_TICKS_PER_SLOT,
1095            executed_transaction_count: NUM_DATA_ENTRIES * 10,
1096            blockhash: last_entry_hash,
1097        };
1098
1099        // Whatever the order of insertion it should to notify the sealed block before slot status
1100        blockstore
1101            .process_replay_event(first_shred_recv.into())
1102            .unwrap();
1103        blockstore
1104            .process_replay_event(completed_block.into())
1105            .unwrap();
1106        for e in entries {
1107            blockstore.process_replay_event(e.into()).unwrap();
1108        }
1109        blockstore.process_consensus_event(slot_confirmed.into());
1110        blockstore.process_replay_event(summary.into()).unwrap();
1111
1112        let actual = blockstore.pop_next_unprocess_blockstore_update().unwrap();
1113        let BlockStateMachineOutput::FrozenBlock(frozen_block) = actual else {
1114            panic!("Expected frozen block");
1115        };
1116        assert_eq!(frozen_block.slot, 1);
1117
1118        let BlockStateMachineOutput::SlotStatus(status) =
1119            blockstore.pop_next_unprocess_blockstore_update().unwrap()
1120        else {
1121            panic!("Expected slot status update");
1122        };
1123
1124        assert_eq!(status.slot, 1);
1125        assert_eq!(status.commitment, CommitmentLevel::Processed);
1126
1127        let actual = blockstore.pop_next_unprocess_blockstore_update().unwrap();
1128        let BlockStateMachineOutput::SlotStatus(status) = actual else {
1129            panic!("Expected slot status update");
1130        };
1131        assert_eq!(status.slot, 1);
1132        assert_eq!(status.commitment, CommitmentLevel::Confirmed);
1133
1134        let actual = blockstore.pop_next_unprocess_blockstore_update();
1135        assert!(actual.is_none());
1136    }
1137
1138    #[test]
1139    pub fn it_should_detect_retroactively_rooted_slots() {
1140        // Retroactively rooted slots are slots that were not rooted at the time of the slot status update,
1141        // but were later rooted by a descendant slot status update.
1142        let mut blockstore = super::BlocksStateMachine::default();
1143
1144        let slot1_processed = SlotCommitmentStatusUpdate {
1145            parent_slot: None,
1146            slot: 1,
1147            commitment: CommitmentLevel::Processed,
1148        };
1149
1150        let slot2_finalized = SlotCommitmentStatusUpdate {
1151            parent_slot: Some(1),
1152            slot: 2,
1153            commitment: CommitmentLevel::Finalized,
1154        };
1155
1156        let slot1_first_shred_recv = SlotLifecycleUpdate {
1157            slot: 1,
1158            parent_slot: None,
1159            stage: SlotLifecycle::FirstShredReceived,
1160        };
1161
1162        let slot2_first_shred_recv = SlotLifecycleUpdate {
1163            slot: 2,
1164            parent_slot: Some(1),
1165            stage: SlotLifecycle::FirstShredReceived,
1166        };
1167
1168        let slot1_completed_block = SlotLifecycleUpdate {
1169            slot: 1,
1170            parent_slot: None,
1171            stage: SlotLifecycle::Completed,
1172        };
1173
1174        let slot2_completed_block = SlotLifecycleUpdate {
1175            slot: 2,
1176            parent_slot: Some(1),
1177            stage: SlotLifecycle::Completed,
1178        };
1179
1180        const NUM_DATA_ENTRIES: u64 = 64;
1181        let slot1_entries = generate_entries(1, 64, 10);
1182        let slot2_entries = generate_entries(2, 64, 10);
1183
1184        let last_entry_hash1 = slot1_entries.last().unwrap().entry_hash;
1185        let last_entry_hash2 = slot2_entries.last().unwrap().entry_hash;
1186
1187        let slot1_summary = BlockSummary {
1188            slot: 1,
1189            parent_slot: 0,
1190            entry_count: NUM_DATA_ENTRIES + DEFAULT_TICKS_PER_SLOT,
1191            executed_transaction_count: NUM_DATA_ENTRIES * 10,
1192            blockhash: last_entry_hash1,
1193        };
1194
1195        let slot2_summary = BlockSummary {
1196            slot: 2,
1197            parent_slot: 1,
1198            entry_count: NUM_DATA_ENTRIES + DEFAULT_TICKS_PER_SLOT,
1199            executed_transaction_count: NUM_DATA_ENTRIES * 10,
1200            blockhash: last_entry_hash2,
1201        };
1202
1203        // Whatever the order of insertion it should to notify the sealed block before slot status
1204        blockstore
1205            .process_replay_event(slot1_first_shred_recv.into())
1206            .unwrap();
1207        blockstore
1208            .process_replay_event(slot1_completed_block.into())
1209            .unwrap();
1210        for e in slot1_entries {
1211            blockstore.process_replay_event(e.into()).unwrap();
1212        }
1213        blockstore
1214            .process_replay_event(slot1_summary.into())
1215            .unwrap();
1216        // We only insert the slot status update for Confirmed, missing Processed
1217        blockstore.process_consensus_event(slot1_processed.into());
1218
1219        let actual = blockstore.pop_next_unprocess_blockstore_update().unwrap();
1220        let BlockStateMachineOutput::FrozenBlock(frozen_block) = actual else {
1221            panic!("Expected frozen block");
1222        };
1223        assert_eq!(frozen_block.slot, 1);
1224
1225        let BlockStateMachineOutput::SlotStatus(status) =
1226            blockstore.pop_next_unprocess_blockstore_update().unwrap()
1227        else {
1228            panic!("Expected slot status update");
1229        };
1230
1231        assert_eq!(status.slot, 1);
1232        assert_eq!(status.commitment, CommitmentLevel::Processed);
1233
1234        // Now we insert the second slot, which will retroactively root the first slot
1235        blockstore
1236            .process_replay_event(slot2_first_shred_recv.into())
1237            .unwrap();
1238        blockstore
1239            .process_replay_event(slot2_completed_block.into())
1240            .unwrap();
1241        for e in slot2_entries {
1242            blockstore.process_replay_event(e.into()).unwrap();
1243        }
1244        blockstore
1245            .process_replay_event(slot2_summary.into())
1246            .unwrap();
1247        blockstore.process_consensus_event(slot2_finalized.into());
1248
1249        let actual = blockstore.pop_next_unprocess_blockstore_update().unwrap();
1250        let BlockStateMachineOutput::FrozenBlock(frozen_block) = actual else {
1251            panic!("Expected frozen block");
1252        };
1253        assert_eq!(frozen_block.slot, 2);
1254
1255        for expected_cl in iter_to_commitment(&CommitmentLevel::Finalized) {
1256            let actual = blockstore.pop_next_unprocess_blockstore_update().unwrap();
1257            let BlockStateMachineOutput::SlotStatus(status) = actual else {
1258                panic!("Expected slot status update");
1259            };
1260            assert_eq!(status.slot, 2);
1261            assert_eq!(status.commitment, expected_cl);
1262        }
1263        // Now we should have a retroactively rooted slot for slot 1
1264        for expected_cl in [CommitmentLevel::Confirmed, CommitmentLevel::Finalized] {
1265            let actual = blockstore.pop_next_unprocess_blockstore_update().unwrap();
1266            let BlockStateMachineOutput::SlotStatus(status) = actual else {
1267                panic!("Expected slot status update");
1268            };
1269            assert_eq!(status.slot, 1);
1270            assert_eq!(status.commitment, expected_cl);
1271        }
1272    }
1273
1274    #[test]
1275    pub fn it_should_handle_rolledback_slot() {
1276        // During an duplicate unconfirmed slot, we may process a slot, then restart all over again from bank_created.
1277        // the state machine should be able to handle this case and not panic, and correctly process the new slot lifecycle update.
1278        let mut blockstore = super::BlocksStateMachine::default();
1279
1280        let bank_created = SlotLifecycleUpdate {
1281            slot: 1,
1282            parent_slot: None,
1283            stage: SlotLifecycle::CreatedBank,
1284        };
1285
1286        let completed_block = SlotLifecycleUpdate {
1287            slot: 1,
1288            parent_slot: None,
1289            stage: SlotLifecycle::Completed,
1290        };
1291
1292        let slot_status_update = SlotCommitmentStatusUpdate {
1293            slot: 1,
1294            parent_slot: None,
1295            commitment: CommitmentLevel::Processed,
1296        };
1297
1298        const NUM_DATA_ENTRIES: u64 = 64;
1299        let entries = generate_entries(1, NUM_DATA_ENTRIES, 10);
1300        let last_entry_hash = entries.last().unwrap().entry_hash;
1301        let summary = BlockSummary {
1302            slot: 1,
1303            parent_slot: 0,
1304            entry_count: NUM_DATA_ENTRIES + DEFAULT_TICKS_PER_SLOT,
1305            executed_transaction_count: NUM_DATA_ENTRIES * 10,
1306            blockhash: last_entry_hash,
1307        };
1308
1309        // Whatever the order of insertion it should to notify the sealed block before slot status
1310        blockstore
1311            .process_replay_event(bank_created.into())
1312            .unwrap();
1313        blockstore
1314            .process_replay_event(completed_block.into())
1315            .unwrap();
1316        for e in &entries {
1317            blockstore.process_replay_event(e.clone().into()).unwrap();
1318        }
1319        blockstore
1320            .process_replay_event(summary.clone().into())
1321            .unwrap();
1322        blockstore.process_consensus_event(slot_status_update.clone().into());
1323
1324        let actual = blockstore.pop_next_unprocess_blockstore_update();
1325        assert!(matches!(
1326            actual,
1327            Some(super::BlockStateMachineOutput::BankCreated(_))
1328        ));
1329
1330        let actual = blockstore.pop_next_unprocess_blockstore_update();
1331        assert!(matches!(
1332            actual,
1333            Some(super::BlockStateMachineOutput::FrozenBlock(_))
1334        ));
1335        let actual = blockstore.pop_next_unprocess_blockstore_update();
1336        assert!(matches!(
1337            actual,
1338            Some(super::BlockStateMachineOutput::SlotStatus(_))
1339        ));
1340        let actual = blockstore.pop_next_unprocess_blockstore_update();
1341        assert!(actual.is_none());
1342
1343        // Now we receive a new bank created for the same slot, which mean the previous slot lifecycle is rolled back.
1344        blockstore
1345            .process_replay_event(bank_created.into())
1346            .unwrap();
1347        for e in &entries {
1348            blockstore.process_replay_event(e.clone().into()).unwrap();
1349        }
1350        blockstore
1351            .process_replay_event(summary.clone().into())
1352            .unwrap();
1353        blockstore.process_consensus_event(slot_status_update.clone().into());
1354
1355        let actual = blockstore.pop_next_unprocess_blockstore_update();
1356        assert!(matches!(
1357            actual,
1358            Some(super::BlockStateMachineOutput::BankReset(_))
1359        ));
1360
1361        let actual = blockstore.pop_next_unprocess_blockstore_update();
1362        assert!(matches!(
1363            actual,
1364            Some(super::BlockStateMachineOutput::FrozenBlock(_))
1365        ));
1366        let actual = blockstore.pop_next_unprocess_blockstore_update();
1367        assert!(actual.is_some());
1368        assert!(matches!(
1369            actual,
1370            Some(super::BlockStateMachineOutput::SlotStatus(_))
1371        ));
1372        let actual = blockstore.pop_next_unprocess_blockstore_update();
1373        assert!(actual.is_none());
1374    }
1375}
yellowstone_block_machine/state_machine.rs

yellowstone_block_machine/
state_machine.rs
