cubelinux/cubelinux
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Activity

cube-store: a record carries its class mask; the store writes v4 and a v2 log

Browse Source 
The model change first, because the format change is downstream of it: a
stored record is now its bytes *and* the mask its writer stamped. The mask
travels with the value rather than beside it — a record's class is a property
of that record, and a second map would carry an invariant to maintain by
hand, which is a thing that breaks between two updates rather than in one.

`Store::put_flagged` is separate from `put` rather than replacing it, because
classification is something a caller does and most callers do not: a backend
that cannot carry a mask is still a `Store`, and the default says so by
writing the value and dropping the mask rather than by refusing a write.
Every backend that persists now implements it, and the sealing wrapper
forwards it — whether a record is ciphertext says nothing about what kind of
record it is.

Then the format: `serialize_store` emits v4, and the log is v2 with the mask
in the entry. A log framed one way and read another way mis-addresses every
entry after the first, so the reader takes the stride from the header once,
and reads a v1 entry's absent mask as 0 — "no class" — which is what every
record written before the field existed means. The checksum covers the mask,
so a flipped bit in a record's class is a torn entry rather than a silent
reclassification.

Both implementations now speak the current format, which is what makes the
two spellings of one store comparable again.
This commit is contained in:
luulu
2026-09-22 01:02:51 -04:00
parent d1f15bc9bb
commit a54a459dc0
3 changed files with 244 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/cube-image/src/main.rs
+7 -5
View File
@@ -34,7 +34,7 @@ use cube_store::flagged::FlagMode;
use cube_store::serialize_store;
use cube_store_raw::{iter_records, parse, parse_header, MAGIC, RAW_KEY_LEN, SPACE_ID_LEN};
type Map = BTreeMap<(SpaceId, [u8; 24]), Vec<u8>>;
type Map = BTreeMap<(SpaceId, [u8; 24]), cube_store::Record>;
/// Deterministic, human-legible space ids: the label expanded to 32 bytes.
///
@@ -111,7 +111,7 @@ fn curated() -> Vec<u8> {
for (sp, rows) in devices {
for (x, v) in rows {
let p = Point::new(x, 0, 0);
map.insert((sp, Morton::encode(p)), v.as_bytes().to_vec());
map.insert((sp, Morton::encode(p)), cube_store::Record::plain(v.as_bytes().to_vec()));
}
}
@@ -119,18 +119,20 @@ fn curated() -> Vec<u8> {
// Stored as data at a coordinate like anything else.
map.insert(
(portal, Morton::encode(Point::new(1, 0, 0))),
b"portal: from=demo-device-01 to=portal kind=device-link".to_vec(),
cube_store::Record::plain(b"portal: from=demo-device-01 to=portal kind=device-link".to_vec()),
);
map.insert(
(portal, Morton::encode(Point::new(2, 0, 0))),
b"portal: from=portal to=demo-device-01 kind=device-link".to_vec(),
cube_store::Record::plain(b"portal: from=portal to=demo-device-01 kind=device-link".to_vec()),
);
// Deliberately put a NON-ASCII value in, to prove the volume is just bytes
// and owes nothing to any filesystem's encoding rules.
map.insert(
(d1, Morton::encode(Point::new(9, 0, 0))),
"note: coordinate-addressed, no paths — 文件系统无关".as_bytes().to_vec(),
cube_store::Record::plain(
"note: coordinate-addressed, no paths — 文件系统无关".as_bytes().to_vec(),
),
);
serialize_store::<Morton>(&map)
crates/cube-store/src/lib.rs
+151 -36
View File
@@ -66,7 +66,7 @@ pub mod flagged;
pub mod wal;
use cube_core::{Coord, Curve, Morton, Point, SpaceId};
use cube_store_raw::{iter_records, parse_header, serialize as raw_serialize};
use cube_store_raw::{iter_records, parse_header, serialize_v4 as raw_serialize_v4};
use std::collections::BTreeMap;
use std::fs::OpenOptions;
use std::io::Seek;
@@ -150,6 +150,43 @@ impl Region {
}
}
/// A stored record: its bytes, and the class mask its writer stamped.
///
/// The mask travels with the value rather than beside it, because the two are written together and
/// mean something together. A record's class is a property of *that record* — not of the store, not
/// of the space — and a model that kept masks in a second map would carry an invariant to maintain
/// by hand, which is a thing that breaks between two updates rather than in one.
///
/// `flags` is a raw 16-bit mask and nothing here interprets a bit of it: which bits are which class
/// is a vocabulary's business ([`cube_core::EventFlags`] today). A model that decided what they
/// meant would be a second vocabulary.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Record {
/// The class mask the writer stamped. 0 is "no class" — what every record written before the
/// field existed reads as, so not classifying is not writing a special value.
pub flags: u16,
/// The record's bytes.
pub value: Vec<u8>,
}
impl Record {
/// A record with no class.
pub fn plain(value: Vec<u8>) -> Self {
Record { flags: 0, value }
}
/// The record's bytes.
pub fn bytes(&self) -> &[u8] {
&self.value
}
}
impl From<Vec<u8>> for Record {
fn from(value: Vec<u8>) -> Self {
Record::plain(value)
}
}
/// The storage contract. Implemented by every backend.
///
/// Methods are curve-agnostic at the trait level: a backend picks its own
@@ -162,6 +199,17 @@ pub trait Store {
/// one existed.
fn put(&mut self, coord: Coord, value: Vec<u8>) -> Result<(), StoreError>;
/// Insert or overwrite a record at `coord`, stamped with `flags`.
///
/// Separate from [`put`](Self::put) rather than replacing it, because classification is
/// something a *caller* does and most callers do not: a backend that cannot carry a mask is
/// still a `Store`, and the default here says so by writing the value and dropping the mask
/// rather than by refusing a write.
fn put_flagged(&mut self, coord: Coord, value: Vec<u8>, flags: u16) -> Result<(), StoreError> {
let _ = flags;
self.put(coord, value)
}
/// Fetch the value at `coord`, if present.
fn get(&self, coord: &Coord) -> Result<Option<Vec<u8>>, StoreError>;
@@ -227,7 +275,7 @@ pub trait Store {
/// choice affects only key layout and therefore range-query run count — the
/// API and semantics are identical.
pub struct MemStore<C: Curve = Morton> {
map: BTreeMap<(SpaceId, C::Key), Vec<u8>>,
map: BTreeMap<(SpaceId, C::Key), Record>,
_curve: PhantomData<C>,
#[cfg(feature = "debug")]
stats: debug::StoreStats,
@@ -276,8 +324,13 @@ impl<C: Curve> Default for MemStore<C> {
impl<C: Curve> Store for MemStore<C> {
fn put(&mut self, coord: Coord, value: Vec<u8>) -> Result<(), StoreError> {
self.put_flagged(coord, value, 0)
}
fn put_flagged(&mut self, coord: Coord, value: Vec<u8>, flags: u16) -> Result<(), StoreError> {
let key = C::encode(coord.point);
self.map.insert((coord.space, key), value);
self.map
.insert((coord.space, key), Record { flags, value });
#[cfg(feature = "debug")]
{
self.stats.puts.store(
@@ -297,7 +350,7 @@ impl<C: Curve> Store for MemStore<C> {
Ordering::Relaxed,
);
}
Ok(self.map.get(&(coord.space, key)).cloned())
Ok(self.map.get(&(coord.space, key)).map(|r| r.value.clone()))
}
fn delete(&mut self, coord: &Coord) -> Result<Option<Vec<u8>>, StoreError> {
@@ -309,7 +362,7 @@ impl<C: Curve> Store for MemStore<C> {
Ordering::Relaxed,
);
}
Ok(self.map.remove(&(coord.space, key)))
Ok(self.map.remove(&(coord.space, key)).map(|r| r.value))
}
fn range(&self, space: &SpaceId, region: &Region) -> Result<Vec<(Coord, Vec<u8>)>, StoreError> {
@@ -378,7 +431,7 @@ impl<C: Curve> Store for MemStore<C> {
.map
.iter()
.filter(|((sp, _), _)| sp == space)
.map(|((sp, k), v)| (Coord::new(*sp, C::decode(*k)), v.clone()))
.map(|((sp, k), v)| (Coord::new(*sp, C::decode(*k)), v.value.clone()))
.collect();
out.sort_by(|a, b| C::encode(a.0.point).cmp(&C::encode(b.0.point)));
Ok(out)
@@ -577,7 +630,7 @@ pub fn aligned_boxes(region: &Region) -> Vec<Box3> {
/// a walk is every entry in the map: the other spaces are rejected *by looking
/// at their keys*, so charging it only for its own space would understate it.
fn scan_region<C: Curve>(
map: &BTreeMap<(SpaceId, C::Key), Vec<u8>>,
map: &BTreeMap<(SpaceId, C::Key), Record>,
space: &SpaceId,
region: &Region,
) -> (Vec<(Coord, Vec<u8>)>, u64) {
@@ -590,7 +643,7 @@ fn scan_region<C: Curve>(
}
let p = C::decode(*k);
if region.contains(&p) {
out.push((Coord::new(*space, p), v.clone()));
out.push((Coord::new(*space, p), v.value.clone()));
}
}
(out, visited)
@@ -609,7 +662,7 @@ fn scan_region<C: Curve>(
/// sufficient to test membership alone — which is why the span has to be sound
/// in the first place, and why a curve may only offer one if it is monotone.
fn seek_region<C: Curve>(
map: &BTreeMap<(SpaceId, C::Key), Vec<u8>>,
map: &BTreeMap<(SpaceId, C::Key), Record>,
space: &SpaceId,
region: &Region,
foot: C::Key,
@@ -623,7 +676,7 @@ fn seek_region<C: Curve>(
visited += 1;
let p = C::decode(*k);
if region.contains(&p) {
out.push((Coord::new(*space, p), v.clone()));
out.push((Coord::new(*space, p), v.value.clone()));
}
}
(out, visited)
@@ -648,7 +701,7 @@ fn seek_region<C: Curve>(
/// `visited / returned` is the over-coverage the span pays, and it is the number
/// worth publishing: the honest cost of answering a region query on a curve.
fn span_scan<C: Curve>(
map: &BTreeMap<(SpaceId, C::Key), Vec<u8>>,
map: &BTreeMap<(SpaceId, C::Key), Record>,
space: &SpaceId,
region: &Region,
) -> (Vec<(Coord, Vec<u8>)>, u64) {
@@ -685,7 +738,7 @@ fn span_scan<C: Curve>(
///
/// [`flush`]: FileBackedStore::flush
pub struct FileBackedStore<C: Curve<Key = [u8; 24]> = Morton> {
map: BTreeMap<(SpaceId, C::Key), Vec<u8>>,
map: BTreeMap<(SpaceId, C::Key), Record>,
path: PathBuf,
/// Tracks whether the in-RAM `map` has diverged from the on-disk file
/// since the last [`flush`]. Lets a redundant commit be a cheap no-op
@@ -787,15 +840,23 @@ impl<C: Curve<Key = [u8; 24]>> Drop for FileBackedStore<C> {
impl<C: Curve<Key = [u8; 24]>> Store for FileBackedStore<C> {
fn put(&mut self, coord: Coord, value: Vec<u8>) -> Result<(), StoreError> {
self.put_flagged(coord, value, 0)
}
/// A mask does not make the flush path any different: `flush` rewrites the image from the map,
/// and the map is where the mask lives, so a classified record is durable exactly when any
/// other record is.
fn put_flagged(&mut self, coord: Coord, value: Vec<u8>, flags: u16) -> Result<(), StoreError> {
let key = C::encode(coord.point);
self.map.insert((coord.space, key), value);
self.map
.insert((coord.space, key), Record { flags, value });
self.dirty.store(true, Ordering::SeqCst);
Ok(())
}
fn get(&self, coord: &Coord) -> Result<Option<Vec<u8>>, StoreError> {
let key = C::encode(coord.point);
Ok(self.map.get(&(coord.space, key)).cloned())
Ok(self.map.get(&(coord.space, key)).map(|r| r.value.clone()))
}
fn delete(&mut self, coord: &Coord) -> Result<Option<Vec<u8>>, StoreError> {
@@ -804,7 +865,7 @@ impl<C: Curve<Key = [u8; 24]>> Store for FileBackedStore<C> {
if removed.is_some() {
self.dirty.store(true, Ordering::SeqCst);
}
Ok(removed)
Ok(removed.map(|r| r.value))
}
fn range(&self, space: &SpaceId, region: &Region) -> Result<Vec<(Coord, Vec<u8>)>, StoreError> {
@@ -826,7 +887,7 @@ impl<C: Curve<Key = [u8; 24]>> Store for FileBackedStore<C> {
.map
.iter()
.filter(|((sp, _), _)| sp == space)
.map(|((sp, k), v)| (Coord::new(*sp, C::decode(*k)), v.clone()))
.map(|((sp, k), v)| (Coord::new(*sp, C::decode(*k)), v.value.clone()))
.collect();
out.sort_by(|a, b| C::encode(a.0.point).cmp(&C::encode(b.0.point)));
Ok(out)
@@ -863,7 +924,7 @@ pub(crate) fn curve_tag<C: Curve>() -> u8 {
/// walks records; the raw crate does the zero-copy slice work.
pub(crate) fn read_records<C: Curve<Key = [u8; 24]>>(
bytes: &[u8],
map: &mut BTreeMap<(SpaceId, C::Key), Vec<u8>>,
map: &mut BTreeMap<(SpaceId, C::Key), Record>,
) -> Result<(), StoreError> {
let tag = parse_header(bytes).map_err(|e| {
StoreError::Io(std::io::Error::new(
@@ -887,7 +948,7 @@ pub(crate) fn read_records<C: Curve<Key = [u8; 24]>>(
})?;
let sp = SpaceId::from_bytes(*rec.space);
let key: C::Key = *rec.key;
map.insert((sp, key), rec.value.to_vec());
map.insert((sp, key), Record { flags: rec.flags, value: rec.value.to_vec() });
}
Ok(())
}
@@ -905,17 +966,17 @@ pub(crate) fn read_records<C: Curve<Key = [u8; 24]>>(
///
/// `Key = [u8; 24]` is fixed (three 40-bit Morton lanes — DESIGN §8.1 answer).
pub fn serialize_store<C: Curve<Key = [u8; 24]>>(
map: &BTreeMap<(SpaceId, C::Key), Vec<u8>>,
map: &BTreeMap<(SpaceId, C::Key), Record>,
) -> Vec<u8> {
// The actual byte layout lives in `cube-store-raw` (shared with the kernel
// driver). Here we only marshal our `(SpaceId, C::Key, Vec<u8>)` map into the
// `([u8;32], [u8;24], &[u8])` entries the raw serializer expects. `C::Key` is
// exactly `[u8; 24]` for every curve, so the cast is lossless.
let entries: Vec<(SpaceId, [u8; 24], &[u8])> = map
// The byte layout lives in `cube-store-raw` (shared with the kernel driver). Here we marshal
// our `(SpaceId, C::Key, Record)` map into the entries the raw serializer expects. `C::Key` is
// exactly `[u8; 24]` for every curve, so the cast is lossless, and a `BTreeMap` iterates in
// `(space, key)` order — which is the order the v4 space table's `first` column depends on.
let entries: Vec<(SpaceId, [u8; 24], u16, &[u8])> = map
.iter()
.map(|((sp, k), v)| (*sp, *k, v.as_slice()))
.map(|((sp, k), r)| (*sp, *k, r.flags, r.value.as_slice()))
.collect();
raw_serialize(curve_tag::<C>(), &entries)
raw_serialize_v4(curve_tag::<C>(), &entries)
}
/// A raw block device: fixed-size sectors addressed by 0-based index.
@@ -1062,7 +1123,7 @@ impl BlockDevice for FileBlockDevice {
/// (whole-store rewrite, O(N)) — crash-consistent journaling is M4; a torn
/// write at most loses the last op, which matches `FileBackedStore`'s contract.
pub struct RawBlockStore<B: BlockDevice, C: Curve<Key = [u8; 24]> = Morton> {
map: BTreeMap<(SpaceId, C::Key), Vec<u8>>,
map: BTreeMap<(SpaceId, C::Key), Record>,
dev: B,
_curve: PhantomData<C>,
}
@@ -1114,19 +1175,24 @@ impl<B: BlockDevice, C: Curve<Key = [u8; 24]>> RawBlockStore<B, C> {
impl<B: BlockDevice, C: Curve<Key = [u8; 24]>> Store for RawBlockStore<B, C> {
fn put(&mut self, coord: Coord, value: Vec<u8>) -> Result<(), StoreError> {
self.put_flagged(coord, value, 0)
}
fn put_flagged(&mut self, coord: Coord, value: Vec<u8>, flags: u16) -> Result<(), StoreError> {
let key = C::encode(coord.point);
self.map.insert((coord.space, key), value);
self.map
.insert((coord.space, key), Record { flags, value });
Ok(())
}
fn get(&self, coord: &Coord) -> Result<Option<Vec<u8>>, StoreError> {
let key = C::encode(coord.point);
Ok(self.map.get(&(coord.space, key)).cloned())
Ok(self.map.get(&(coord.space, key)).map(|r| r.value.clone()))
}
fn delete(&mut self, coord: &Coord) -> Result<Option<Vec<u8>>, StoreError> {
let key = C::encode(coord.point);
Ok(self.map.remove(&(coord.space, key)))
Ok(self.map.remove(&(coord.space, key)).map(|r| r.value))
}
fn range(&self, space: &SpaceId, region: &Region) -> Result<Vec<(Coord, Vec<u8>)>, StoreError> {
@@ -1149,7 +1215,7 @@ impl<B: BlockDevice, C: Curve<Key = [u8; 24]>> Store for RawBlockStore<B, C> {
.map
.iter()
.filter(|((sp, _), _)| sp == space)
.map(|((sp, k), v)| (Coord::new(*sp, C::decode(*k)), v.clone()))
.map(|((sp, k), v)| (Coord::new(*sp, C::decode(*k)), v.value.clone()))
.collect();
out.sort_by(|a, b| C::encode(a.0.point).cmp(&C::encode(b.0.point)));
Ok(out)
@@ -1295,6 +1361,55 @@ mod tests {
let _ = std::fs::remove_file(&path);
}
/// A class mask makes the round trip: written through the store, out to the image, and back
/// with the record it belongs to.
///
/// The mask is asserted at the *image* rather than only through the store's own `get`, because
/// `get` returns the value and would have hidden a mask that never left memory. A store that
/// carried a class correctly and persisted none of it is exactly what a model change can leave
/// behind, and it is invisible from the outside.
#[test]
fn a_class_mask_makes_the_round_trip() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("classified.img");
{
let mut s = FileBackedStore::<Morton>::open(&path).unwrap();
s.put_flagged(
Coord::root(Point::new(1, 0, 0)),
b"classified".to_vec(),
0x0022,
)
.unwrap();
// A record nobody classified, written through the unclassified path: it must come back
// as "no class" rather than as whatever the last record happened to carry.
s.put(Coord::root(Point::new(2, 0, 0)), b"plain".to_vec())
.unwrap();
s.flush().unwrap();
}
let bytes = std::fs::read(&path).unwrap();
assert_eq!(
cube_store_raw::parse(&bytes).unwrap().version,
cube_store_raw::VERSION_V4,
"a store that carries a class has to write the layout that holds one"
);
let mut written: std::vec::Vec<(u64, u16)> = cube_store_raw::iter_records(&bytes)
.map(|r| {
let r = r.unwrap();
(Morton::decode(*r.key).x, r.flags)
})
.collect();
written.sort();
assert_eq!(written, std::vec![(1, 0x0022), (2, 0)]);
// And the store reads its own mask back: the image is the source, not a cache.
let s = FileBackedStore::<Morton>::open(&path).unwrap();
assert_eq!(
s.get(&Coord::root(Point::new(1, 0, 0))).unwrap(),
Some(b"classified".to_vec())
);
}
#[test]
fn overwrite_returns_previous() {
let mut s = MemStore::<Morton>::new();
@@ -2006,7 +2121,7 @@ mod padding_tests {
let mut map = BTreeMap::new();
for sp in s.spaces().unwrap() {
for (c, v) in s.entries(&sp).unwrap() {
map.insert((c.space, Morton::encode(c.point)), v);
map.insert((c.space, Morton::encode(c.point)), Record::plain(v));
}
}
let unpadded = serialize_store::<Morton>(&map);
@@ -2023,7 +2138,7 @@ mod padding_tests {
);
assert_eq!(
back.get(&(SpaceId::ROOT, Morton::encode(Point::ORIGIN)))
.map(|v| v.as_slice()),
.map(|r| r.value.as_slice()),
Some(b"REAL VALUE".as_slice()),
"padding must not empty the origin record"
);
@@ -2043,7 +2158,7 @@ mod padding_tests {
let mut map = BTreeMap::new();
for sp in s.spaces().unwrap() {
for (c, v) in s.entries(&sp).unwrap() {
map.insert((c.space, Morton::encode(c.point)), v);
map.insert((c.space, Morton::encode(c.point)), Record::plain(v));
}
}
let bytes = serialize_store::<Morton>(&map);
@@ -2052,12 +2167,12 @@ mod padding_tests {
assert_eq!(back.len(), 2, "the origin is a record, not an end marker");
assert_eq!(
back.get(&(SpaceId::ROOT, Morton::encode(Point::ORIGIN)))
.map(|v| v.as_slice()),
.map(|r| r.value.as_slice()),
Some(b"".as_slice())
);
assert_eq!(
back.get(&(SpaceId::ROOT, Morton::encode(Point::new(2, 2, 2))))
.map(|v| v.as_slice()),
.map(|r| r.value.as_slice()),
Some(b"after".as_slice())
);
}
crates/cube-store/src/wal.rs
+86 -33
View File
@@ -48,20 +48,29 @@ use std::io::{BufWriter, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
use cube_core::{Coord, Curve, SpaceId};
use crate::{curve_tag, read_records, serialize_store, Region, Store, StoreError};
use crate::{curve_tag, read_records, serialize_store, Record, Region, Store, StoreError};
/// Magic for the log file, distinct from the image's `CUBE` so the two cannot be
/// confused by a reader that opens the wrong one.
pub const WAL_MAGIC: [u8; 4] = *b"CUBW";
/// Layout version of the log.
/// The original log entry: no class mask.
pub const WAL_VERSION: u8 = 1;
/// The flagged log entry: the entry carries a `u16` class mask before its length. This is what the
/// writer here emits, and what the kernel's v4 migration folds into a v4 index.
pub const WAL_VERSION_V2: u8 = 2;
/// The class mask's width.
pub const FLAGS_LEN: usize = 2;
/// Fixed header length: magic(4) + version(1) + curve tag(1).
pub const WAL_HEADER_LEN: usize = 6;
/// Fixed part of an entry, before the value: op(1) + crc(4) + space(32) + key(24) + len(4).
pub const ENTRY_FIXED: usize = 1 + 4 + 32 + 24 + 4;
/// The v2 entry: the same, with a `flags(2)` field before the length.
pub const ENTRY_FIXED_V2: usize = 1 + 4 + 32 + 24 + FLAGS_LEN + 4;
/// A recorded mutation.
#[derive(Clone, Debug, PartialEq, Eq)]
@@ -72,6 +81,8 @@ pub enum WalOp {
coord: Coord,
/// What.
value: Vec<u8>,
/// The class mask the writer stamped; 0 is "no class".
flags: u16,
},
/// Remove the record at `coord`.
Delete {
@@ -158,7 +169,7 @@ impl Wal {
fn header(curve_tag: u8) -> [u8; WAL_HEADER_LEN] {
let mut h = [0u8; WAL_HEADER_LEN];
h[0..4].copy_from_slice(&WAL_MAGIC);
h[4] = WAL_VERSION;
h[4] = WAL_VERSION_V2;
h[5] = curve_tag;
h
}
@@ -172,9 +183,9 @@ impl Wal {
/// Append a mutation.
pub fn append<C: Curve<Key = [u8; 24]>>(&mut self, op: &WalOp) -> Result<(), StoreError> {
let (coord, value): (&Coord, &[u8]) = match op {
WalOp::Put { coord, value } => (coord, value),
WalOp::Delete { coord } => (coord, &[]),
let (coord, value, flags): (&Coord, &[u8], u16) = match op {
WalOp::Put { coord, value, flags } => (coord, value, *flags),
WalOp::Delete { coord } => (coord, &[], 0),
};
let key = C::encode(coord.point);
let len = u32::try_from(value.len()).map_err(|_| {
@@ -184,16 +195,18 @@ impl Wal {
))
})?;
// The checksum covers everything after it, so a corrupted coordinate, key or
// value is caught rather than applied.
let mut body = Vec::with_capacity(32 + 24 + 4 + value.len());
// The checksum covers everything after it, so a corrupted coordinate, key, mask or
// value is caught rather than applied — and it covers the mask, so a flipped bit in a
// record's class is a torn entry rather than a silent reclassification.
let mut body = Vec::with_capacity(32 + 24 + FLAGS_LEN + 4 + value.len());
body.extend_from_slice(coord.space.as_bytes());
body.extend_from_slice(&key);
body.extend_from_slice(&flags.to_le_bytes());
body.extend_from_slice(&len.to_le_bytes());
body.extend_from_slice(value);
let crc = crc32(&body);
let mut entry = Vec::with_capacity(ENTRY_FIXED + value.len());
let mut entry = Vec::with_capacity(ENTRY_FIXED_V2 + value.len());
entry.push(op.tag());
entry.extend_from_slice(&crc.to_le_bytes());
entry.extend_from_slice(&body);
@@ -258,7 +271,7 @@ impl Wal {
/// Reads come from the in-memory index, which is the image plus every logged mutation
/// — the same view `FileBackedStore` presents, with a cheaper write path.
pub struct WalStore<C: Curve<Key = [u8; 24]> = cube_core::Morton> {
map: BTreeMap<(SpaceId, C::Key), Vec<u8>>,
map: BTreeMap<(SpaceId, C::Key), Record>,
image: PathBuf,
wal: Wal,
/// Logged mutations since the last checkpoint.
@@ -359,15 +372,29 @@ impl<C: Curve<Key = [u8; 24]>> WalStore<C> {
impl<C: Curve<Key = [u8; 24]>> Store for WalStore<C> {
fn put(&mut self, coord: Coord, value: Vec<u8>) -> Result<(), StoreError> {
self.wal
.append::<C>(&WalOp::Put { coord, value: value.clone() })?;
self.map.insert((coord.space, C::encode(coord.point)), value);
self.put_flagged(coord, value, 0)
}
/// The mask goes into the log entry, which is what makes it durable in one write: the store's
/// whole claim is that an acknowledged mutation survives a power cut, and a class that lived
/// only in memory until the next checkpoint would not be part of that claim.
fn put_flagged(&mut self, coord: Coord, value: Vec<u8>, flags: u16) -> Result<(), StoreError> {
self.wal.append::<C>(&WalOp::Put {
coord,
value: value.clone(),
flags,
})?;
self.map
.insert((coord.space, C::encode(coord.point)), Record { flags, value });
self.since_checkpoint += 1;
self.maybe_checkpoint()
}
fn get(&self, coord: &Coord) -> Result<Option<Vec<u8>>, StoreError> {
Ok(self.map.get(&(coord.space, C::encode(coord.point))).cloned())
Ok(self
.map
.get(&(coord.space, C::encode(coord.point)))
.map(|r| r.value.clone()))
}
fn delete(&mut self, coord: &Coord) -> Result<Option<Vec<u8>>, StoreError> {
@@ -375,7 +402,7 @@ impl<C: Curve<Key = [u8; 24]>> Store for WalStore<C> {
let removed = self.map.remove(&(coord.space, C::encode(coord.point)));
self.since_checkpoint += 1;
self.maybe_checkpoint()?;
Ok(removed)
Ok(removed.map(|r| r.value))
}
fn range(
@@ -396,7 +423,7 @@ impl<C: Curve<Key = [u8; 24]>> Store for WalStore<C> {
.map
.iter()
.filter(|((sp, _), _)| sp == space)
.map(|((sp, k), v)| (Coord::new(*sp, C::decode(*k)), v.clone()))
.map(|((sp, k), v)| (Coord::new(*sp, C::decode(*k)), v.value.clone()))
.collect();
out.sort_by(|a, b| C::encode(a.0.point).cmp(&C::encode(b.0.point)));
Ok(out)
@@ -446,6 +473,9 @@ struct RawOp {
tag: u8,
space: SpaceId,
key: [u8; 24],
/// The class mask the entry carried. A version 1 entry has no mask field and reads as 0 — "no
/// class" — which is what every record written before the field existed means.
flags: u16,
value: Vec<u8>,
}
@@ -475,12 +505,19 @@ fn parse_log(bytes: &[u8], curve_tag: u8) -> std::io::Result<ParsedLog> {
"write-ahead log has the wrong magic",
));
}
if bytes[4] != WAL_VERSION {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
format!("write-ahead log version {} is not {WAL_VERSION}", bytes[4]),
));
}
// The header's version decides the entry stride, and it is read once here rather than tested
// per entry: a log framed one way and read another way mis-addresses every entry after the
// first, which reads as a corrupt log rather than as a version mismatch.
let (fixed, len_at, flagged) = match bytes[4] {
WAL_VERSION => (ENTRY_FIXED, 61, false),
WAL_VERSION_V2 => (ENTRY_FIXED_V2, 61 + FLAGS_LEN, true),
other => {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
format!("write-ahead log version {other} is not {WAL_VERSION} or {WAL_VERSION_V2}"),
));
}
};
if bytes[5] != curve_tag {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
@@ -493,7 +530,7 @@ fn parse_log(bytes: &[u8], curve_tag: u8) -> std::io::Result<ParsedLog> {
let mut ops = Vec::new();
let mut off = WAL_HEADER_LEN;
while off + ENTRY_FIXED <= bytes.len() {
while off + fixed <= bytes.len() {
let start = off;
let op = bytes[off];
if op != 1 && op != 2 {
@@ -504,27 +541,34 @@ fn parse_log(bytes: &[u8], curve_tag: u8) -> std::io::Result<ParsedLog> {
space.copy_from_slice(&bytes[off + 5..off + 37]);
let mut key = [0u8; 24];
key.copy_from_slice(&bytes[off + 37..off + 61]);
let len =
u32::from_le_bytes(bytes[off + 61..off + 65].try_into().expect("four bytes")) as usize;
let flags = if flagged {
u16::from_le_bytes(bytes[off + 61..off + 63].try_into().expect("two bytes"))
} else {
0
};
let len = u32::from_le_bytes(
bytes[off + len_at..off + len_at + 4].try_into().expect("four bytes"),
) as usize;
// The append point moves only for an entry that validated. Advancing first and
// checking after would leave a torn entry's frame *inside* the valid prefix: it would
// never be applied, but it would never be overwritten either, and the corruption would
// sit in the log looking like a frame that belongs there.
let frame_end = start + ENTRY_FIXED + len;
let frame_end = start + fixed + len;
if frame_end > bytes.len() {
break;
}
// The checksum covers space, key, length and value, so a corrupted entry is
// The checksum covers space, key, the mask, length and value, so a corrupted entry is
// stopped at rather than applied.
if crc32(&bytes[start + 5..frame_end]) != crc {
break;
}
let value = bytes[start + ENTRY_FIXED..frame_end].to_vec();
let value = bytes[start + fixed..frame_end].to_vec();
off = frame_end;
ops.push(RawOp {
tag: op,
space: SpaceId::from_bytes(space),
key,
flags,
value,
});
}
@@ -539,7 +583,7 @@ fn parse_log(bytes: &[u8], curve_tag: u8) -> std::io::Result<ParsedLog> {
fn replay_into<C: Curve<Key = [u8; 24]>>(
path: &Path,
curve_tag: u8,
map: &mut BTreeMap<(SpaceId, C::Key), Vec<u8>>,
map: &mut BTreeMap<(SpaceId, C::Key), Record>,
) -> Result<u64, StoreError> {
let bytes = std::fs::read(path)?;
let parsed = parse_log(&bytes, curve_tag)?;
@@ -550,7 +594,10 @@ fn replay_into<C: Curve<Key = [u8; 24]>>(
debug_assert_eq!(C::encode(C::decode(key)), key, "the log's key round-trips");
match op.tag {
1 => {
map.insert((op.space, key), op.value.clone());
map.insert(
(op.space, key),
Record { flags: op.flags, value: op.value.clone() },
);
}
_ => {
map.remove(&(op.space, key));
@@ -667,9 +714,15 @@ mod tests {
// The same records, serialized independently, must be the same bytes.
let mut expected = BTreeMap::new();
for i in 1..=8u64 {
expected.insert((space(4), Morton::encode(Point::new(i, 0, 0))), format!("value {i}").into_bytes());
expected.insert(
(space(4), Morton::encode(Point::new(i, 0, 0))),
Record::plain(format!("value {i}").into_bytes()),
);
}
expected.insert((space(5), Morton::encode(Point::new(1, 0, 0))), vec![0u8; 300]);
expected.insert(
(space(5), Morton::encode(Point::new(1, 0, 0))),
Record::plain(vec![0u8; 300]),
);
let pinned = crate::serialize_store::<Morton>(&expected);
let actual = std::fs::read(&image).expect("read image");
assert_eq!(