aion/durability/
seq.rs

1//! Per-workflow sequence-head tracking.
2
3use crate::durability::DurabilityError;
4
5/// Tracks the current persisted sequence head for one workflow history.
6///
7/// The head is the store's `expected_seq` value: empty histories start at `0`, and after appending
8/// `N` events the head advances by exactly `N`. Callers should advance this tracker only after the
9/// backing store has accepted the append.
10#[derive(Clone, Debug, Default, PartialEq, Eq)]
11pub struct SequenceHead {
12    head: u64,
13}
14
15impl SequenceHead {
16    /// Creates a tracker for a fresh workflow history at head `0`.
17    #[must_use]
18    pub const fn new() -> Self {
19        Self::from_head(0)
20    }
21
22    /// Creates a tracker from an explicitly derived persisted head.
23    #[must_use]
24    pub const fn from_head(head: u64) -> Self {
25        Self { head }
26    }
27
28    /// Returns the current store head to use as `expected_seq`.
29    #[must_use]
30    pub const fn current(&self) -> u64 {
31        self.head
32    }
33
34    /// Returns the sequence number for the first event in the next append batch.
35    #[must_use]
36    pub const fn next_seq(&self) -> Option<u64> {
37        self.head.checked_add(1)
38    }
39
40    /// Marks a successful append and advances the head by the number of appended events.
41    ///
42    /// Advancing by zero is accepted as a no-op because stores may accept empty append batches, but
43    /// the Recorder only records non-empty single-event batches.
44    ///
45    /// # Errors
46    ///
47    /// Returns [`DurabilityError::HistoryShape`] if the event count cannot fit into `u64` or if the
48    /// resulting sequence head would overflow.
49    pub fn mark_append_success(&mut self, event_count: usize) -> Result<(), DurabilityError> {
50        let count = u64::try_from(event_count).map_err(|error| DurabilityError::HistoryShape {
51            reason: format!("append batch length does not fit in u64: {error}"),
52        })?;
53        self.head = self
54            .head
55            .checked_add(count)
56            .ok_or_else(|| DurabilityError::HistoryShape {
57                reason: format!("sequence head overflow advancing {} by {count}", self.head),
58            })?;
59        Ok(())
60    }
61}
62
63#[cfg(test)]
64mod tests {
65    use super::SequenceHead;
66
67    #[test]
68    fn starts_at_zero_by_default() {
69        let sequence = SequenceHead::new();
70
71        assert_eq!(sequence.current(), 0);
72        assert_eq!(sequence.next_seq(), Some(1));
73    }
74
75    #[test]
76    fn starts_from_explicit_head() {
77        let sequence = SequenceHead::from_head(7);
78
79        assert_eq!(sequence.current(), 7);
80        assert_eq!(sequence.next_seq(), Some(8));
81    }
82
83    #[test]
84    fn next_sequence_overflow_is_detectable_by_advance() {
85        let mut sequence = SequenceHead::from_head(u64::MAX);
86
87        assert_eq!(sequence.next_seq(), None);
88        assert!(sequence.mark_append_success(1).is_err());
89        assert_eq!(sequence.current(), u64::MAX);
90    }
91
92    #[test]
93    fn advances_only_after_success_is_marked() -> Result<(), Box<dyn std::error::Error>> {
94        let mut sequence = SequenceHead::from_head(3);
95
96        let simulated_failure: Result<(), ()> = Err(());
97        if simulated_failure.is_ok() {
98            sequence.mark_append_success(2)?;
99        }
100        assert_eq!(sequence.current(), 3);
101
102        sequence.mark_append_success(2)?;
103        assert_eq!(sequence.current(), 5);
104        assert_eq!(sequence.next_seq(), Some(6));
105        Ok(())
106    }
107}
aion/durability/seq.rs

aion/durability/
seq.rs
