aedb 0.2.12

Embedded Rust storage engine with transactional commits, WAL durability, and snapshot-consistent reads
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
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117

use std::time::{Duration, Instant};

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum GroupCommitPhase {
    Filling,
    Flushing,
    Complete,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum GroupCommitFlushReason {
    MaxGroupSize,
    MaxGroupDelay,
    IngressDrained,
    StructuralBarrier,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) struct GroupCommitPolicy {
    pub max_group_size: usize,
    pub max_group_delay: Duration,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) struct GroupCommitEpochSnapshot {
    pub phase: GroupCommitPhase,
    pub epoch: u64,
    pub pending_commits: usize,
    pub flush_reason: Option<GroupCommitFlushReason>,
}

#[derive(Debug)]
pub(super) struct GroupCommitStateMachine {
    phase: GroupCommitPhase,
    epoch: u64,
    policy: GroupCommitPolicy,
    filling_started: Option<Instant>,
    pending_commits: usize,
    flush_reason: Option<GroupCommitFlushReason>,
}

impl GroupCommitStateMachine {
    pub(super) fn new(policy: GroupCommitPolicy) -> Self {
        Self {
            phase: GroupCommitPhase::Complete,
            epoch: 0,
            policy: sanitize_policy(policy),
            filling_started: None,
            pending_commits: 0,
            flush_reason: None,
        }
    }

    pub(super) fn begin_filling(&mut self, now: Instant, policy: GroupCommitPolicy) {
        self.policy = sanitize_policy(policy);
        self.phase = GroupCommitPhase::Filling;
        self.filling_started = Some(now);
        self.pending_commits = 0;
        self.flush_reason = None;
    }

    pub(super) fn record_pending_commits(&mut self, commits: usize) {
        self.pending_commits = commits;
    }

    pub(super) fn begin_flushing(&mut self, reason: GroupCommitFlushReason) {
        assert!(
            matches!(self.phase, GroupCommitPhase::Filling),
            "group commit phase invariant violated: begin_flushing outside Filling"
        );
        self.phase = GroupCommitPhase::Flushing;
        self.flush_reason = Some(reason);
        self.epoch = self.epoch.saturating_add(1);
    }

    pub(super) fn complete_flush(&mut self) {
        assert!(
            matches!(self.phase, GroupCommitPhase::Flushing),
            "group commit phase invariant violated: complete_flush outside Flushing"
        );
        self.phase = GroupCommitPhase::Complete;
        self.pending_commits = 0;
        self.filling_started = None;
    }

    pub(super) fn reached_size_limit(&self) -> bool {
        self.pending_commits >= self.policy.max_group_size
    }

    pub(super) fn reached_delay_limit(&self, now: Instant) -> bool {
        self.filling_started
            .is_some_and(|started| now.duration_since(started) >= self.policy.max_group_delay)
    }

    pub(super) fn snapshot(&self) -> GroupCommitEpochSnapshot {
        GroupCommitEpochSnapshot {
            phase: self.phase,
            epoch: self.epoch,
            pending_commits: self.pending_commits,
            flush_reason: self.flush_reason,
        }
    }
}

fn sanitize_policy(policy: GroupCommitPolicy) -> GroupCommitPolicy {
    GroupCommitPolicy {
        max_group_size: policy.max_group_size.max(1),
        max_group_delay: if policy.max_group_delay.is_zero() {
            Duration::from_micros(1)
        } else {
            policy.max_group_delay
        },
    }
}

#[cfg(test)]
mod tests;