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
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153

use std::fmt::Debug;
use std::marker::PhantomData;
use std::ops::RangeBounds;

use crate::defensive::check_range_matches_entries;
use crate::AppData;
use crate::AppDataResponse;
use crate::EffectiveMembership;
use crate::Entry;
use crate::EntryPayload;
use crate::InitialState;
use crate::LogId;
use crate::LogIdOptionExt;
use crate::RaftStorage;
use crate::StorageError;

/// StorageHelper provides additional methods to access a RaftStorage implementation.
pub struct StorageHelper<'a, D, R, S, Sto>
where
    D: AppData,
    R: AppDataResponse,

    S: RaftStorage<D, R>,
    Sto: AsRef<S>,
{
    pub(crate) sto: &'a Sto,
    _p: PhantomData<(D, R, S)>,
}

impl<'a, D, R, S, Sto> StorageHelper<'a, D, R, S, Sto>
where
    D: AppData,
    R: AppDataResponse,
    S: RaftStorage<D, R>,
    Sto: AsRef<S>,
{
    pub fn new(sto: &'a Sto) -> Self {
        Self {
            sto,
            _p: Default::default(),
        }
    }

    /// Get Raft's state information from storage.
    ///
    /// When the Raft node is first started, it will call this interface to fetch the last known state from stable
    /// storage.
    pub async fn get_initial_state(&self) -> Result<InitialState, StorageError> {
        let hs = self.sto.as_ref().read_hard_state().await?;
        let st = self.sto.as_ref().get_log_state().await?;
        let mut last_log_id = st.last_log_id;
        let (last_applied, _) = self.sto.as_ref().last_applied_state().await?;
        let membership = self.get_membership().await?;

        // Clean up dirty state: snapshot is installed but logs are not cleaned.
        if last_log_id < last_applied {
            self.sto.as_ref().purge_logs_upto(last_applied.unwrap()).await?;
            last_log_id = last_applied;
        }

        Ok(InitialState {
            last_log_id,
            last_applied,
            hard_state: hs.unwrap_or_default(),
            last_membership: membership,
        })
    }

    /// Returns the last membership config found in log or state machine.
    pub async fn get_membership(&self) -> Result<Option<EffectiveMembership>, StorageError> {
        let (_, sm_mem) = self.sto.as_ref().last_applied_state().await?;

        let sm_mem_index = match &sm_mem {
            None => 0,
            Some(mem) => mem.log_id.index,
        };

        let log_mem = self.last_membership_in_log(sm_mem_index + 1).await?;

        if log_mem.is_some() {
            return Ok(log_mem);
        }

        Ok(sm_mem)
    }

    /// Get the latest membership config found in the log.
    ///
    /// This method should returns membership with the greatest log index which is `>=since_index`.
    /// If no such membership log is found, it returns `None`, e.g., when logs are cleaned after being applied.
    #[tracing::instrument(level = "trace", skip(self))]
    pub async fn last_membership_in_log(&self, since_index: u64) -> Result<Option<EffectiveMembership>, StorageError> {
        let st = self.sto.as_ref().get_log_state().await?;

        let mut end = st.last_log_id.next_index();
        let start = std::cmp::max(st.last_purged_log_id.next_index(), since_index);
        let step = 64;

        while start < end {
            let step_start = std::cmp::max(start, end.saturating_sub(step));
            let entries = self.sto.as_ref().try_get_log_entries(step_start..end).await?;

            for ent in entries.iter().rev() {
                if let EntryPayload::Membership(ref mem) = ent.payload {
                    return Ok(Some(EffectiveMembership {
                        log_id: ent.log_id,
                        membership: mem.clone(),
                    }));
                }
            }

            end = end.saturating_sub(step);
        }

        Ok(None)
    }

    /// Get a series of log entries from storage.
    ///
    /// Similar to `try_get_log_entries` except an error will be returned if there is an entry not found in the
    /// specified range.
    pub async fn get_log_entries<RB: RangeBounds<u64> + Clone + Debug + Send + Sync>(
        &self,
        range: RB,
    ) -> Result<Vec<Entry<D>>, StorageError> {
        let res = self.sto.as_ref().try_get_log_entries(range.clone()).await?;

        check_range_matches_entries(range, &res)?;

        Ok(res)
    }

    /// Try to get an log entry.
    ///
    /// It does not return an error if the log entry at `log_index` is not found.
    pub async fn try_get_log_entry(&self, log_index: u64) -> Result<Option<Entry<D>>, StorageError> {
        let mut res = self.sto.as_ref().try_get_log_entries(log_index..(log_index + 1)).await?;
        Ok(res.pop())
    }

    /// Get the log id of the entry at `index`.
    pub async fn get_log_id(&self, log_index: u64) -> Result<LogId, StorageError> {
        let st = self.sto.as_ref().get_log_state().await?;

        if Some(log_index) == st.last_purged_log_id.index() {
            return Ok(st.last_purged_log_id.unwrap());
        }

        let entries = self.get_log_entries(log_index..=log_index).await?;

        Ok(entries[0].log_id)
    }
}