mace/utils/
merge_operator.rs

1//! User-provided merge operators (runtime-only, never persisted).
2
3use std::sync::Arc;
4
5/// user-provided merge operator for a bucket
6///
7/// The operator turns a base value plus a chain of merge operands into the logical value:
8///
9/// ```text
10/// apply(base, combine_operands(a, b)) == apply(apply(base, a), b)
11/// ```
12///
13/// `base = None` stands for "key absent or deleted" and belongs to the domain of `apply`.
14/// Implementations must be deterministic and must satisfy associativity, commutativity of
15/// `combine_operands`, and the base action law above. They must not read time, randomness,
16/// external I/O, or mutable process state, and must not produce observable side effects:
17/// compaction and split may retry or redo fold attempts.
18///
19/// Reads combine operands from newest to oldest; compaction combines them from oldest to newest.
20/// Commutativity and associativity make the two orders equivalent. The methods are total and
21/// must return non-empty results; an empty or oversized result is an engine-detected violation.
22/// `None` from `apply` is the only logical deletion. The engine does not invoke the operator during
23/// merge admission or catch panics.
24///
25/// The engine stores only this runtime handle and never persists operator identity. Callers own
26/// semantic compatibility across process restarts.
27pub trait MergeOperator: Send + Sync {
28    /// Combines two operands into one equivalent operand. Must return a non-empty
29    /// equivalent operand for every pair the operator accepted.
30    fn combine_operands(&self, key: &[u8], left: &[u8], right: &[u8]) -> Vec<u8>;
31
32    /// Applies one operand to a base value. `None` deletes or leaves the key absent;
33    /// `Some(value)` must be non-empty.
34    fn apply(&self, key: &[u8], base: Option<&[u8]>, operand: &[u8]) -> Option<Vec<u8>>;
35}
36
37/// Default 64-bit unsigned integer additive merge operator
38///
39/// Operands and base values are little-endian `u64` deltas: `apply` adds the operand onto the base
40/// (zero when the key is absent or deleted), and `combine_operands` adds the two operands. A payload
41/// that is not exactly 8 bytes decodes as zero. ("corruption will be treated as 0"); the operator
42/// stays total for every input and never returns an empty value.
43#[derive(Default, Clone, Copy, Debug)]
44pub struct U64AddOperator;
45
46impl U64AddOperator {
47    fn decode(raw: &[u8]) -> u64 {
48        if raw.len() == size_of::<u64>() {
49            u64::from_le_bytes(raw.try_into().expect("exact u64 slice"))
50        } else {
51            // reference behavior: a malformed payload contributes zero
52            log::error!(
53                "uint64 value corruption, size: {} > {}",
54                raw.len(),
55                size_of::<u64>()
56            );
57            0
58        }
59    }
60
61    fn add(left: u64, right: u64) -> Vec<u8> {
62        left.wrapping_add(right).to_le_bytes().to_vec()
63    }
64}
65
66impl MergeOperator for U64AddOperator {
67    fn combine_operands(&self, _key: &[u8], left: &[u8], right: &[u8]) -> Vec<u8> {
68        Self::add(Self::decode(left), Self::decode(right))
69    }
70
71    fn apply(&self, _key: &[u8], base: Option<&[u8]>, operand: &[u8]) -> Option<Vec<u8>> {
72        let base = base.map(Self::decode).unwrap_or(0);
73        Some(Self::add(base, Self::decode(operand)))
74    }
75}
76
77/// Convenience handle for bucket options that want the default counter algebra.
78pub fn u64_add_operator() -> Option<Arc<dyn MergeOperator>> {
79    Some(Arc::new(U64AddOperator))
80}
81
82#[cfg(test)]
83mod tests {
84    use super::{MergeOperator, U64AddOperator};
85
86    fn decode(raw: &[u8]) -> u64 {
87        u64::from_le_bytes(raw.try_into().unwrap())
88    }
89
90    #[test]
91    fn u64_add_operator_folds_onto_base_and_absence() {
92        let op = U64AddOperator;
93        // absent base: operand alone
94        let v = op.apply(b"k", None, &5u64.to_le_bytes()).unwrap();
95        assert_eq!(decode(&v), 5);
96        // existing base: base + operand
97        let v = op
98            .apply(b"k", Some(&10u64.to_le_bytes()), &5u64.to_le_bytes())
99            .unwrap();
100        assert_eq!(decode(&v), 15);
101        // combine: left + right
102        let v = op.combine_operands(b"k", &5u64.to_le_bytes(), &7u64.to_le_bytes());
103        assert_eq!(decode(&v), 12);
104        // wrapping arithmetic, mirroring the reference operator
105        let v = op
106            .apply(b"k", Some(&u64::MAX.to_le_bytes()), &2u64.to_le_bytes())
107            .unwrap();
108        assert_eq!(decode(&v), 1);
109    }
110
111    #[test]
112    fn u64_add_operator_treats_malformed_payload_as_zero() {
113        let op = U64AddOperator;
114        // non-8-byte operand contributes zero (reference behavior)
115        let v = op
116            .apply(b"k", Some(&10u64.to_le_bytes()), b"short")
117            .unwrap();
118        assert_eq!(decode(&v), 10);
119        // malformed base decodes as zero
120        let v = op.apply(b"k", Some(b"bad"), &5u64.to_le_bytes()).unwrap();
121        assert_eq!(decode(&v), 5);
122        // outputs are never empty
123        let v = op.apply(b"k", None, &0u64.to_le_bytes()).unwrap();
124        assert_eq!(decode(&v), 0);
125    }
126}
mace/utils/merge_operator.rs

mace/utils/
merge_operator.rs
