reifydb_codec/log/
index.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use crc32fast::Hasher;
5use reifydb_value::{byte_size::ByteSize, reifydb_assertions, value::datetime::DateTime};
6
7use crate::log::{LogIndex, LogVersion, Position};
8
9pub const HEADER_BYTES: usize = 40;
10
11pub const ENTRY_BYTES: usize = 12;
12
13pub const MAGIC: u32 = u32::from_le_bytes(*b"RIDX");
14
15pub const DEFAULT_INTERVAL: ByteSize = ByteSize::from_kib(4);
16
17#[derive(Debug, Clone, Copy, PartialEq, Eq)]
18pub struct TimestampRange {
19	pub min: DateTime,
20	pub max: DateTime,
21}
22
23#[derive(Debug, Clone, Copy, PartialEq, Eq)]
24pub struct Header {
25	pub magic: u32,
26	pub base_version: LogVersion,
27	pub base_index: LogIndex,
28	pub timestamps: Option<TimestampRange>,
29}
30
31impl Header {
32	pub fn new(base_version: LogVersion, base_index: LogIndex) -> Self {
33		Self {
34			magic: MAGIC,
35			base_version,
36			base_index,
37			timestamps: None,
38		}
39	}
40
41	pub fn decode(buf: &[u8; HEADER_BYTES]) -> Option<Self> {
42		let magic = u32::from_le_bytes(buf[0..4].try_into().unwrap());
43		if magic != MAGIC {
44			return None;
45		}
46		let found = u32::from_le_bytes(buf[4..8].try_into().unwrap());
47		if found != checksum_bytes(&buf[8..]) {
48			return None;
49		}
50		let min = DateTime::from_order(u64::from_le_bytes(buf[24..32].try_into().unwrap()));
51		let max = DateTime::from_order(u64::from_le_bytes(buf[32..40].try_into().unwrap()));
52		Some(Self {
53			magic,
54			base_version: LogVersion::new(u64::from_le_bytes(buf[8..16].try_into().unwrap())),
55			base_index: LogIndex::new(u64::from_le_bytes(buf[16..24].try_into().unwrap())),
56			timestamps: (min <= max).then_some(TimestampRange {
57				min,
58				max,
59			}),
60		})
61	}
62
63	pub fn encode(&self) -> [u8; HEADER_BYTES] {
64		let (min, max) = match self.timestamps {
65			Some(range) => (range.min, range.max),
66			None => (DateTime::MAX, DateTime::MIN),
67		};
68		let mut out = [0u8; HEADER_BYTES];
69		out[0..4].copy_from_slice(&self.magic.to_le_bytes());
70		out[8..16].copy_from_slice(&self.base_version.as_u64().to_le_bytes());
71		out[16..24].copy_from_slice(&self.base_index.as_u64().to_le_bytes());
72		out[24..32].copy_from_slice(&min.to_order().to_le_bytes());
73		out[32..40].copy_from_slice(&max.to_order().to_le_bytes());
74		let checksum = checksum_bytes(&out[8..]);
75		out[4..8].copy_from_slice(&checksum.to_le_bytes());
76		out
77	}
78}
79
80fn checksum_bytes(bytes: &[u8]) -> u32 {
81	let mut hasher = Hasher::new();
82	hasher.update(bytes);
83	hasher.finalize()
84}
85
86#[derive(Debug, Clone, Copy, PartialEq, Eq)]
87pub struct Entry {
88	pub version: LogVersion,
89	pub index: LogIndex,
90	pub position: Position,
91}
92
93pub fn encode_entry(header: &Header, entry: Entry) -> [u8; ENTRY_BYTES] {
94	reifydb_assertions! {
95		assert!(
96			entry.version >= header.base_version,
97			"version {} is below the base version {} the index was created with, and the delta wraps \
98			 silently into an entry no lookup can follow",
99			entry.version,
100			header.base_version
101		);
102		assert!(
103			entry.version.as_u64() - header.base_version.as_u64() <= u32::MAX as u64,
104			"a delta of {} overflows the four byte delta field (limit={})",
105			entry.version.as_u64() - header.base_version.as_u64(),
106			u32::MAX
107		);
108		assert!(
109			entry.index >= header.base_index,
110			"index {} is below the base index {} the index was created with, and the delta wraps \
111			 silently into an entry no lookup can follow",
112			entry.index,
113			header.base_index
114		);
115		assert!(
116			entry.index.as_u64() - header.base_index.as_u64() <= u32::MAX as u64,
117			"an index delta of {} overflows the four byte delta field (limit={})",
118			entry.index.as_u64() - header.base_index.as_u64(),
119			u32::MAX
120		);
121		assert!(
122			entry.position.as_u64() <= u32::MAX as u64,
123			"a position of {} overflows the four byte position field, so the segment must stay under \
124			 {} bytes",
125			entry.position,
126			u32::MAX
127		);
128	}
129	let mut out = [0u8; ENTRY_BYTES];
130	out[0..4].copy_from_slice(&((entry.version.as_u64() - header.base_version.as_u64()) as u32).to_le_bytes());
131	out[4..8].copy_from_slice(&((entry.index.as_u64() - header.base_index.as_u64()) as u32).to_le_bytes());
132	out[8..12].copy_from_slice(&(entry.position.as_u64() as u32).to_le_bytes());
133	out
134}
135
136pub fn decode_entry(header: &Header, buf: &[u8; ENTRY_BYTES]) -> Option<Entry> {
137	Some(Entry {
138		version: LogVersion::new(
139			header.base_version
140				.as_u64()
141				.checked_add(u32::from_le_bytes(buf[0..4].try_into().unwrap()) as u64)?,
142		),
143		index: LogIndex::new(
144			header.base_index
145				.as_u64()
146				.checked_add(u32::from_le_bytes(buf[4..8].try_into().unwrap()) as u64)?,
147		),
148		position: Position::new(u32::from_le_bytes(buf[8..12].try_into().unwrap()) as u64),
149	})
150}
reifydb_codec/log/index.rs

reifydb_codec/log/
index.rs
