hyphae_storage/
mutation.rs

1// SPDX-License-Identifier: Apache-2.0
2
3use thiserror::Error;
4
5use crate::log::MAX_OPERATION_BYTES;
6
7const PUT: u8 = 1;
8const DELETE: u8 = 2;
9const PUT_HEADER_LENGTH: usize = 13;
10const DELETE_HEADER_LENGTH: usize = 5;
11
12/// Maximum encoded key length accepted by the embedded KV layer.
13pub const MAX_KEY_BYTES: usize = 1024 * 1024;
14
15/// A deterministic mutation persisted inside one transaction operation frame.
16#[derive(Clone, Debug, Eq, PartialEq)]
17pub enum Mutation {
18    /// Replaces the value stored under a nonempty binary key.
19    Put {
20        /// Binary key.
21        key: Vec<u8>,
22        /// Binary value.
23        value: Vec<u8>,
24    },
25    /// Removes a binary key. Deleting a missing key is idempotent.
26    Delete {
27        /// Binary key.
28        key: Vec<u8>,
29    },
30}
31
32impl Mutation {
33    /// Creates a put mutation.
34    pub fn put(key: impl Into<Vec<u8>>, value: impl Into<Vec<u8>>) -> Self {
35        Self::Put {
36            key: key.into(),
37            value: value.into(),
38        }
39    }
40
41    /// Creates a delete mutation.
42    pub fn delete(key: impl Into<Vec<u8>>) -> Self {
43        Self::Delete { key: key.into() }
44    }
45
46    pub(crate) fn encode(&self) -> Result<Vec<u8>, MutationError> {
47        match self {
48            Self::Put { key, value } => encode_put(key, value),
49            Self::Delete { key } => encode_delete(key),
50        }
51    }
52
53    pub(crate) fn decode(encoded: &[u8]) -> Result<Self, MutationError> {
54        let Some(kind) = encoded.first().copied() else {
55            return Err(MutationError::Malformed);
56        };
57        match kind {
58            PUT => decode_put(encoded),
59            DELETE => decode_delete(encoded),
60            kind => Err(MutationError::UnknownKind { kind }),
61        }
62    }
63}
64
65/// Failure while validating or decoding a persisted mutation.
66#[derive(Clone, Debug, Error, Eq, PartialEq)]
67pub enum MutationError {
68    /// Empty keys are forbidden.
69    #[error("mutation key must not be empty")]
70    EmptyKey,
71
72    /// A key exceeds the stable storage limit.
73    #[error("mutation key is {length} bytes; maximum is {maximum}")]
74    KeyTooLarge {
75        /// Actual key length.
76        length: usize,
77        /// Maximum key length.
78        maximum: usize,
79    },
80
81    /// The full encoded operation exceeds one log frame.
82    #[error("encoded mutation is {length} bytes; maximum is {maximum}")]
83    OperationTooLarge {
84        /// Actual encoded length.
85        length: usize,
86        /// Maximum operation length.
87        maximum: usize,
88    },
89
90    /// Persisted bytes do not form one canonical mutation.
91    #[error("malformed persisted mutation")]
92    Malformed,
93
94    /// The mutation kind is not defined by this disk format.
95    #[error("unknown persisted mutation kind {kind}")]
96    UnknownKind {
97        /// Raw kind byte.
98        kind: u8,
99    },
100}
101
102fn encode_put(key: &[u8], value: &[u8]) -> Result<Vec<u8>, MutationError> {
103    validate_key(key)?;
104    let key_length = u32::try_from(key.len()).map_err(|_| MutationError::KeyTooLarge {
105        length: key.len(),
106        maximum: MAX_KEY_BYTES,
107    })?;
108    let value_length =
109        u64::try_from(value.len()).map_err(|_| MutationError::OperationTooLarge {
110            length: usize::MAX,
111            maximum: MAX_OPERATION_BYTES,
112        })?;
113    let encoded_length = PUT_HEADER_LENGTH
114        .checked_add(key.len())
115        .and_then(|length| length.checked_add(value.len()))
116        .ok_or(MutationError::OperationTooLarge {
117            length: usize::MAX,
118            maximum: MAX_OPERATION_BYTES,
119        })?;
120    validate_operation_length(encoded_length)?;
121
122    let mut encoded = Vec::with_capacity(encoded_length);
123    encoded.push(PUT);
124    encoded.extend_from_slice(&key_length.to_le_bytes());
125    encoded.extend_from_slice(&value_length.to_le_bytes());
126    encoded.extend_from_slice(key);
127    encoded.extend_from_slice(value);
128    Ok(encoded)
129}
130
131fn encode_delete(key: &[u8]) -> Result<Vec<u8>, MutationError> {
132    validate_key(key)?;
133    let key_length = u32::try_from(key.len()).map_err(|_| MutationError::KeyTooLarge {
134        length: key.len(),
135        maximum: MAX_KEY_BYTES,
136    })?;
137    let encoded_length =
138        DELETE_HEADER_LENGTH
139            .checked_add(key.len())
140            .ok_or(MutationError::OperationTooLarge {
141                length: usize::MAX,
142                maximum: MAX_OPERATION_BYTES,
143            })?;
144    validate_operation_length(encoded_length)?;
145
146    let mut encoded = Vec::with_capacity(encoded_length);
147    encoded.push(DELETE);
148    encoded.extend_from_slice(&key_length.to_le_bytes());
149    encoded.extend_from_slice(key);
150    Ok(encoded)
151}
152
153fn decode_put(encoded: &[u8]) -> Result<Mutation, MutationError> {
154    if encoded.len() < PUT_HEADER_LENGTH {
155        return Err(MutationError::Malformed);
156    }
157    let key_length = usize::try_from(u32::from_le_bytes(copy_array(&encoded[1..5])))
158        .map_err(|_| MutationError::Malformed)?;
159    let value_length = usize::try_from(u64::from_le_bytes(copy_array(&encoded[5..13])))
160        .map_err(|_| MutationError::Malformed)?;
161    let key_end = PUT_HEADER_LENGTH
162        .checked_add(key_length)
163        .ok_or(MutationError::Malformed)?;
164    let value_end = key_end
165        .checked_add(value_length)
166        .ok_or(MutationError::Malformed)?;
167    if value_end != encoded.len() {
168        return Err(MutationError::Malformed);
169    }
170    let key = encoded[PUT_HEADER_LENGTH..key_end].to_vec();
171    validate_key(&key)?;
172    validate_operation_length(encoded.len())?;
173    Ok(Mutation::Put {
174        key,
175        value: encoded[key_end..value_end].to_vec(),
176    })
177}
178
179fn decode_delete(encoded: &[u8]) -> Result<Mutation, MutationError> {
180    if encoded.len() < DELETE_HEADER_LENGTH {
181        return Err(MutationError::Malformed);
182    }
183    let key_length = usize::try_from(u32::from_le_bytes(copy_array(&encoded[1..5])))
184        .map_err(|_| MutationError::Malformed)?;
185    let key_end = DELETE_HEADER_LENGTH
186        .checked_add(key_length)
187        .ok_or(MutationError::Malformed)?;
188    if key_end != encoded.len() {
189        return Err(MutationError::Malformed);
190    }
191    let key = encoded[DELETE_HEADER_LENGTH..key_end].to_vec();
192    validate_key(&key)?;
193    validate_operation_length(encoded.len())?;
194    Ok(Mutation::Delete { key })
195}
196
197pub(crate) fn validate_key(key: &[u8]) -> Result<(), MutationError> {
198    if key.is_empty() {
199        return Err(MutationError::EmptyKey);
200    }
201    if key.len() > MAX_KEY_BYTES {
202        return Err(MutationError::KeyTooLarge {
203            length: key.len(),
204            maximum: MAX_KEY_BYTES,
205        });
206    }
207    Ok(())
208}
209
210fn validate_operation_length(length: usize) -> Result<(), MutationError> {
211    if length > MAX_OPERATION_BYTES {
212        return Err(MutationError::OperationTooLarge {
213            length,
214            maximum: MAX_OPERATION_BYTES,
215        });
216    }
217    Ok(())
218}
219
220fn copy_array<const N: usize>(source: &[u8]) -> [u8; N] {
221    let mut output = [0_u8; N];
222    output.copy_from_slice(source);
223    output
224}
225
226#[cfg(test)]
227mod tests {
228    use std::error::Error;
229
230    use super::{MAX_KEY_BYTES, Mutation, MutationError};
231
232    #[test]
233    fn mutation_codec_round_trips_binary_values() -> Result<(), Box<dyn Error>> {
234        let mutations = [
235            Mutation::put([0, 1, 2], [255, 0, 3]),
236            Mutation::delete([7, 8, 9]),
237        ];
238        for mutation in mutations {
239            assert_eq!(Mutation::decode(&mutation.encode()?)?, mutation);
240        }
241        Ok(())
242    }
243
244    #[test]
245    fn mutation_codec_rejects_noncanonical_lengths() -> Result<(), Box<dyn Error>> {
246        let mut encoded = Mutation::put(b"key", b"value").encode()?;
247        encoded.push(0);
248        assert_eq!(Mutation::decode(&encoded), Err(MutationError::Malformed));
249        Ok(())
250    }
251
252    #[test]
253    fn keys_are_bounded_before_encoding() {
254        let result = Mutation::delete(vec![0; MAX_KEY_BYTES + 1]).encode();
255        assert!(matches!(result, Err(MutationError::KeyTooLarge { .. })));
256    }
257}
hyphae_storage/mutation.rs

hyphae_storage/
mutation.rs
