nodedb_types/
write_version.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! The version a committed write gives the rows it writes.
4//!
5//! A write that applies a data-group Raft entry takes the entry's log
6//! position. Every replica applies the entry at the same position, so a
7//! version read on one replica compares with the versions every other
8//! replica holds.
9
10use serde::{Deserialize, Serialize};
11
12/// The position of the write that produced a row version, within one vShard.
13///
14/// A write that applies a data-group entry records `epoch`, the epoch at
15/// which its vShard moved to the entry's group, and `index`, the entry's log
16/// index. `local` is `0`.
17///
18/// A write that applies no entry records its WAL LSN in `local`, on top of
19/// the latest version its vShard holds on the core. Such a write exists on
20/// one node only: a single-node write, or a local-only write on a cluster
21/// node. It sorts after every version its vShard already holds and before
22/// the next entry the vShard applies.
23///
24/// Versions compare only within one vShard. The writes of one vShard are
25/// totally ordered by `(epoch, index, local)`.
26#[derive(
27    Debug,
28    Clone,
29    Copy,
30    Default,
31    PartialEq,
32    Eq,
33    PartialOrd,
34    Ord,
35    Hash,
36    Serialize,
37    Deserialize,
38    rkyv::Archive,
39    rkyv::Serialize,
40    rkyv::Deserialize,
41    zerompk::ToMessagePack,
42    zerompk::FromMessagePack,
43)]
44pub struct WriteVersion {
45    /// The epoch at which the vShard moved to the group that applied the
46    /// write. `0` while the vShard stays in its initial group.
47    pub epoch: u64,
48    /// The log index of the data-group entry that applied the write.
49    pub index: u64,
50    /// The WAL LSN of a write that applies no entry, `0` otherwise.
51    pub local: u64,
52}
53
54impl WriteVersion {
55    /// The version below every write.
56    pub const ZERO: Self = Self {
57        epoch: 0,
58        index: 0,
59        local: 0,
60    };
61
62    /// The version of a write that applies the entry at `index`, in `epoch`.
63    pub const fn logged(epoch: u64, index: u64) -> Self {
64        Self {
65            epoch,
66            index,
67            local: 0,
68        }
69    }
70
71    /// The version of a write that applies no entry, at WAL LSN `lsn`, on a
72    /// vShard whose latest version is `latest`.
73    pub const fn local_after(latest: Self, lsn: u64) -> Self {
74        Self {
75            epoch: latest.epoch,
76            index: latest.index,
77            local: lsn,
78        }
79    }
80
81    /// How far `self` sits above `older` in one epoch: the log entries
82    /// between them, or the WAL LSNs between two writes of one entry
83    /// position. `None` across epochs, or when `older` is not below `self`.
84    pub fn distance_above(self, older: Self) -> Option<u64> {
85        if self.epoch != older.epoch || self <= older {
86            return None;
87        }
88        if self.index > older.index {
89            return Some(self.index - older.index);
90        }
91        Some(self.local.saturating_sub(older.local))
92    }
93}
94
95/// One vShard's write version.
96#[derive(
97    Debug,
98    Clone,
99    Copy,
100    PartialEq,
101    Eq,
102    PartialOrd,
103    Ord,
104    Hash,
105    Serialize,
106    Deserialize,
107    rkyv::Archive,
108    rkyv::Serialize,
109    rkyv::Deserialize,
110    zerompk::ToMessagePack,
111    zerompk::FromMessagePack,
112)]
113pub struct ShardVersion {
114    /// The vShard whose history positions `version`.
115    pub vshard: u32,
116    pub version: WriteVersion,
117}
118
119#[cfg(test)]
120mod tests {
121    use super::*;
122
123    #[test]
124    fn a_later_entry_sorts_after_a_local_write_of_the_earlier_one() {
125        let entry = WriteVersion::logged(0, 10);
126        let local = WriteVersion::local_after(entry, 900);
127        let next = WriteVersion::logged(0, 11);
128        assert!(entry < local);
129        assert!(local < next);
130    }
131
132    #[test]
133    fn a_later_epoch_sorts_after_every_index_of_an_earlier_one() {
134        assert!(WriteVersion::logged(0, 1_000) < WriteVersion::logged(5, 1));
135    }
136
137    #[test]
138    fn distance_counts_entries_then_wal_lsns_within_one_epoch() {
139        let base = WriteVersion::logged(2, 10);
140        assert_eq!(WriteVersion::logged(2, 15).distance_above(base), Some(5));
141        let local = WriteVersion::local_after(base, 40);
142        assert_eq!(
143            WriteVersion::local_after(base, 90).distance_above(local),
144            Some(50)
145        );
146        assert_eq!(WriteVersion::logged(3, 1).distance_above(base), None);
147        assert_eq!(base.distance_above(WriteVersion::logged(2, 12)), None);
148    }
149
150    #[test]
151    fn versions_round_trip_through_msgpack() {
152        let version = ShardVersion {
153            vshard: 7,
154            version: WriteVersion {
155                epoch: 3,
156                index: 41,
157                local: 9,
158            },
159        };
160        let bytes = zerompk::to_msgpack_vec(&version).expect("encode");
161        let decoded: ShardVersion = zerompk::from_msgpack(&bytes).expect("decode");
162        assert_eq!(decoded, version);
163    }
164}
nodedb_types/write_version.rs

nodedb_types/
write_version.rs
