nexir_mvcc/
codec.rs

1use crate::error::CodecError;
2use crate::types::{CommittedVersion, Intent, Mutation, Timestamp, TxnId};
3
4// Codec version byte. Only 0x02 is defined.
5const CODEC_VERSION: u8 = 0x02;
6const RECORD_COMMITTED: u8 = 0x01;
7const RECORD_INTENT: u8 = 0x02;
8const FLAG_PRESENT: u8 = 0x01;
9const FLAG_ABSENT: u8 = 0x00;
10const MUTATION_PUT: u8 = 0x01;
11const MUTATION_DELETE: u8 = 0x02;
12
13fn encode_u32(v: u32) -> [u8; 4] {
14    v.to_be_bytes()
15}
16
17fn decode_u32(buf: &[u8], offset: &mut usize) -> Result<u32, CodecError> {
18    if buf.len() < *offset + 4 {
19        return Err(CodecError::Decode("truncated u32".into()));
20    }
21    let val = u32::from_be_bytes([
22        buf[*offset],
23        buf[*offset + 1],
24        buf[*offset + 2],
25        buf[*offset + 3],
26    ]);
27    *offset += 4;
28    Ok(val)
29}
30
31fn encode_u64(v: u64) -> [u8; 8] {
32    v.to_be_bytes()
33}
34
35fn decode_u64(buf: &[u8], offset: &mut usize) -> Result<u64, CodecError> {
36    if buf.len() < *offset + 8 {
37        return Err(CodecError::Decode("truncated u64".into()));
38    }
39    let val = u64::from_be_bytes([
40        buf[*offset],
41        buf[*offset + 1],
42        buf[*offset + 2],
43        buf[*offset + 3],
44        buf[*offset + 4],
45        buf[*offset + 5],
46        buf[*offset + 6],
47        buf[*offset + 7],
48    ]);
49    *offset += 8;
50    Ok(val)
51}
52
53fn encode_u128(v: u128) -> [u8; 16] {
54    v.to_be_bytes()
55}
56
57fn decode_u128(buf: &[u8], offset: &mut usize) -> Result<u128, CodecError> {
58    if buf.len() < *offset + 16 {
59        return Err(CodecError::Decode("truncated u128".into()));
60    }
61    let bytes: [u8; 16] = buf[*offset..*offset + 16]
62        .try_into()
63        .map_err(|_| CodecError::Decode("truncated u128".into()))?;
64    *offset += 16;
65    Ok(u128::from_be_bytes(bytes))
66}
67
68fn encode_bytes(buf: &mut Vec<u8>, bytes: &[u8]) -> Result<(), CodecError> {
69    let len =
70        u32::try_from(bytes.len()).map_err(|_| CodecError::Encode("byte slice too long".into()))?;
71    buf.extend_from_slice(&encode_u32(len));
72    buf.extend_from_slice(bytes);
73    Ok(())
74}
75
76fn decode_bytes(buf: &[u8], offset: &mut usize) -> Result<Vec<u8>, CodecError> {
77    let len = decode_u32(buf, offset)? as usize;
78    if buf.len() < *offset + len {
79        return Err(CodecError::Decode("truncated byte slice".into()));
80    }
81    let bytes = buf[*offset..*offset + len].to_vec();
82    *offset += len;
83    Ok(bytes)
84}
85
86/// Encodes a committed version to bytes.
87pub fn encode_committed(version: &CommittedVersion) -> Result<Vec<u8>, CodecError> {
88    let mut buf = Vec::new();
89    buf.push(CODEC_VERSION);
90    buf.push(RECORD_COMMITTED);
91    encode_bytes(&mut buf, &version.key)?;
92    buf.extend_from_slice(&encode_u128(version.commit_ts.0));
93    match &version.value {
94        Some(v) => {
95            buf.push(FLAG_PRESENT);
96            encode_bytes(&mut buf, v)?;
97        }
98        None => {
99            buf.push(FLAG_ABSENT);
100        }
101    }
102    Ok(buf)
103}
104
105/// Decodes a committed version from bytes.
106pub fn decode_committed(buf: &[u8]) -> Result<CommittedVersion, CodecError> {
107    let mut offset = 0;
108    if buf.is_empty() {
109        return Err(CodecError::Decode("empty buffer".into()));
110    }
111    let version = buf[offset];
112    offset += 1;
113    if version != CODEC_VERSION {
114        return Err(CodecError::Decode(format!(
115            "unknown codec version {}",
116            version
117        )));
118    }
119    if buf.len() < offset + 1 {
120        return Err(CodecError::Decode("missing record type".into()));
121    }
122    let record_type = buf[offset];
123    offset += 1;
124    if record_type != RECORD_COMMITTED {
125        return Err(CodecError::Decode(format!(
126            "expected committed record type {}, got {}",
127            RECORD_COMMITTED, record_type
128        )));
129    }
130    let key = decode_bytes(buf, &mut offset)?;
131    let commit_ts = Timestamp(decode_u128(buf, &mut offset)?);
132    if buf.len() < offset + 1 {
133        return Err(CodecError::Decode("missing value flag".into()));
134    }
135    let flag = buf[offset];
136    offset += 1;
137    let value = if flag == FLAG_PRESENT {
138        Some(decode_bytes(buf, &mut offset)?)
139    } else if flag == FLAG_ABSENT {
140        None
141    } else {
142        return Err(CodecError::Decode(format!("unknown value flag {}", flag)));
143    };
144    if offset != buf.len() {
145        return Err(CodecError::Decode("trailing garbage".into()));
146    }
147    Ok(CommittedVersion {
148        key,
149        commit_ts,
150        value,
151    })
152}
153
154/// Encodes an intent to bytes.
155pub fn encode_intent(intent: &Intent) -> Result<Vec<u8>, CodecError> {
156    let mut buf = Vec::new();
157    buf.push(CODEC_VERSION);
158    buf.push(RECORD_INTENT);
159    encode_bytes(&mut buf, &intent.key)?;
160    buf.extend_from_slice(&encode_u64(intent.txn_id.0));
161    buf.extend_from_slice(&encode_u128(intent.start_ts.0));
162    match intent.min_commit_ts {
163        Some(ts) => {
164            buf.push(FLAG_PRESENT);
165            buf.extend_from_slice(&encode_u128(ts.0));
166        }
167        None => {
168            buf.push(FLAG_ABSENT);
169        }
170    }
171    match &intent.mutation {
172        Mutation::Put(v) => {
173            buf.push(MUTATION_PUT);
174            encode_bytes(&mut buf, v)?;
175        }
176        Mutation::Delete => {
177            buf.push(MUTATION_DELETE);
178        }
179    }
180    Ok(buf)
181}
182
183/// Decodes an intent from bytes.
184pub fn decode_intent(buf: &[u8]) -> Result<Intent, CodecError> {
185    let mut offset = 0;
186    if buf.is_empty() {
187        return Err(CodecError::Decode("empty buffer".into()));
188    }
189    let version = buf[offset];
190    offset += 1;
191    if version != CODEC_VERSION {
192        return Err(CodecError::Decode(format!(
193            "unknown codec version {}",
194            version
195        )));
196    }
197    if buf.len() < offset + 1 {
198        return Err(CodecError::Decode("missing record type".into()));
199    }
200    let record_type = buf[offset];
201    offset += 1;
202    if record_type != RECORD_INTENT {
203        return Err(CodecError::Decode(format!(
204            "expected intent record type {}, got {}",
205            RECORD_INTENT, record_type
206        )));
207    }
208    let key = decode_bytes(buf, &mut offset)?;
209    let txn_id = TxnId(decode_u64(buf, &mut offset)?);
210    let start_ts = Timestamp(decode_u128(buf, &mut offset)?);
211    if buf.len() < offset + 1 {
212        return Err(CodecError::Decode("missing min_commit_ts flag".into()));
213    }
214    let flag = buf[offset];
215    offset += 1;
216    let min_commit_ts = if flag == FLAG_PRESENT {
217        Some(Timestamp(decode_u128(buf, &mut offset)?))
218    } else if flag == FLAG_ABSENT {
219        None
220    } else {
221        return Err(CodecError::Decode(format!(
222            "unknown min_commit_ts flag {}",
223            flag
224        )));
225    };
226    if buf.len() < offset + 1 {
227        return Err(CodecError::Decode("missing mutation type".into()));
228    }
229    let mutation_type = buf[offset];
230    offset += 1;
231    let mutation = if mutation_type == MUTATION_PUT {
232        Mutation::Put(decode_bytes(buf, &mut offset)?)
233    } else if mutation_type == MUTATION_DELETE {
234        Mutation::Delete
235    } else {
236        return Err(CodecError::Decode(format!(
237            "unknown mutation type {}",
238            mutation_type
239        )));
240    };
241    if offset != buf.len() {
242        return Err(CodecError::Decode("trailing garbage".into()));
243    }
244    Ok(Intent {
245        key,
246        txn_id,
247        start_ts,
248        mutation,
249        min_commit_ts,
250    })
251}
nexir_mvcc/codec.rs

nexir_mvcc/
codec.rs
