prolly/prolly/
tombstone.rs

1//! Logical deletion value helpers.
2//!
3//! A tombstone is an ordinary value envelope that represents a delete without
4//! physically removing the key immediately. This is useful for local-first sync,
5//! peer replication, audit trails, and deferred compaction.
6
7use std::collections::BTreeMap;
8
9use serde::{Deserialize, Serialize};
10
11use super::error::{Error, Mutation};
12
13const TOMBSTONE_MAGIC: &[u8; 4] = b"PLTD";
14const TOMBSTONE_WIRE_VERSION: u8 = 1;
15const HEADER_LEN: usize = 4 + 1 + 8 + 4 + 4;
16const TIMESTAMP_OFFSET: usize = 5;
17const ACTOR_LEN_OFFSET: usize = 13;
18const METADATA_COUNT_OFFSET: usize = 17;
19const METADATA_ENTRY_HEADER_LEN: usize = 4 + 8;
20
21/// Logical deletion value with causal metadata.
22///
23/// The actor identifies who or what performed the delete. The timestamp is a
24/// caller-provided Unix millisecond value. Causal metadata is sorted by key when
25/// encoded, making the byte representation deterministic.
26#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
27pub struct Tombstone {
28    /// Actor that created the tombstone.
29    pub actor: Vec<u8>,
30    /// Unix timestamp in milliseconds supplied by the caller.
31    pub timestamp_millis: u64,
32    /// Application-defined causal metadata.
33    pub causal_metadata: BTreeMap<String, Vec<u8>>,
34}
35
36impl Tombstone {
37    /// Create a tombstone with no causal metadata.
38    pub fn new(actor: impl Into<Vec<u8>>, timestamp_millis: u64) -> Self {
39        Self {
40            actor: actor.into(),
41            timestamp_millis,
42            causal_metadata: BTreeMap::new(),
43        }
44    }
45
46    /// Add causal metadata and return the tombstone for chaining.
47    pub fn with_causal_metadata(
48        mut self,
49        key: impl Into<String>,
50        value: impl Into<Vec<u8>>,
51    ) -> Self {
52        self.insert_causal_metadata(key, value);
53        self
54    }
55
56    /// Insert or replace one causal metadata entry.
57    pub fn insert_causal_metadata(
58        &mut self,
59        key: impl Into<String>,
60        value: impl Into<Vec<u8>>,
61    ) -> &mut Self {
62        self.causal_metadata.insert(key.into(), value.into());
63        self
64    }
65
66    /// Borrow one causal metadata value.
67    pub fn causal_metadata(&self, key: &str) -> Option<&[u8]> {
68        self.causal_metadata.get(key).map(Vec::as_slice)
69    }
70
71    /// Encode this tombstone as deterministic bytes suitable for a leaf value.
72    pub fn to_bytes(&self) -> Result<Vec<u8>, Error> {
73        let actor_len = checked_u32_len(self.actor.len(), "actor")?;
74        let metadata_count = checked_u32_len(self.causal_metadata.len(), "causal metadata count")?;
75
76        let mut out = Vec::with_capacity(
77            HEADER_LEN
78                .saturating_add(self.actor.len())
79                .saturating_add(metadata_encoded_len(&self.causal_metadata)),
80        );
81        out.extend_from_slice(TOMBSTONE_MAGIC);
82        out.push(TOMBSTONE_WIRE_VERSION);
83        out.extend_from_slice(&self.timestamp_millis.to_be_bytes());
84        out.extend_from_slice(&actor_len.to_be_bytes());
85        out.extend_from_slice(&metadata_count.to_be_bytes());
86        out.extend_from_slice(&self.actor);
87
88        for (key, value) in &self.causal_metadata {
89            let key_bytes = key.as_bytes();
90            let key_len = checked_u32_len(key_bytes.len(), "causal metadata key")?;
91            let value_len = checked_u64_len(value.len(), "causal metadata value")?;
92            out.extend_from_slice(&key_len.to_be_bytes());
93            out.extend_from_slice(&value_len.to_be_bytes());
94            out.extend_from_slice(key_bytes);
95            out.extend_from_slice(value);
96        }
97
98        Ok(out)
99    }
100
101    /// Decode a tombstone envelope.
102    pub fn from_bytes(bytes: &[u8]) -> Result<Self, Error> {
103        decode_tombstone(bytes)
104    }
105
106    /// Decode stored bytes, returning `Ok(None)` for non-tombstone values.
107    ///
108    /// Values that start with the tombstone magic but contain an invalid
109    /// tombstone envelope return an error.
110    pub fn from_stored_bytes(bytes: &[u8]) -> Result<Option<Self>, Error> {
111        if is_tombstone_value(bytes) {
112            Self::from_bytes(bytes).map(Some)
113        } else {
114            Ok(None)
115        }
116    }
117
118    /// Return true if bytes start with the tombstone magic prefix.
119    pub fn is_encoded(bytes: &[u8]) -> bool {
120        is_tombstone_value(bytes)
121    }
122
123    /// Build an upsert mutation that stores this tombstone at `key`.
124    pub fn to_upsert_mutation(&self, key: impl Into<Vec<u8>>) -> Result<Mutation, Error> {
125        tombstone_upsert(key, self)
126    }
127}
128
129/// Return true if bytes start with the tombstone magic prefix.
130pub fn is_tombstone_value(bytes: &[u8]) -> bool {
131    bytes.starts_with(TOMBSTONE_MAGIC)
132}
133
134/// Build an upsert mutation that stores `tombstone` at `key`.
135pub fn tombstone_upsert(key: impl Into<Vec<u8>>, tombstone: &Tombstone) -> Result<Mutation, Error> {
136    Ok(Mutation::Upsert {
137        key: key.into(),
138        val: tombstone.to_bytes()?,
139    })
140}
141
142/// Return a physical delete mutation when `stored_value` is a tombstone.
143///
144/// This is the small compaction primitive: scan a retained range, detect
145/// tombstone values, then batch the returned delete mutations once the
146/// application knows the tombstone no longer needs to be replicated.
147pub fn tombstone_compaction(
148    key: impl Into<Vec<u8>>,
149    stored_value: &[u8],
150) -> Result<Option<Mutation>, Error> {
151    match Tombstone::from_stored_bytes(stored_value)? {
152        Some(_) => Ok(Some(Mutation::Delete { key: key.into() })),
153        None => Ok(None),
154    }
155}
156
157fn decode_tombstone(bytes: &[u8]) -> Result<Tombstone, Error> {
158    if bytes.len() < HEADER_LEN {
159        return Err(invalid_tombstone("tombstone envelope is too short"));
160    }
161    if !bytes.starts_with(TOMBSTONE_MAGIC) {
162        return Err(invalid_tombstone("tombstone missing PLTD magic"));
163    }
164
165    let wire_version = bytes[4];
166    if wire_version != TOMBSTONE_WIRE_VERSION {
167        return Err(invalid_tombstone(format!(
168            "unsupported tombstone wire version {wire_version}"
169        )));
170    }
171
172    let timestamp_millis = read_u64(bytes, TIMESTAMP_OFFSET)?;
173    let actor_len = read_u32(bytes, ACTOR_LEN_OFFSET)? as usize;
174    let metadata_count = read_u32(bytes, METADATA_COUNT_OFFSET)? as usize;
175    let actor_end = checked_add(HEADER_LEN, actor_len, "actor")?;
176    if actor_end > bytes.len() {
177        return Err(invalid_tombstone("actor length exceeds envelope length"));
178    }
179
180    let actor = bytes[HEADER_LEN..actor_end].to_vec();
181    let mut offset = actor_end;
182    let mut causal_metadata = BTreeMap::new();
183
184    for _ in 0..metadata_count {
185        let entry_header_end = checked_add(offset, METADATA_ENTRY_HEADER_LEN, "metadata header")?;
186        if entry_header_end > bytes.len() {
187            return Err(invalid_tombstone("metadata entry header is truncated"));
188        }
189
190        let key_len = read_u32(bytes, offset)? as usize;
191        let value_len = usize::try_from(read_u64(bytes, offset + 4)?)
192            .map_err(|_| invalid_tombstone("metadata value length does not fit in usize"))?;
193        offset = entry_header_end;
194
195        let key_end = checked_add(offset, key_len, "metadata key")?;
196        let value_end = checked_add(key_end, value_len, "metadata value")?;
197        if value_end > bytes.len() {
198            return Err(invalid_tombstone(
199                "metadata key/value length exceeds envelope length",
200            ));
201        }
202
203        let key = decode_utf8(&bytes[offset..key_end], "metadata key")?;
204        let value = bytes[key_end..value_end].to_vec();
205        if causal_metadata.insert(key, value).is_some() {
206            return Err(invalid_tombstone("duplicate metadata key"));
207        }
208        offset = value_end;
209    }
210
211    if offset != bytes.len() {
212        return Err(invalid_tombstone(format!(
213            "tombstone has {} trailing bytes",
214            bytes.len() - offset
215        )));
216    }
217
218    Ok(Tombstone {
219        actor,
220        timestamp_millis,
221        causal_metadata,
222    })
223}
224
225fn metadata_encoded_len(metadata: &BTreeMap<String, Vec<u8>>) -> usize {
226    metadata.iter().fold(0usize, |len, (key, value)| {
227        len.saturating_add(METADATA_ENTRY_HEADER_LEN)
228            .saturating_add(key.len())
229            .saturating_add(value.len())
230    })
231}
232
233fn checked_u32_len(len: usize, field: &str) -> Result<u32, Error> {
234    u32::try_from(len).map_err(|_| invalid_tombstone(format!("{field} is too large")))
235}
236
237fn checked_u64_len(len: usize, field: &str) -> Result<u64, Error> {
238    u64::try_from(len).map_err(|_| invalid_tombstone(format!("{field} is too large")))
239}
240
241fn checked_add(base: usize, len: usize, field: &str) -> Result<usize, Error> {
242    base.checked_add(len)
243        .ok_or_else(|| invalid_tombstone(format!("{field} length overflows usize")))
244}
245
246fn read_u32(bytes: &[u8], offset: usize) -> Result<u32, Error> {
247    let value = bytes
248        .get(offset..offset + 4)
249        .ok_or_else(|| invalid_tombstone("tombstone header is truncated"))?;
250    Ok(u32::from_be_bytes(
251        value.try_into().expect("fixed slice length"),
252    ))
253}
254
255fn read_u64(bytes: &[u8], offset: usize) -> Result<u64, Error> {
256    let value = bytes
257        .get(offset..offset + 8)
258        .ok_or_else(|| invalid_tombstone("tombstone header is truncated"))?;
259    Ok(u64::from_be_bytes(
260        value.try_into().expect("fixed slice length"),
261    ))
262}
263
264fn decode_utf8(bytes: &[u8], field: &str) -> Result<String, Error> {
265    std::str::from_utf8(bytes)
266        .map(str::to_owned)
267        .map_err(|err| invalid_tombstone(format!("{field} is not valid UTF-8: {err}")))
268}
269
270fn invalid_tombstone(message: impl Into<String>) -> Error {
271    Error::Deserialize(format!("invalid tombstone: {}", message.into()))
272}
273
274#[cfg(test)]
275mod tests {
276    use super::*;
277
278    #[test]
279    fn tombstone_round_trips_and_encodes_metadata_deterministically() {
280        let left = Tombstone::new(b"agent-a".to_vec(), 1_700_000)
281            .with_causal_metadata("right-root", b"r1".to_vec())
282            .with_causal_metadata("left-root", b"l1".to_vec());
283        let right = Tombstone::new(b"agent-a".to_vec(), 1_700_000)
284            .with_causal_metadata("left-root", b"l1".to_vec())
285            .with_causal_metadata("right-root", b"r1".to_vec());
286
287        assert_eq!(left.to_bytes().unwrap(), right.to_bytes().unwrap());
288
289        let decoded = Tombstone::from_bytes(&left.to_bytes().unwrap()).unwrap();
290        assert_eq!(decoded, left);
291        assert_eq!(decoded.causal_metadata("left-root"), Some(b"l1".as_slice()));
292    }
293
294    #[test]
295    fn stored_bytes_distinguish_non_tombstones_from_invalid_tombstones() {
296        assert_eq!(Tombstone::from_stored_bytes(b"value").unwrap(), None);
297        assert!(matches!(
298            Tombstone::from_stored_bytes(b"PLTD"),
299            Err(Error::Deserialize(_))
300        ));
301    }
302
303    #[test]
304    fn compaction_returns_delete_only_for_tombstones() {
305        let tombstone = Tombstone::new(b"actor".to_vec(), 42);
306        let mutation = tombstone_compaction(b"k".to_vec(), &tombstone.to_bytes().unwrap())
307            .unwrap()
308            .unwrap();
309        assert_eq!(mutation, Mutation::Delete { key: b"k".to_vec() });
310
311        assert_eq!(tombstone_compaction(b"k".to_vec(), b"value").unwrap(), None);
312    }
313}
prolly/prolly/tombstone.rs

prolly/prolly/
tombstone.rs
