use std::fmt;
use std::time::Duration;
use crate::branch::{Timestamp, current_timestamp};
use crate::sync::ballot::{Ballot, Stamp};
use crate::sync::topology::SyncNodeId;
const MAGIC: &[u8; 8] = b"HMTTL001";
const HEADER_LEN: usize = MAGIC.len() + EXPIRY_WIDTH;
const EXPIRY_WIDTH: usize = 8;
const NEVER_EXPIRES: u64 = u64::MAX;
const STAMP_MAGIC: &[u8; 8] = b"HMSTMP01";
const COUNTER_WIDTH: usize = 8;
const SEQ_WIDTH: usize = 8;
const NODE_LEN_WIDTH: usize = 4;
const KIND_WIDTH: usize = 1;
pub type ExpiryTimestamp = Timestamp;
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct TtlEntry {
expires_at: Option<ExpiryTimestamp>,
value: Vec<u8>,
}
impl TtlEntry {
#[must_use]
pub const fn never(value: Vec<u8>) -> Self {
Self {
expires_at: None,
value,
}
}
#[must_use]
pub const fn expiring(value: Vec<u8>, expires_at: ExpiryTimestamp) -> Self {
Self {
expires_at: Some(expires_at),
value,
}
}
pub fn with_ttl(value: Vec<u8>, ttl: Duration) -> Result<Self, TtlError> {
Ok(Self::expiring(value, expires_at_from_ttl(ttl)?))
}
#[must_use]
pub const fn expires_at(&self) -> Option<ExpiryTimestamp> {
self.expires_at
}
#[must_use]
pub fn value(&self) -> &[u8] {
&self.value
}
#[must_use]
pub fn into_value(self) -> Vec<u8> {
self.value
}
#[must_use]
pub fn is_expired_at(&self, now: ExpiryTimestamp) -> bool {
self.expires_at.is_some_and(|expires_at| expires_at <= now)
}
#[must_use]
pub fn encode(&self) -> Vec<u8> {
let mut encoded = Vec::with_capacity(HEADER_LEN.saturating_add(self.value.len()));
encoded.extend_from_slice(MAGIC);
encoded.extend_from_slice(&self.expires_at.unwrap_or(NEVER_EXPIRES).to_be_bytes());
encoded.extend_from_slice(&self.value);
encoded
}
pub fn decode(bytes: &[u8]) -> Result<Option<Self>, TtlDecodeError> {
if !bytes.starts_with(MAGIC) {
return Ok(None);
}
let expires_at = decode_expiry(bytes)?;
let value = bytes
.get(HEADER_LEN..)
.ok_or(TtlDecodeError::TruncatedEnvelope { len: bytes.len() })?
.to_vec();
Ok(Some(Self { expires_at, value }))
}
}
const KIND_VALUE: u8 = 0x00;
const KIND_TOMBSTONE: u8 = 0x01;
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum EntryKind {
Value {
expires_at: Option<ExpiryTimestamp>,
value: Vec<u8>,
},
Tombstone,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StampedEntry {
stamp: Stamp,
kind: EntryKind,
}
impl StampedEntry {
#[must_use]
pub const fn new(stamp: Stamp, expires_at: Option<ExpiryTimestamp>, value: Vec<u8>) -> Self {
Self {
stamp,
kind: EntryKind::Value { expires_at, value },
}
}
#[must_use]
pub const fn tombstone(stamp: Stamp) -> Self {
Self {
stamp,
kind: EntryKind::Tombstone,
}
}
#[must_use]
pub const fn stamp(&self) -> &Stamp {
&self.stamp
}
#[must_use]
pub const fn kind(&self) -> &EntryKind {
&self.kind
}
#[must_use]
pub const fn is_tombstone(&self) -> bool {
matches!(self.kind, EntryKind::Tombstone)
}
#[must_use]
pub const fn expires_at(&self) -> Option<ExpiryTimestamp> {
match self.kind {
EntryKind::Value { expires_at, .. } => expires_at,
EntryKind::Tombstone => None,
}
}
#[must_use]
pub fn value(&self) -> Option<&[u8]> {
match &self.kind {
EntryKind::Value { value, .. } => Some(value),
EntryKind::Tombstone => None,
}
}
#[must_use]
pub fn into_value(self) -> Option<Vec<u8>> {
match self.kind {
EntryKind::Value { value, .. } => Some(value),
EntryKind::Tombstone => None,
}
}
#[must_use]
pub fn is_expired_at(&self, now: ExpiryTimestamp) -> bool {
self.expires_at().is_some_and(|expires_at| expires_at <= now)
}
#[must_use]
pub fn encode(&self) -> Vec<u8> {
let node = self.stamp.epoch.node.as_str().as_bytes();
let value_len = self.value().map_or(0, <[u8]>::len);
let header = STAMP_MAGIC.len()
+ COUNTER_WIDTH
+ NODE_LEN_WIDTH
+ node.len()
+ SEQ_WIDTH
+ KIND_WIDTH
+ EXPIRY_WIDTH;
let mut encoded = Vec::with_capacity(header.saturating_add(value_len));
encoded.extend_from_slice(STAMP_MAGIC);
encoded.extend_from_slice(&self.stamp.epoch.counter.to_be_bytes());
let node_len = u32::try_from(node.len()).unwrap_or(u32::MAX);
encoded.extend_from_slice(&node_len.to_be_bytes());
encoded.extend_from_slice(node);
encoded.extend_from_slice(&self.stamp.seq.to_be_bytes());
match &self.kind {
EntryKind::Value { expires_at, value } => {
encoded.push(KIND_VALUE);
encoded.extend_from_slice(&expires_at.unwrap_or(NEVER_EXPIRES).to_be_bytes());
encoded.extend_from_slice(value);
}
EntryKind::Tombstone => {
encoded.push(KIND_TOMBSTONE);
}
}
encoded
}
pub fn decode(bytes: &[u8]) -> Result<Option<Self>, TtlDecodeError> {
if !bytes.starts_with(STAMP_MAGIC) {
return Ok(None);
}
let mut cursor = STAMP_MAGIC.len();
let counter = read_u64(bytes, &mut cursor)?;
let node_len = read_u32(bytes, &mut cursor)? as usize;
let node_bytes = read_slice(bytes, &mut cursor, node_len)?;
let node = String::from_utf8(node_bytes.to_vec())
.map_err(|_error| TtlDecodeError::TruncatedEnvelope { len: bytes.len() })?;
let seq = read_u64(bytes, &mut cursor)?;
let kind_byte = read_slice(bytes, &mut cursor, KIND_WIDTH)?[0];
let stamp = Stamp::new(Ballot::new(counter, SyncNodeId::new(node)), seq);
let kind = match kind_byte {
KIND_TOMBSTONE => EntryKind::Tombstone,
KIND_VALUE => {
let expiry_raw = read_u64(bytes, &mut cursor)?;
let value = bytes
.get(cursor..)
.ok_or(TtlDecodeError::TruncatedEnvelope { len: bytes.len() })?
.to_vec();
EntryKind::Value {
expires_at: (expiry_raw != NEVER_EXPIRES).then_some(expiry_raw),
value,
}
}
_ => return Err(TtlDecodeError::TruncatedEnvelope { len: bytes.len() }),
};
Ok(Some(Self { stamp, kind }))
}
}
#[must_use]
pub fn encode_stamped(value: Vec<u8>, stamp: Stamp, expires_at: Option<ExpiryTimestamp>) -> Vec<u8> {
StampedEntry::new(stamp, expires_at, value).encode()
}
#[must_use]
pub fn encode_stamped_tombstone(stamp: Stamp) -> Vec<u8> {
StampedEntry::tombstone(stamp).encode()
}
fn read_u64(bytes: &[u8], cursor: &mut usize) -> Result<u64, TtlDecodeError> {
let slice = read_slice(bytes, cursor, 8)?;
let array: [u8; 8] = slice
.try_into()
.map_err(|_error| TtlDecodeError::TruncatedEnvelope { len: bytes.len() })?;
Ok(u64::from_be_bytes(array))
}
fn read_u32(bytes: &[u8], cursor: &mut usize) -> Result<u32, TtlDecodeError> {
let slice = read_slice(bytes, cursor, 4)?;
let array: [u8; 4] = slice
.try_into()
.map_err(|_error| TtlDecodeError::TruncatedEnvelope { len: bytes.len() })?;
Ok(u32::from_be_bytes(array))
}
fn read_slice<'a>(
bytes: &'a [u8],
cursor: &mut usize,
len: usize,
) -> Result<&'a [u8], TtlDecodeError> {
let end = cursor
.checked_add(len)
.ok_or(TtlDecodeError::TruncatedEnvelope { len: bytes.len() })?;
let slice = bytes
.get(*cursor..end)
.ok_or(TtlDecodeError::TruncatedEnvelope { len: bytes.len() })?;
*cursor = end;
Ok(slice)
}
pub fn expires_at_from_ttl(ttl: Duration) -> Result<ExpiryTimestamp, TtlError> {
let ttl_nanos = duration_nanos(ttl)?;
let now = current_timestamp();
now.checked_add(ttl_nanos)
.filter(|expires_at| *expires_at != NEVER_EXPIRES)
.ok_or(TtlError::TimestampOverflow)
}
pub fn encode_optional_ttl(value: Vec<u8>, ttl: Option<Duration>) -> Result<Vec<u8>, TtlError> {
match ttl {
Some(ttl) => TtlEntry::with_ttl(value, ttl).map(|entry| entry.encode()),
None if value.starts_with(MAGIC) || value.starts_with(STAMP_MAGIC) => {
Ok(TtlEntry::never(value).encode())
}
None => Ok(value),
}
}
pub fn encode_stamped_optional_ttl(
value: Vec<u8>,
stamp: Stamp,
ttl: Option<Duration>,
) -> Result<Vec<u8>, TtlError> {
let expires_at = match ttl {
Some(ttl) => Some(expires_at_from_ttl(ttl)?),
None => None,
};
Ok(encode_stamped(value, stamp, expires_at))
}
fn duration_nanos(ttl: Duration) -> Result<u64, TtlError> {
u64::try_from(ttl.as_nanos()).map_err(|_| TtlError::TimestampOverflow)
}
fn decode_expiry(bytes: &[u8]) -> Result<Option<ExpiryTimestamp>, TtlDecodeError> {
let expiry_bytes: [u8; EXPIRY_WIDTH] = bytes
.get(MAGIC.len()..HEADER_LEN)
.and_then(|bytes| bytes.try_into().ok())
.ok_or(TtlDecodeError::TruncatedEnvelope { len: bytes.len() })?;
let raw = u64::from_be_bytes(expiry_bytes);
Ok((raw != NEVER_EXPIRES).then_some(raw))
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum TtlError {
TimestampOverflow,
}
impl fmt::Display for TtlError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::TimestampOverflow => write!(formatter, "ttl expiry timestamp overflow"),
}
}
}
impl std::error::Error for TtlError {}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum TtlDecodeError {
TruncatedEnvelope { len: usize },
}
impl fmt::Display for TtlDecodeError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::TruncatedEnvelope { len } => {
write!(formatter, "truncated ttl envelope with length {len}")
}
}
}
}
impl std::error::Error for TtlDecodeError {}
#[cfg(test)]
mod tests {
use super::{
MAGIC, STAMP_MAGIC, StampedEntry, TtlEntry, encode_optional_ttl, encode_stamped,
encode_stamped_tombstone,
};
use crate::sync::ballot::{Ballot, Stamp};
use crate::sync::topology::SyncNodeId;
use crate::tree::Hash;
use crate::ttl::filter::{Visibility, visible_value_at};
fn stamp(counter: u64, node: &str, seq: u64) -> Stamp {
Stamp::new(Ballot::new(counter, SyncNodeId::new(node)), seq)
}
#[test]
fn raw_values_decode_as_legacy_never_expiring() -> Result<(), Box<dyn std::error::Error>> {
assert_eq!(TtlEntry::decode(b"plain")?, None);
Ok(())
}
#[test]
fn stamped_envelope_round_trips_stamp_ttl_and_value()
-> Result<(), Box<dyn std::error::Error>> {
let entry = StampedEntry::new(
stamp(7, "owner-node-\u{00e9}", 0xdead_beef),
Some(42),
b"payload".to_vec(),
);
let decoded = StampedEntry::decode(&entry.encode())?.ok_or("missing stamped envelope")?;
assert_eq!(decoded.stamp(), &stamp(7, "owner-node-\u{00e9}", 0xdead_beef));
assert_eq!(decoded.expires_at(), Some(42));
assert_eq!(decoded.value(), Some(b"payload".as_slice()));
assert!(!decoded.is_tombstone());
let never = StampedEntry::new(stamp(1, "n", 5), None, b"v".to_vec());
let decoded = StampedEntry::decode(&never.encode())?.ok_or("missing")?;
assert_eq!(decoded.expires_at(), None);
assert_eq!(decoded.stamp(), &stamp(1, "n", 5));
Ok(())
}
#[test]
fn non_stamped_bytes_decode_to_none() -> Result<(), Box<dyn std::error::Error>> {
assert_eq!(StampedEntry::decode(b"plain")?, None);
assert_eq!(StampedEntry::decode(&TtlEntry::expiring(b"v".to_vec(), 9).encode())?, None);
Ok(())
}
#[test]
fn stamped_tombstone_round_trips_and_reads_as_absent()
-> Result<(), Box<dyn std::error::Error>> {
let encoded = encode_stamped_tombstone(stamp(11, "owner", 4));
let decoded = StampedEntry::decode(&encoded)?.ok_or("missing tombstone")?;
assert!(decoded.is_tombstone());
assert_eq!(decoded.stamp(), &stamp(11, "owner", 4));
assert_eq!(decoded.value(), None);
assert_eq!(decoded.expires_at(), None);
assert_eq!(decoded.into_value(), None);
assert_eq!(visible_value_at(&encoded, 0)?, Visibility::Expired);
assert_eq!(visible_value_at(&encoded, u64::MAX)?, Visibility::Expired);
assert_eq!(visible_value_at(&encoded, 0)?.into_option(), None);
Ok(())
}
#[test]
fn empty_value_is_not_a_tombstone() -> Result<(), Box<dyn std::error::Error>> {
let empty_value = encode_stamped(Vec::new(), stamp(3, "n", 0), None);
let tombstone = encode_stamped_tombstone(stamp(3, "n", 0));
assert_ne!(empty_value, tombstone, "empty value must differ from tombstone");
let value = StampedEntry::decode(&empty_value)?.ok_or("missing value")?;
assert!(!value.is_tombstone());
assert_eq!(value.value(), Some(b"".as_slice()));
assert_eq!(visible_value_at(&empty_value, 0)?, Visibility::Live(Vec::new()));
let tomb = StampedEntry::decode(&tombstone)?.ok_or("missing tombstone")?;
assert!(tomb.is_tombstone());
Ok(())
}
#[test]
fn same_value_different_stamps_hash_identically()
-> Result<(), Box<dyn std::error::Error>> {
let value = b"logical-value".to_vec();
let a = encode_stamped(value.clone(), stamp(3, "A", 0), None);
let b = encode_stamped(value.clone(), stamp(9, "B", 17), None);
assert_ne!(a, b, "different stamps must encode to different envelope bytes");
let visible_a = visible_value_at(&a, 0)?;
let visible_b = visible_value_at(&b, 0)?;
assert_eq!(visible_a, Visibility::Live(value.clone()));
assert_eq!(visible_b, Visibility::Live(value.clone()));
let hash_a = Hash::of(&visible_a.into_option().ok_or("expired")?);
let hash_b = Hash::of(&visible_b.into_option().ok_or("expired")?);
assert_eq!(hash_a, hash_b, "the stamp must NOT enter the CAS identity");
assert_eq!(hash_a, Hash::of(&value));
Ok(())
}
#[test]
fn stamped_value_colliding_with_magic_round_trips()
-> Result<(), Box<dyn std::error::Error>> {
let mut value = STAMP_MAGIC.to_vec();
value.extend_from_slice(b"inner");
let encoded = encode_stamped(value.clone(), stamp(2, "z", 1), None);
let decoded = StampedEntry::decode(&encoded)?.ok_or("missing")?;
assert_eq!(decoded.value(), Some(value.as_slice()));
assert_eq!(visible_value_at(&encoded, 0)?, Visibility::Live(value));
Ok(())
}
#[test]
fn envelope_round_trips_expiry_and_value() -> Result<(), Box<dyn std::error::Error>> {
let entry = TtlEntry::expiring(b"payload".to_vec(), 42);
let decoded = TtlEntry::decode(&entry.encode())?.ok_or("missing ttl envelope")?;
assert_eq!(decoded.expires_at(), Some(42));
assert_eq!(decoded.value(), b"payload");
Ok(())
}
#[test]
fn optional_none_stores_raw_value_unenveloped() -> Result<(), Box<dyn std::error::Error>> {
let encoded = encode_optional_ttl(b"value".to_vec(), None)?;
assert_eq!(encoded, b"value");
assert_eq!(TtlEntry::decode(&encoded)?, None);
Ok(())
}
#[test]
fn optional_none_envelopes_a_value_that_collides_with_the_magic()
-> Result<(), Box<dyn std::error::Error>> {
let mut raw = MAGIC.to_vec();
raw.extend_from_slice(b"payload");
let encoded = encode_optional_ttl(raw.clone(), None)?;
let decoded =
TtlEntry::decode(&encoded)?.ok_or("magic-colliding value must be enveloped")?;
assert_eq!(decoded.expires_at(), None);
assert_eq!(decoded.value(), raw.as_slice());
Ok(())
}
}