chainfold 0.3.0

event-sourced chain fold engine: total order, fork recovery, durable snapshots
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

#[cfg(not(feature = "std"))]
use alloc::vec::Vec;
#[cfg(feature = "std")]
use std::vec::Vec;

use crate::{
    position::Position,
    ring::BlockRing,
};

/// Fixed K-slot ring of engine restore points.
#[derive(Debug)]
pub(crate) struct CheckpointRing<F> {
    slots: Vec<Option<Slot<F>>>,
    next: usize,
}

#[derive(Debug)]
pub(crate) struct Slot<F> {
    pub(crate) fold: F,
    pub(crate) cursor: Option<Position>,
}

/// True when the engine's ring can still serve the slot's window.
fn live<F>(slot: &Slot<F>, ring: &BlockRing) -> bool {
    slot.cursor.is_none_or(|cursor| ring.observes(cursor.block))
}

impl<F> CheckpointRing<F> {
    pub(crate) fn new(slots: usize) -> Self {
        Self {
            slots: (0..slots).map(|_| None).collect(),
            next: 0,
        }
    }

    /// Live slots oldest first, since the next write position holds the oldest.
    fn oldest_first<'a>(
        &'a self,
        ring: &'a BlockRing,
    ) -> impl DoubleEndedIterator<Item = &'a Slot<F>> {
        let (newest, oldest) = self.slots.split_at(self.next);
        oldest
            .iter()
            .chain(newest)
            .flatten()
            .filter(move |slot| live(slot, ring))
    }

    pub(crate) fn count(&self, ring: &BlockRing) -> usize {
        self.oldest_first(ring).count()
    }

    /// Stores a slot at the next write position; overwrites the oldest when full.
    pub(crate) fn store(&mut self, slot: Slot<F>) {
        let len = self.slots.len();
        if len == 0 {
            return;
        }
        self.slots[self.next] = Some(slot);
        self.next = (self.next + 1) % len;
    }

    /// Oldest live slot; the mirror of best_at_or_below's newest-first scan.
    pub(crate) fn oldest<'a>(&'a self, ring: &'a BlockRing) -> Option<&'a Slot<F>> {
        self.oldest_first(ring).next()
    }

    /// Newest live slot with cursor block at or below the argument; empty-cursor slots
    /// always qualify.
    #[cold]
    pub(crate) fn best_at_or_below<'a>(
        &'a self,
        block: u64,
        ring: &'a BlockRing,
    ) -> Option<&'a Slot<F>> {
        self.oldest_first(ring)
            .rev()
            .find(|slot| slot.cursor.is_none_or(|cursor| cursor.block <= block))
    }

    /// Drops slots with cursor block strictly above the argument.
    #[cold]
    pub(crate) fn drop_above(&mut self, block: u64) {
        for slot in &mut self.slots {
            slot.take_if(|slot| slot.cursor.is_some_and(|cursor| cursor.block > block));
        }
    }

    pub(crate) fn clear(&mut self) {
        self.slots.fill_with(|| None);
    }
}