From 436a9a71ee33f3ff8a57e4059307026ffc330bfe Mon Sep 17 00:00:00 2001 From: dapplion <35266934+dapplion@users.noreply.github.com> Date: Sun, 28 Jun 2026 14:00:32 +0200 Subject: [PATCH 1/7] Add static_file_storage crate Standalone slot-indexed append-only file storage library. One StaticFile owns a directory of data_*, data_*.off, and column.conf files; durable on return for put / put_batch, with one fsync per underlying file in the batched path instead of per-slot. u64 slots, no consensus types - any caller (cold-DB freezer, lcli ERA importer) can depend on this without depending on store / beacon_chain. File format documented in the lib.rs header. Benches at benches/large_writes.rs target the performance-critical ERA-load write path: ~30 MB no-compression state-snapshot records. Four cases: put_single, put_batch_eight_in_one_file (no-compression and snappy variants), put_batch_32_cross_file. --- Cargo.lock | 10 + Cargo.toml | 1 + common/static_file_storage/Cargo.toml | 16 + .../benches/large_writes.rs | 154 +++++ common/static_file_storage/src/lib.rs | 644 ++++++++++++++++++ 5 files changed, 825 insertions(+) create mode 100644 common/static_file_storage/Cargo.toml create mode 100644 common/static_file_storage/benches/large_writes.rs create mode 100644 common/static_file_storage/src/lib.rs diff --git a/Cargo.lock b/Cargo.lock index 9cb2617310c..ec1873ff55b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8650,6 +8650,16 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f" +[[package]] +name = "static_file_storage" +version = "0.1.0" +dependencies = [ + "criterion", + "parking_lot", + "snap", + "tempfile", +] + [[package]] name = "store" version = "0.2.0" diff --git a/Cargo.toml b/Cargo.toml index 3cbe113d522..5484865a919 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -40,6 +40,7 @@ members = [ "common/oneshot_broadcast", "common/pretty_reqwest_error", "common/slot_clock", + "common/static_file_storage", "common/system_health", "common/target_check", "common/task_executor", diff --git a/common/static_file_storage/Cargo.toml b/common/static_file_storage/Cargo.toml new file mode 100644 index 00000000000..01c034b6800 --- /dev/null +++ b/common/static_file_storage/Cargo.toml @@ -0,0 +1,16 @@ +[package] +name = "static_file_storage" +version = "0.1.0" +edition = { workspace = true } + +[dependencies] +parking_lot = { workspace = true } +snap = { workspace = true } + +[dev-dependencies] +criterion = { workspace = true } +tempfile = { workspace = true } + +[[bench]] +name = "large_writes" +harness = false diff --git a/common/static_file_storage/benches/large_writes.rs b/common/static_file_storage/benches/large_writes.rs new file mode 100644 index 00000000000..0141f0cd641 --- /dev/null +++ b/common/static_file_storage/benches/large_writes.rs @@ -0,0 +1,154 @@ +//! Benchmarks for the write path under ERA-load-shaped record sizes. +//! +//! The targeted profile is mainnet ERA import: ~10–30 MB state-snapshot and +//! state-diff records into columns where `compression: false` (the production +//! setting for those columns, since HDiff is already compressed internally). +//! +//! Each iteration fixtures a fresh temp directory so the bench measures +//! steady-state put cost, not first-open or healing overhead. +//! +//! Run with: `cargo bench -p static_file_storage --bench large_writes`. + +use criterion::{BatchSize, Criterion, Throughput, criterion_group, criterion_main}; +use static_file_storage::{Config, StaticFile}; +use std::hint::black_box; +use std::path::PathBuf; + +const RECORD_30MB: usize = 30 * 1024 * 1024; +const RECORD_4MB: usize = 4 * 1024 * 1024; + +const NO_COMPRESSION: Config = Config { + record_type: [0x04, 0x00], + compression: false, + max_value_bytes: 1024 * 1024 * 1024, +}; + +const SNAPPY: Config = Config { + record_type: [0x04, 0x00], + compression: true, + max_value_bytes: 1024 * 1024 * 1024, +}; + +fn payload(seed: u64, len: usize) -> Vec { + // xorshift fill — fast, deterministic, incompressible enough that snappy + // can't shortcut (we care about the path it actually takes in production + // where HDiff outputs are already entropy-dense). + let mut buf = vec![0u8; len]; + let mut s = seed.wrapping_add(0x9E37_79B9_7F4A_7C15); + for chunk in buf.chunks_mut(8) { + s ^= s << 13; + s ^= s >> 7; + s ^= s << 17; + chunk.copy_from_slice(&s.to_le_bytes()[..chunk.len()]); + } + buf +} + +fn fresh_dir() -> (tempfile::TempDir, PathBuf) { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().to_path_buf(); + (dir, path) +} + +fn bench_put_single_30mb(c: &mut Criterion) { + let value = payload(1, RECORD_30MB); + let mut group = c.benchmark_group("put_single_30mb_no_compression"); + group.sample_size(20); + group.throughput(Throughput::Bytes(RECORD_30MB as u64)); + group.bench_function("put", |b| { + b.iter_batched( + || { + let (dir, path) = fresh_dir(); + let sf = StaticFile::open(path, NO_COMPRESSION).expect("open"); + (dir, sf) + }, + |(_dir, sf)| { + sf.put(0, black_box(&value)).expect("put"); + }, + BatchSize::PerIteration, + ); + }); + group.finish(); +} + +fn bench_put_batch_eight_30mb_one_file(c: &mut Criterion) { + let values: Vec<(u64, Vec)> = (0..8).map(|i| (i, payload(i, RECORD_30MB))).collect(); + let total_bytes = (RECORD_30MB * 8) as u64; + let mut group = c.benchmark_group("put_batch_eight_30mb_in_one_file_no_compression"); + group.sample_size(20); + group.throughput(Throughput::Bytes(total_bytes)); + group.bench_function("put_batch", |b| { + b.iter_batched( + || { + let (dir, path) = fresh_dir(); + let sf = StaticFile::open(path, NO_COMPRESSION).expect("open"); + (dir, sf, values.clone()) + }, + |(_dir, sf, items)| { + sf.put_batch(items).expect("put_batch"); + }, + BatchSize::PerIteration, + ); + }); + group.finish(); +} + +fn bench_put_batch_eight_30mb_snappy(c: &mut Criterion) { + // Same shape as the no-compression bench but with snappy on, to quantify + // the cost we explicitly avoid by setting `compression: false` for the + // state-snapshot / state-diff columns. + let values: Vec<(u64, Vec)> = (0..8).map(|i| (i, payload(i, RECORD_30MB))).collect(); + let total_bytes = (RECORD_30MB * 8) as u64; + let mut group = c.benchmark_group("put_batch_eight_30mb_in_one_file_snappy"); + group.sample_size(10); + group.throughput(Throughput::Bytes(total_bytes)); + group.bench_function("put_batch", |b| { + b.iter_batched( + || { + let (dir, path) = fresh_dir(); + let sf = StaticFile::open(path, SNAPPY).expect("open"); + (dir, sf, values.clone()) + }, + |(_dir, sf, items)| { + sf.put_batch(items).expect("put_batch"); + }, + BatchSize::PerIteration, + ); + }); + group.finish(); +} + +fn bench_put_batch_32_records_4mb_cross_file(c: &mut Criterion) { + // 32 × 4 MB records spanning a file boundary at slot 8192. Catches + // anything quadratic in the per-group rewrite path. + let values: Vec<(u64, Vec)> = (8176u64..8208) + .map(|i| (i, payload(i, RECORD_4MB))) + .collect(); + let total_bytes = (RECORD_4MB * values.len()) as u64; + let mut group = c.benchmark_group("put_batch_32_records_4mb_cross_file_no_compression"); + group.sample_size(20); + group.throughput(Throughput::Bytes(total_bytes)); + group.bench_function("put_batch", |b| { + b.iter_batched( + || { + let (dir, path) = fresh_dir(); + let sf = StaticFile::open(path, NO_COMPRESSION).expect("open"); + (dir, sf, values.clone()) + }, + |(_dir, sf, items)| { + sf.put_batch(items).expect("put_batch"); + }, + BatchSize::PerIteration, + ); + }); + group.finish(); +} + +criterion_group!( + benches, + bench_put_single_30mb, + bench_put_batch_eight_30mb_one_file, + bench_put_batch_eight_30mb_snappy, + bench_put_batch_32_records_4mb_cross_file, +); +criterion_main!(benches); diff --git a/common/static_file_storage/src/lib.rs b/common/static_file_storage/src/lib.rs new file mode 100644 index 00000000000..f2cf28c82c8 --- /dev/null +++ b/common/static_file_storage/src/lib.rs @@ -0,0 +1,644 @@ +//! Slot-indexed, append-only static file storage. +//! +//! `StaticFile` owns one directory containing: +//! +//! ```text +//! / +//! data_{file_id:05} # file_id = slot / SLOTS_PER_FILE +//! data_{file_id:05}.off # SLOTS_PER_FILE × u64 LE offsets, 0 = no record +//! column.conf # 36-byte commit marker, atomic-renamed +//! ``` +//! +//! # File format +//! +//! Data file: e2store version record (`65 32 00 00 00 00 00 00`), then records +//! appended as `type[2] | length[4 LE] | reserved[2]=0 | payload` (snappy- +//! framed if `Config::compression`). +//! +//! `column.conf`: `b"LHSTBLK2" | highest_slot u64 LE (u64::MAX = empty) | +//! current_data_len u64 LE | record_type[2] | compression u8 | reserved | max_value_bytes u64 LE`. +//! Atomic update: write `.tmp`, fsync, rename, fsync dir. +//! +//! # Put contract +//! +//! Durable on return. Slots arrive ascending **or** are identical-value +//! re-puts of an already-committed slot (so migration retries after a +//! mid-loop crash are safe). Previously-skipped slots (offset 0) cannot +//! be filled — that would break the append-only data file. +//! +//! # Recovery on open +//! +//! Data file truncated to `current_data_len`; `.off` entries beyond +//! `highest_slot` cleared. The `column.conf` rename is the commit point. + +use parking_lot::Mutex; +use snap::{read::FrameDecoder, write::FrameEncoder}; +use std::{ + fmt, + fs::{self, File, OpenOptions}, + io::{self, Read, Seek, SeekFrom, Write}, + path::{Path, PathBuf}, +}; + +pub const SLOTS_PER_FILE: u64 = 8192; +const OFFSET_SIZE: u64 = 8; +const OFFSET_FILE_LEN: u64 = SLOTS_PER_FILE * OFFSET_SIZE; +const CONFIG_FILE: &str = "column.conf"; +const CONFIG_TMP_FILE: &str = "column.conf.tmp"; +const DATA_FILE_PREFIX: &str = "data_"; +const CONFIG_MAGIC: &[u8; 8] = b"LHSTBLK2"; +const CONFIG_LEN: usize = 36; +/// Empty-store sentinel for `highest_written_slot` in the per-file config. +const EMPTY_SLOT: u64 = u64::MAX; +/// e2store version record, written once at the start of each data file. +const VERSION_RECORD: [u8; 8] = [0x65, 0x32, 0, 0, 0, 0, 0, 0]; + +const COMPRESSION_NONE: u8 = 0; +const COMPRESSION_SNAPPY: u8 = 1; + +/// On-disk format settings for one `StaticFile`. On first creation the build's +/// values are persisted; on re-open the persisted values win, with +/// `max_value_bytes` ratcheted upward if the build allows larger records. +#[derive(Debug, Clone, Copy)] +pub struct Config { + /// e2store record type tag. + pub record_type: [u8; 2], + /// Whether values are snappy-framed before write. + pub compression: bool, + /// Upper bound on a single decoded record's size in bytes. + pub max_value_bytes: u64, +} + +struct ConfigOnDisk { + highest_written_slot: Option, + current_data_len: u64, + record_type: [u8; 2], + compression: bool, + max_value_bytes: u64, +} + +pub type Result = std::result::Result; + +#[derive(Debug)] +pub enum Error { + Io(io::Error), + Compression(io::Error), + Invalid(String), +} + +impl fmt::Display for Error { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Io(e) => write!(f, "static file io error: {e}"), + Self::Compression(e) => write!(f, "static file compression error: {e}"), + Self::Invalid(message) => write!(f, "static file invalid data: {message}"), + } + } +} + +impl std::error::Error for Error { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::Io(e) | Self::Compression(e) => Some(e), + Self::Invalid(_) => None, + } + } +} + +impl From for Error { + fn from(e: io::Error) -> Self { + Self::Io(e) + } +} + +/// Slot-keyed file set. Owns one directory of `data_*`, `data_*.off`, and +/// `column.conf` files. +#[derive(Debug)] +pub struct StaticFile { + root_dir: PathBuf, + config: Config, + highest_written_slot: Mutex>, +} + +impl StaticFile { + /// Open or create a `StaticFile` rooted at `root_dir`. On first creation, + /// `defaults` is persisted; on re-open, the persisted values win for + /// `record_type` and `compression`, and `max_value_bytes` is ratcheted up + /// to `max(on_disk, defaults)`. + pub fn open(root_dir: PathBuf, defaults: Config) -> Result { + fs::create_dir_all(&root_dir)?; + + // First-open: persist current-build defaults. Re-open: persisted + // settings win over `defaults`, which preserves on-disk readability + // even if the build's defaults change later. + let config_path = root_dir.join(CONFIG_FILE); + let tmp_path = root_dir.join(CONFIG_TMP_FILE); + if !config_path.exists() { + atomic_write_config(&config_path, &tmp_path, &root_dir, None, 0, &defaults)?; + } + + let on_disk = read_config(&config_path)?; + // record_type and compression are sticky — they're load-bearing for + // reading old records, so on-disk wins over build-time defaults. + // max_value_bytes is a soft bound used to cap accepted record sizes; + // ratchet it up if the build's default is larger so a newer build + // can write bigger records than an older one persisted, then + // re-persist immediately so future opens see the new bound. + let max_value_bytes = on_disk.max_value_bytes.max(defaults.max_value_bytes); + let config = Config { + record_type: on_disk.record_type, + compression: on_disk.compression, + max_value_bytes, + }; + if max_value_bytes != on_disk.max_value_bytes { + atomic_write_config( + &config_path, + &tmp_path, + &root_dir, + on_disk.highest_written_slot, + on_disk.current_data_len, + &config, + )?; + } + + let handle = Self { + root_dir, + config, + highest_written_slot: Mutex::new(None), + }; + + if let Some(slot) = on_disk.highest_written_slot { + handle.heal_current_file(slot, on_disk.current_data_len)?; + } + *handle.highest_written_slot.lock() = on_disk.highest_written_slot; + + Ok(handle) + } + + /// Slot of the most recently committed record, if any. + pub fn highest_written_slot(&self) -> Option { + *self.highest_written_slot.lock() + } + + /// Read the record at `slot`, if present. + pub fn get(&self, slot: u64) -> Result>> { + let Some(highest_written_slot) = *self.highest_written_slot.lock() else { + return Ok(None); + }; + if slot > highest_written_slot { + return Ok(None); + } + self.read_record(slot) + } + + /// `true` if a record exists at `slot`. Cheaper than `get` — only the + /// `.off` sidecar is consulted; the data file is not read. + pub fn contains(&self, slot: u64) -> Result { + let Some(highest_written_slot) = *self.highest_written_slot.lock() else { + return Ok(false); + }; + if slot > highest_written_slot { + return Ok(false); + } + Ok(self.read_offset(file_id(slot), slot)? != 0) + } + + /// Durably store `bytes` at `slot`. Slots must arrive strictly ascending, + /// or be an identical-value re-put of a previously committed slot + /// (re-puts at any committed slot are idempotent so migration retries + /// after a mid-loop crash are safe). A previously-skipped slot (offset + /// zero) cannot be filled — that would break the append-only data file. + pub fn put(&self, slot: u64, bytes: &[u8]) -> Result<()> { + let mut highest_written_slot = self.highest_written_slot.lock(); + if let Some(highest) = *highest_written_slot + && slot <= highest + { + let existing = self.read_record(slot)?.ok_or_else(|| { + Error::Invalid(format!( + "re-put at slot {slot} <= highest {highest} but no record exists; \ + cannot fill a previously-skipped slot" + )) + })?; + if existing == bytes { + return Ok(()); + } + return Err(Error::Invalid(format!( + "re-put at slot {slot} with mismatched value" + ))); + } + + let payload = if self.config.compression { + compress_record(bytes)? + } else { + bytes.to_vec() + }; + let payload_len = + u32::try_from(payload.len()).map_err(|_| Error::Invalid("record too large".into()))?; + + let target_file_id = file_id(slot); + // Discard an uncommitted next-file tail after a crash. + let reset_file = (*highest_written_slot).map(file_id) != Some(target_file_id); + let off_pos = offset_position(slot); + let data_path = self.data_path(target_file_id); + let off_path = self.offset_path(target_file_id); + + let mut data_file = OpenOptions::new() + .read(true) + .append(true) + .create(true) + .open(&data_path)?; + if reset_file { + data_file.set_len(0)?; + } + + if data_file.metadata()?.len() == 0 { + data_file.write_all(&VERSION_RECORD)?; + } + + let offset = data_file.seek(SeekFrom::End(0))?; + write_record( + &mut data_file, + self.config.record_type, + payload_len, + &payload, + )?; + let data_len = data_file.seek(SeekFrom::End(0))?; + // Data and offset files must hit disk before the config commit marker. + data_file.sync_all()?; + + let mut off_file = OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(&off_path)?; + if reset_file { + off_file.set_len(0)?; + } + if off_file.metadata()?.len() < OFFSET_FILE_LEN { + off_file.set_len(OFFSET_FILE_LEN)?; + } + off_file.seek(SeekFrom::Start(off_pos))?; + off_file.write_all(&offset.to_le_bytes())?; + off_file.sync_all()?; + + // Atomic config update is the commit point. + self.write_config(Some(slot), data_len)?; + *highest_written_slot = Some(slot); + + Ok(()) + } + + /// Append `items` with one fsync per file (data + offset), not per slot. + /// Whole batch is durable on return — the same caller-visible contract as + /// `put` — but with O(1) syncs per underlying file instead of O(n) per + /// item. + /// + /// Walks `items` once, grouping them by `file_id`. For each group, opens + /// the data file and offset file once, appends every record's bytes + /// (collecting `(slot, offset)` pairs in memory), writes the offset + /// table, fsyncs both files, then commits via `write_config`. Idempotent + /// re-put of `items[0]` at `highest_written_slot` is honored as in `put`. + pub fn put_batch(&self, items: Vec<(u64, Vec)>) -> Result<()> { + if items.is_empty() { + return Ok(()); + } + + // Validate ascending order up front (cheap, catches caller bugs). + for w in items.windows(2) { + if w[1].0 <= w[0].0 { + return Err(Error::Invalid( + "put_batch slots must be strictly ascending".into(), + )); + } + } + + let mut highest_written_slot = self.highest_written_slot.lock(); + let mut iter = items.into_iter().peekable(); + + // Idempotent re-put: if the first item is exactly highest_written_slot + // with matching bytes, drop it from the batch. + if let (Some(highest), Some((first_slot, _))) = (*highest_written_slot, iter.peek()) { + if *first_slot < highest { + return Err(Error::Invalid( + "put_batch out of order vs highest_written_slot".into(), + )); + } + if *first_slot == highest { + let (slot, value) = iter.next().expect("peeked"); + let existing = self + .read_record(slot)? + .ok_or_else(|| Error::Invalid("missing record at highest slot".into()))?; + if existing != value { + return Err(Error::Invalid("re-put with mismatched value".into())); + } + } + } + + // Group remaining items by file_id, write each group with a single + // fsync per file. + let mut last_slot: Option = None; + let mut last_data_len: u64 = 0; + while iter.peek().is_some() { + let target_file_id = file_id(iter.peek().expect("peeked").0); + let mut group: Vec<(u64, Vec)> = Vec::new(); + while let Some(&(slot, _)) = iter.peek() { + if file_id(slot) != target_file_id { + break; + } + group.push(iter.next().expect("peeked")); + } + + let reset_file = (*highest_written_slot).map(file_id) != Some(target_file_id); + let data_path = self.data_path(target_file_id); + let off_path = self.offset_path(target_file_id); + + // Data file: append all records, then fsync once. + let mut data_file = OpenOptions::new() + .read(true) + .append(true) + .create(true) + .open(&data_path)?; + if reset_file { + data_file.set_len(0)?; + } + if data_file.metadata()?.len() == 0 { + data_file.write_all(&VERSION_RECORD)?; + } + // BufWriter coalesces the small-record header writes (8 bytes) and + // the small payloads into larger syscalls. + let mut offsets: Vec<(u64, u64)> = Vec::with_capacity(group.len()); + { + let mut writer = std::io::BufWriter::with_capacity(1 << 20, &mut data_file); + let mut cursor = writer.get_ref().metadata()?.len(); + for (slot, value) in &group { + let payload: std::borrow::Cow<'_, [u8]> = if self.config.compression { + compress_record(value)?.into() + } else { + value.as_slice().into() + }; + let payload_len = u32::try_from(payload.len()) + .map_err(|_| Error::Invalid("record too large".into()))?; + offsets.push((*slot, cursor)); + // Inline `write_record` to avoid the `&mut File` -> BufWriter mismatch. + writer.write_all(&self.config.record_type)?; + writer.write_all(&payload_len.to_le_bytes())?; + writer.write_all(&0u16.to_le_bytes())?; + writer.write_all(&payload)?; + cursor += 8 + payload.len() as u64; + } + writer.flush()?; + } + let data_len = data_file.seek(SeekFrom::End(0))?; + data_file.sync_all()?; + + // Offset file: open, ensure full size, write all offsets in seek+write + // pairs (8 bytes each), then fsync once. + let mut off_file = OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(&off_path)?; + if reset_file { + off_file.set_len(0)?; + } + if off_file.metadata()?.len() < OFFSET_FILE_LEN { + off_file.set_len(OFFSET_FILE_LEN)?; + } + for (slot, offset) in &offsets { + off_file.seek(SeekFrom::Start(offset_position(*slot)))?; + off_file.write_all(&offset.to_le_bytes())?; + } + off_file.sync_all()?; + + // Track final slot/data_len for the single config commit at end of batch. + if let Some((s, _)) = group.last() { + last_slot = Some(*s); + last_data_len = data_len; + } + *highest_written_slot = last_slot; + } + + // Single atomic config commit covering the whole batch. + if let Some(s) = last_slot { + self.write_config(Some(s), last_data_len)?; + } + + Ok(()) + } + + /// Read a record at `slot` without consulting the writer mutex. Used by + /// callers that already hold the lock (`put` for the idempotency check). + fn read_record(&self, slot: u64) -> Result>> { + let file_id = file_id(slot); + let offset = self.read_offset(file_id, slot)?; + if offset == 0 { + return Ok(None); + } + + let data_path = self.data_path(file_id); + let mut data_file = File::open(&data_path)?; + data_file.seek(SeekFrom::Start(offset))?; + + let mut header = [0; 8]; + data_file.read_exact(&mut header)?; + if header[0..2] != self.config.record_type || header[6..8] != [0, 0] { + return Err(Error::Invalid("invalid record header".into())); + } + + let len = u32::from_le_bytes([header[2], header[3], header[4], header[5]]) as usize; + let mut payload = vec![0; len]; + data_file.read_exact(&mut payload)?; + + if self.config.compression { + decompress_record(&payload, self.config.max_value_bytes).map(Some) + } else { + if (payload.len() as u64) > self.config.max_value_bytes { + return Err(Error::Invalid("record exceeds size limit".into())); + } + Ok(Some(payload)) + } + } + + fn heal_current_file(&self, slot: u64, current_data_len: u64) -> Result<()> { + let file_id = file_id(slot); + let data_path = self.data_path(file_id); + let data_file = OpenOptions::new().read(true).write(true).open(&data_path)?; + let data_len = data_file.metadata()?.len(); + if data_len < current_data_len { + return Err(Error::Invalid( + "data file shorter than committed length".into(), + )); + } + if data_len != current_data_len { + data_file.set_len(current_data_len)?; + data_file.sync_all()?; + } + + let off_path = self.offset_path(file_id); + let mut off_file = OpenOptions::new().read(true).write(true).open(&off_path)?; + let required_len = offset_position(slot) + OFFSET_SIZE; + let off_len = off_file.metadata()?.len(); + if off_len < required_len { + return Err(Error::Invalid( + "offset file shorter than committed slot".into(), + )); + } + if off_len < OFFSET_FILE_LEN { + off_file.set_len(OFFSET_FILE_LEN)?; + } + + let clear_start = required_len; + if clear_start < OFFSET_FILE_LEN { + // Remove offsets to entries beyond the committed slot. + off_file.seek(SeekFrom::Start(clear_start))?; + let zeroes = vec![0; (OFFSET_FILE_LEN - clear_start) as usize]; + off_file.write_all(&zeroes)?; + off_file.sync_all()?; + } + + Ok(()) + } + + fn write_config(&self, highest_written_slot: Option, current_data_len: u64) -> Result<()> { + atomic_write_config( + &self.config_path(), + &self.root_dir.join(CONFIG_TMP_FILE), + &self.root_dir, + highest_written_slot, + current_data_len, + &self.config, + ) + } + + fn read_offset(&self, file_id: u64, slot: u64) -> Result { + let off_path = self.offset_path(file_id); + let mut off_file = File::open(&off_path)?; + let mut bytes = [0; 8]; + off_file.seek(SeekFrom::Start(offset_position(slot)))?; + off_file.read_exact(&mut bytes)?; + Ok(u64::from_le_bytes(bytes)) + } + + fn config_path(&self) -> PathBuf { + self.root_dir.join(CONFIG_FILE) + } + + fn data_path(&self, file_id: u64) -> PathBuf { + self.root_dir + .join(format!("{DATA_FILE_PREFIX}{file_id:05}")) + } + + fn offset_path(&self, file_id: u64) -> PathBuf { + self.root_dir + .join(format!("{DATA_FILE_PREFIX}{file_id:05}.off")) + } +} + +fn read_config(path: &Path) -> Result { + let bytes = fs::read(path)?; + if bytes.len() != CONFIG_LEN || &bytes[0..8] != CONFIG_MAGIC { + return Err(Error::Invalid("invalid config".into())); + } + let highest = u64::from_le_bytes(bytes[8..16].try_into().expect("slice length checked")); + let current_data_len = + u64::from_le_bytes(bytes[16..24].try_into().expect("slice length checked")); + let record_type = [bytes[24], bytes[25]]; + let compression = match bytes[26] { + COMPRESSION_NONE => false, + COMPRESSION_SNAPPY => true, + other => { + return Err(Error::Invalid(format!("unknown compression flag {other}"))); + } + }; + let max_value_bytes = + u64::from_le_bytes(bytes[28..36].try_into().expect("slice length checked")); + Ok(ConfigOnDisk { + highest_written_slot: (highest != EMPTY_SLOT).then_some(highest), + current_data_len, + record_type, + compression, + max_value_bytes, + }) +} + +fn atomic_write_config( + config_path: &Path, + tmp_path: &Path, + root_dir: &Path, + highest_written_slot: Option, + current_data_len: u64, + config: &Config, +) -> Result<()> { + let mut bytes = [0u8; CONFIG_LEN]; + bytes[0..8].copy_from_slice(CONFIG_MAGIC); + bytes[8..16].copy_from_slice(&highest_written_slot.unwrap_or(EMPTY_SLOT).to_le_bytes()); + bytes[16..24].copy_from_slice(¤t_data_len.to_le_bytes()); + bytes[24..26].copy_from_slice(&config.record_type); + bytes[26] = if config.compression { + COMPRESSION_SNAPPY + } else { + COMPRESSION_NONE + }; + bytes[27] = 0; + bytes[28..36].copy_from_slice(&config.max_value_bytes.to_le_bytes()); + + { + let mut tmp = File::create(tmp_path)?; + tmp.write_all(&bytes)?; + tmp.sync_all()?; + } + + fs::rename(tmp_path, config_path)?; + sync_dir(root_dir) +} + +fn file_id(slot: u64) -> u64 { + slot / SLOTS_PER_FILE +} + +fn offset_position(slot: u64) -> u64 { + (slot % SLOTS_PER_FILE) * OFFSET_SIZE +} + +fn compress_record(bytes: &[u8]) -> Result> { + let mut encoder = FrameEncoder::new(Vec::new()); + encoder.write_all(bytes).map_err(Error::Compression)?; + encoder.flush().map_err(Error::Compression)?; + Ok(encoder.get_ref().clone()) +} + +fn write_record( + file: &mut File, + record_type: [u8; 2], + payload_len: u32, + payload: &[u8], +) -> Result<()> { + file.write_all(&record_type)?; + file.write_all(&payload_len.to_le_bytes())?; + file.write_all(&0u16.to_le_bytes())?; + file.write_all(payload)?; + Ok(()) +} + +fn decompress_record(bytes: &[u8], max_value_bytes: u64) -> Result> { + let decoder = FrameDecoder::new(bytes); + let mut limited = decoder.take(max_value_bytes + 1); + let mut decompressed = Vec::new(); + limited + .read_to_end(&mut decompressed) + .map_err(Error::Compression)?; + if decompressed.len() as u64 > max_value_bytes { + return Err(Error::Invalid( + "record exceeds decompressed size limit".into(), + )); + } + Ok(decompressed) +} + +fn sync_dir(path: &Path) -> Result<()> { + let dir = File::open(path)?; + dir.sync_all()?; + Ok(()) +} From 1d3c899d2a0e9f915d45b3f76e744d742d34220e Mon Sep 17 00:00:00 2001 From: dapplion <35266934+dapplion@users.noreply.github.com> Date: Sun, 28 Jun 2026 14:05:31 +0200 Subject: [PATCH 2/7] static_file_storage: use rand_xorshift for bench payload --- Cargo.lock | 2 ++ common/static_file_storage/Cargo.toml | 2 ++ common/static_file_storage/benches/large_writes.rs | 13 +++---------- 3 files changed, 7 insertions(+), 10 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index ec1873ff55b..5abe4fc3119 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8656,6 +8656,8 @@ version = "0.1.0" dependencies = [ "criterion", "parking_lot", + "rand 0.9.2", + "rand_xorshift 0.4.0", "snap", "tempfile", ] diff --git a/common/static_file_storage/Cargo.toml b/common/static_file_storage/Cargo.toml index 01c034b6800..fe8d7d69c1a 100644 --- a/common/static_file_storage/Cargo.toml +++ b/common/static_file_storage/Cargo.toml @@ -9,6 +9,8 @@ snap = { workspace = true } [dev-dependencies] criterion = { workspace = true } +rand = { workspace = true } +rand_xorshift = { workspace = true } tempfile = { workspace = true } [[bench]] diff --git a/common/static_file_storage/benches/large_writes.rs b/common/static_file_storage/benches/large_writes.rs index 0141f0cd641..5857373f894 100644 --- a/common/static_file_storage/benches/large_writes.rs +++ b/common/static_file_storage/benches/large_writes.rs @@ -10,6 +10,8 @@ //! Run with: `cargo bench -p static_file_storage --bench large_writes`. use criterion::{BatchSize, Criterion, Throughput, criterion_group, criterion_main}; +use rand::{RngCore, SeedableRng}; +use rand_xorshift::XorShiftRng; use static_file_storage::{Config, StaticFile}; use std::hint::black_box; use std::path::PathBuf; @@ -30,17 +32,8 @@ const SNAPPY: Config = Config { }; fn payload(seed: u64, len: usize) -> Vec { - // xorshift fill — fast, deterministic, incompressible enough that snappy - // can't shortcut (we care about the path it actually takes in production - // where HDiff outputs are already entropy-dense). let mut buf = vec![0u8; len]; - let mut s = seed.wrapping_add(0x9E37_79B9_7F4A_7C15); - for chunk in buf.chunks_mut(8) { - s ^= s << 13; - s ^= s >> 7; - s ^= s << 17; - chunk.copy_from_slice(&s.to_le_bytes()[..chunk.len()]); - } + XorShiftRng::seed_from_u64(seed).fill_bytes(&mut buf); buf } From 69d76265d3642d3df3aca1ff75e1ad7cade0a4d1 Mon Sep 17 00:00:00 2001 From: dapplion <35266934+dapplion@users.noreply.github.com> Date: Sun, 28 Jun 2026 14:34:46 +0200 Subject: [PATCH 3/7] static_file_storage: simplify batch grouping --- Cargo.lock | 1 - common/static_file_storage/Cargo.toml | 1 - .../benches/large_writes.rs | 147 ++--- common/static_file_storage/src/lib.rs | 620 ++++++++---------- 4 files changed, 344 insertions(+), 425 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 5abe4fc3119..0133e2397eb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8658,7 +8658,6 @@ dependencies = [ "parking_lot", "rand 0.9.2", "rand_xorshift 0.4.0", - "snap", "tempfile", ] diff --git a/common/static_file_storage/Cargo.toml b/common/static_file_storage/Cargo.toml index fe8d7d69c1a..d55ccc10fe2 100644 --- a/common/static_file_storage/Cargo.toml +++ b/common/static_file_storage/Cargo.toml @@ -5,7 +5,6 @@ edition = { workspace = true } [dependencies] parking_lot = { workspace = true } -snap = { workspace = true } [dev-dependencies] criterion = { workspace = true } diff --git a/common/static_file_storage/benches/large_writes.rs b/common/static_file_storage/benches/large_writes.rs index 5857373f894..326ced206e2 100644 --- a/common/static_file_storage/benches/large_writes.rs +++ b/common/static_file_storage/benches/large_writes.rs @@ -1,8 +1,8 @@ //! Benchmarks for the write path under ERA-load-shaped record sizes. //! //! The targeted profile is mainnet ERA import: ~10–30 MB state-snapshot and -//! state-diff records into columns where `compression: false` (the production -//! setting for those columns, since HDiff is already compressed internally). +//! state-diff records. Callers are expected to pre-compress payloads if they +//! want compression; this benchmark measures only static storage write cost. //! //! Each iteration fixtures a fresh temp directory so the bench measures //! steady-state put cost, not first-open or healing overhead. @@ -12,22 +12,16 @@ use criterion::{BatchSize, Criterion, Throughput, criterion_group, criterion_main}; use rand::{RngCore, SeedableRng}; use rand_xorshift::XorShiftRng; -use static_file_storage::{Config, StaticFile}; +use static_file_storage::{Config, SLOTS_PER_FILE, StaticFile}; use std::hint::black_box; use std::path::PathBuf; const RECORD_30MB: usize = 30 * 1024 * 1024; const RECORD_4MB: usize = 4 * 1024 * 1024; +const RECORD_1KB: usize = 1024; -const NO_COMPRESSION: Config = Config { +const STORAGE_CONFIG: Config = Config { record_type: [0x04, 0x00], - compression: false, - max_value_bytes: 1024 * 1024 * 1024, -}; - -const SNAPPY: Config = Config { - record_type: [0x04, 0x00], - compression: true, max_value_bytes: 1024 * 1024 * 1024, }; @@ -43,16 +37,32 @@ fn fresh_dir() -> (tempfile::TempDir, PathBuf) { (dir, path) } +fn borrowed_items(values: &[(u64, Vec)]) -> Vec<(u64, &[u8])> { + values + .iter() + .map(|(slot, value)| (*slot, value.as_slice())) + .collect() +} + +fn batch_values(start_slot: u64, count: u64, record_size: usize) -> Vec<(u64, Vec)> { + (0..count) + .map(|i| { + let slot = start_slot + i; + (slot, payload(slot, record_size)) + }) + .collect() +} + fn bench_put_single_30mb(c: &mut Criterion) { let value = payload(1, RECORD_30MB); - let mut group = c.benchmark_group("put_single_30mb_no_compression"); + let mut group = c.benchmark_group("put_single_30mb"); group.sample_size(20); group.throughput(Throughput::Bytes(RECORD_30MB as u64)); group.bench_function("put", |b| { b.iter_batched( || { let (dir, path) = fresh_dir(); - let sf = StaticFile::open(path, NO_COMPRESSION).expect("open"); + let sf = StaticFile::open(path, STORAGE_CONFIG).expect("open"); (dir, sf) }, |(_dir, sf)| { @@ -64,84 +74,39 @@ fn bench_put_single_30mb(c: &mut Criterion) { group.finish(); } -fn bench_put_batch_eight_30mb_one_file(c: &mut Criterion) { - let values: Vec<(u64, Vec)> = (0..8).map(|i| (i, payload(i, RECORD_30MB))).collect(); - let total_bytes = (RECORD_30MB * 8) as u64; - let mut group = c.benchmark_group("put_batch_eight_30mb_in_one_file_no_compression"); - group.sample_size(20); - group.throughput(Throughput::Bytes(total_bytes)); - group.bench_function("put_batch", |b| { - b.iter_batched( - || { - let (dir, path) = fresh_dir(); - let sf = StaticFile::open(path, NO_COMPRESSION).expect("open"); - (dir, sf, values.clone()) - }, - |(_dir, sf, items)| { - sf.put_batch(items).expect("put_batch"); - }, - BatchSize::PerIteration, - ); - }); - group.finish(); -} - -fn bench_put_batch_eight_30mb_snappy(c: &mut Criterion) { - // Same shape as the no-compression bench but with snappy on, to quantify - // the cost we explicitly avoid by setting `compression: false` for the - // state-snapshot / state-diff columns. - let values: Vec<(u64, Vec)> = (0..8).map(|i| (i, payload(i, RECORD_30MB))).collect(); - let total_bytes = (RECORD_30MB * 8) as u64; - let mut group = c.benchmark_group("put_batch_eight_30mb_in_one_file_snappy"); - group.sample_size(10); - group.throughput(Throughput::Bytes(total_bytes)); - group.bench_function("put_batch", |b| { - b.iter_batched( - || { - let (dir, path) = fresh_dir(); - let sf = StaticFile::open(path, SNAPPY).expect("open"); - (dir, sf, values.clone()) - }, - |(_dir, sf, items)| { - sf.put_batch(items).expect("put_batch"); - }, - BatchSize::PerIteration, - ); - }); - group.finish(); -} - -fn bench_put_batch_32_records_4mb_cross_file(c: &mut Criterion) { - // 32 × 4 MB records spanning a file boundary at slot 8192. Catches - // anything quadratic in the per-group rewrite path. - let values: Vec<(u64, Vec)> = (8176u64..8208) - .map(|i| (i, payload(i, RECORD_4MB))) - .collect(); - let total_bytes = (RECORD_4MB * values.len()) as u64; - let mut group = c.benchmark_group("put_batch_32_records_4mb_cross_file_no_compression"); - group.sample_size(20); - group.throughput(Throughput::Bytes(total_bytes)); - group.bench_function("put_batch", |b| { - b.iter_batched( - || { - let (dir, path) = fresh_dir(); - let sf = StaticFile::open(path, NO_COMPRESSION).expect("open"); - (dir, sf, values.clone()) - }, - |(_dir, sf, items)| { - sf.put_batch(items).expect("put_batch"); - }, - BatchSize::PerIteration, - ); - }); - group.finish(); +fn bench_put_batch(c: &mut Criterion) { + for (name, start_slot, count, record_size, sample_size) in [ + ("eight_30mb_one_file", 0, 8, RECORD_30MB, 20), + ( + "32_records_4mb_cross_file", + SLOTS_PER_FILE - 16, + 32, + RECORD_4MB, + 20, + ), + ("two_full_files_1kb", 0, SLOTS_PER_FILE * 2, RECORD_1KB, 10), + ] { + let values = batch_values(start_slot, count, record_size); + let total_bytes = count * record_size as u64; + let mut group = c.benchmark_group(format!("put_batch_{name}")); + group.sample_size(sample_size); + group.throughput(Throughput::Bytes(total_bytes)); + group.bench_function("put_batch", |b| { + b.iter_batched( + || { + let (dir, path) = fresh_dir(); + let sf = StaticFile::open(path, STORAGE_CONFIG).expect("open"); + (dir, sf, values.clone()) + }, + |(_dir, sf, items)| { + sf.put_batch(borrowed_items(&items)).expect("put_batch"); + }, + BatchSize::PerIteration, + ); + }); + group.finish(); + } } -criterion_group!( - benches, - bench_put_single_30mb, - bench_put_batch_eight_30mb_one_file, - bench_put_batch_eight_30mb_snappy, - bench_put_batch_32_records_4mb_cross_file, -); +criterion_group!(benches, bench_put_single_30mb, bench_put_batch,); criterion_main!(benches); diff --git a/common/static_file_storage/src/lib.rs b/common/static_file_storage/src/lib.rs index f2cf28c82c8..d615f3fd75e 100644 --- a/common/static_file_storage/src/lib.rs +++ b/common/static_file_storage/src/lib.rs @@ -12,11 +12,10 @@ //! # File format //! //! Data file: e2store version record (`65 32 00 00 00 00 00 00`), then records -//! appended as `type[2] | length[4 LE] | reserved[2]=0 | payload` (snappy- -//! framed if `Config::compression`). +//! appended as `type[2] | length[4 LE] | reserved[2]=0 | payload`. //! -//! `column.conf`: `b"LHSTBLK2" | highest_slot u64 LE (u64::MAX = empty) | -//! current_data_len u64 LE | record_type[2] | compression u8 | reserved | max_value_bytes u64 LE`. +//! `column.conf`: `b"LHSTBLK3" | highest_slot u64 LE (u64::MAX = empty) | +//! current_data_len u64 LE | record_type[2] | max_value_bytes u64 LE`. //! Atomic update: write `.tmp`, fsync, rename, fsync dir. //! //! # Put contract @@ -29,10 +28,10 @@ //! # Recovery on open //! //! Data file truncated to `current_data_len`; `.off` entries beyond -//! `highest_slot` cleared. The `column.conf` rename is the commit point. +//! `highest_slot` cleared; files beyond the committed file removed. The +//! `column.conf` rename is the commit point. use parking_lot::Mutex; -use snap::{read::FrameDecoder, write::FrameEncoder}; use std::{ fmt, fs::{self, File, OpenOptions}, @@ -41,48 +40,59 @@ use std::{ }; pub const SLOTS_PER_FILE: u64 = 8192; +type SlotGroup<'a> = (u64, Vec<(u64, &'a [u8])>); + const OFFSET_SIZE: u64 = 8; const OFFSET_FILE_LEN: u64 = SLOTS_PER_FILE * OFFSET_SIZE; const CONFIG_FILE: &str = "column.conf"; const CONFIG_TMP_FILE: &str = "column.conf.tmp"; const DATA_FILE_PREFIX: &str = "data_"; -const CONFIG_MAGIC: &[u8; 8] = b"LHSTBLK2"; -const CONFIG_LEN: usize = 36; +const CONFIG_MAGIC: &[u8; 8] = b"LHSTBLK3"; +const CONFIG_LEN: usize = 34; +const CONFIG_DATA_START: usize = 24; +const CONFIG_DATA_LEN: usize = 10; /// Empty-store sentinel for `highest_written_slot` in the per-file config. const EMPTY_SLOT: u64 = u64::MAX; /// e2store version record, written once at the start of each data file. const VERSION_RECORD: [u8; 8] = [0x65, 0x32, 0, 0, 0, 0, 0, 0]; +/// e2store record header reserved bytes. +const RECORD_RESERVED: [u8; 2] = [0, 0]; -const COMPRESSION_NONE: u8 = 0; -const COMPRESSION_SNAPPY: u8 = 1; - -/// On-disk format settings for one `StaticFile`. On first creation the build's -/// values are persisted; on re-open the persisted values win, with -/// `max_value_bytes` ratcheted upward if the build allows larger records. -#[derive(Debug, Clone, Copy)] +/// On-disk format settings for one `StaticFile`. Persisted on first creation; +/// on re-open the on-disk values must match what the caller passes in. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct Config { /// e2store record type tag. pub record_type: [u8; 2], - /// Whether values are snappy-framed before write. - pub compression: bool, - /// Upper bound on a single decoded record's size in bytes. + /// Upper bound on a single stored record's size in bytes. pub max_value_bytes: u64, } -struct ConfigOnDisk { - highest_written_slot: Option, - current_data_len: u64, - record_type: [u8; 2], - compression: bool, - max_value_bytes: u64, -} +impl Config { + fn serialize(&self) -> [u8; CONFIG_DATA_LEN] { + let mut bytes = [0u8; CONFIG_DATA_LEN]; + bytes[0..2].copy_from_slice(&self.record_type); + bytes[2..10].copy_from_slice(&self.max_value_bytes.to_le_bytes()); + bytes + } -pub type Result = std::result::Result; + fn deserialize(bytes: &[u8]) -> Result { + if bytes.len() != CONFIG_DATA_LEN { + return Err(Error::Invalid("invalid config data length".into())); + } + let record_type = [bytes[0], bytes[1]]; + let max_value_bytes = + u64::from_le_bytes(bytes[2..10].try_into().expect("slice length checked")); + Ok(Self { + record_type, + max_value_bytes, + }) + } +} #[derive(Debug)] pub enum Error { Io(io::Error), - Compression(io::Error), Invalid(String), } @@ -90,7 +100,6 @@ impl fmt::Display for Error { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { Self::Io(e) => write!(f, "static file io error: {e}"), - Self::Compression(e) => write!(f, "static file compression error: {e}"), Self::Invalid(message) => write!(f, "static file invalid data: {message}"), } } @@ -99,7 +108,7 @@ impl fmt::Display for Error { impl std::error::Error for Error { fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { match self { - Self::Io(e) | Self::Compression(e) => Some(e), + Self::Io(e) => Some(e), Self::Invalid(_) => None, } } @@ -117,75 +126,50 @@ impl From for Error { pub struct StaticFile { root_dir: PathBuf, config: Config, + /// `None` = empty store, `Some(s)` = highest committed slot. + /// Persisted as `u64::MAX` (`EMPTY_SLOT`) when `None`. highest_written_slot: Mutex>, } impl StaticFile { /// Open or create a `StaticFile` rooted at `root_dir`. On first creation, - /// `defaults` is persisted; on re-open, the persisted values win for - /// `record_type` and `compression`, and `max_value_bytes` is ratcheted up - /// to `max(on_disk, defaults)`. - pub fn open(root_dir: PathBuf, defaults: Config) -> Result { + /// `config` is persisted. On re-open, the persisted config must equal + /// `config` — any mismatch is a startup error. + pub fn open(root_dir: PathBuf, config: Config) -> Result { fs::create_dir_all(&root_dir)?; - - // First-open: persist current-build defaults. Re-open: persisted - // settings win over `defaults`, which preserves on-disk readability - // even if the build's defaults change later. let config_path = root_dir.join(CONFIG_FILE); let tmp_path = root_dir.join(CONFIG_TMP_FILE); - if !config_path.exists() { - atomic_write_config(&config_path, &tmp_path, &root_dir, None, 0, &defaults)?; - } - let on_disk = read_config(&config_path)?; - // record_type and compression are sticky — they're load-bearing for - // reading old records, so on-disk wins over build-time defaults. - // max_value_bytes is a soft bound used to cap accepted record sizes; - // ratchet it up if the build's default is larger so a newer build - // can write bigger records than an older one persisted, then - // re-persist immediately so future opens see the new bound. - let max_value_bytes = on_disk.max_value_bytes.max(defaults.max_value_bytes); - let config = Config { - record_type: on_disk.record_type, - compression: on_disk.compression, - max_value_bytes, + let (highest, data_len) = if config_path.exists() { + let (highest, data_len, on_disk) = read_config(&config_path)?; + if on_disk != config { + return Err(Error::Invalid(format!( + "config mismatch: on-disk {on_disk:?}, build {config:?}" + ))); + } + (highest, data_len) + } else { + atomic_write_config(&config_path, &tmp_path, &root_dir, None, 0, &config)?; + (None, 0) }; - if max_value_bytes != on_disk.max_value_bytes { - atomic_write_config( - &config_path, - &tmp_path, - &root_dir, - on_disk.highest_written_slot, - on_disk.current_data_len, - &config, - )?; - } let handle = Self { root_dir, config, - highest_written_slot: Mutex::new(None), + highest_written_slot: Mutex::new(highest), }; - - if let Some(slot) = on_disk.highest_written_slot { - handle.heal_current_file(slot, on_disk.current_data_len)?; - } - *handle.highest_written_slot.lock() = on_disk.highest_written_slot; - + handle.heal_to_marker(highest, data_len)?; Ok(handle) } /// Slot of the most recently committed record, if any. - pub fn highest_written_slot(&self) -> Option { + fn highest_written_slot(&self) -> Option { *self.highest_written_slot.lock() } /// Read the record at `slot`, if present. - pub fn get(&self, slot: u64) -> Result>> { - let Some(highest_written_slot) = *self.highest_written_slot.lock() else { - return Ok(None); - }; - if slot > highest_written_slot { + pub fn get(&self, slot: u64) -> Result>, Error> { + if self.highest_written_slot().is_none_or(|h| slot > h) { return Ok(None); } self.read_record(slot) @@ -193,11 +177,8 @@ impl StaticFile { /// `true` if a record exists at `slot`. Cheaper than `get` — only the /// `.off` sidecar is consulted; the data file is not read. - pub fn contains(&self, slot: u64) -> Result { - let Some(highest_written_slot) = *self.highest_written_slot.lock() else { - return Ok(false); - }; - if slot > highest_written_slot { + pub fn contains(&self, slot: u64) -> Result { + if self.highest_written_slot().is_none_or(|h| slot > h) { return Ok(false); } Ok(self.read_offset(file_id(slot), slot)? != 0) @@ -208,37 +189,94 @@ impl StaticFile { /// (re-puts at any committed slot are idempotent so migration retries /// after a mid-loop crash are safe). A previously-skipped slot (offset /// zero) cannot be filled — that would break the append-only data file. - pub fn put(&self, slot: u64, bytes: &[u8]) -> Result<()> { + pub fn put(&self, slot: u64, bytes: &[u8]) -> Result<(), Error> { + self.put_batch(vec![(slot, bytes)]) + } + + /// Append `items` with one fsync per file (data + offset), not per slot. + /// Whole batch is durable on return — the same caller-visible contract as + /// `put` — but with O(1) syncs per underlying file instead of O(n) per + /// item. Items must be strictly ascending. Already-committed prefix items + /// are treated as idempotent re-puts if their bytes match, then skipped. + pub fn put_batch(&self, items: Vec<(u64, &[u8])>) -> Result<(), Error> { + let Some((last_slot, _)) = items.last() else { + return Ok(()); + }; + for w in items.windows(2) { + if w[1].0 <= w[0].0 { + return Err(Error::Invalid( + "put_batch slots must be strictly ascending".into(), + )); + } + } + let mut highest_written_slot = self.highest_written_slot.lock(); - if let Some(highest) = *highest_written_slot - && slot <= highest - { - let existing = self.read_record(slot)?.ok_or_else(|| { + let start = self.skip_committed_prefix(&items, *highest_written_slot)?; + if start == items.len() { + return Ok(()); + } + + // Group remaining items by file_id, write each group with one + // data fsync + one offset fsync. Single config commit at end of batch. + let mut last_data_len: u64 = 0; + for (target_file_id, group) in group_by_file_id(&items[start..]) { + match self.write_group(target_file_id, &group) { + Ok(data_len) => last_data_len = data_len, + Err(e) => { + if let Ok(committed_highest) = self.heal_from_config() { + *highest_written_slot = committed_highest; + } + return Err(e); + } + } + } + + if let Err(e) = self.write_config(Some(*last_slot), last_data_len) { + if let Ok(committed_highest) = self.heal_from_config() { + *highest_written_slot = committed_highest; + } + return Err(e); + } + *highest_written_slot = Some(*last_slot); + Ok(()) + } + + /// Validate already-committed retry items and return the first new item. + /// This lets migration retries overlap with the committed prefix without + /// rewriting offsets or filling skipped slots. + fn skip_committed_prefix( + &self, + items: &[(u64, &[u8])], + highest_written_slot: Option, + ) -> Result { + let Some(highest) = highest_written_slot else { + return Ok(0); + }; + + let mut start = 0; + while start < items.len() && items[start].0 <= highest { + let (slot, value) = &items[start]; + let existing = self.read_record(*slot)?.ok_or_else(|| { Error::Invalid(format!( "re-put at slot {slot} <= highest {highest} but no record exists; \ cannot fill a previously-skipped slot" )) })?; - if existing == bytes { - return Ok(()); + if existing.as_slice() != *value { + return Err(Error::Invalid(format!( + "re-put at slot {slot} with mismatched value" + ))); } - return Err(Error::Invalid(format!( - "re-put at slot {slot} with mismatched value" - ))); + start += 1; } + Ok(start) + } - let payload = if self.config.compression { - compress_record(bytes)? - } else { - bytes.to_vec() - }; - let payload_len = - u32::try_from(payload.len()).map_err(|_| Error::Invalid("record too large".into()))?; - - let target_file_id = file_id(slot); - // Discard an uncommitted next-file tail after a crash. - let reset_file = (*highest_written_slot).map(file_id) != Some(target_file_id); - let off_pos = offset_position(slot); + /// Write `group` (all in `target_file_id`) to disk: append records, fsync + /// the data file, write all offsets, fsync the offset file. Returns the + /// committed data file length. Does NOT update `highest_written_slot` or + /// `column.conf` — the caller commits via `write_config` after this. + fn write_group(&self, target_file_id: u64, group: &[(u64, &[u8])]) -> Result { let data_path = self.data_path(target_file_id); let off_path = self.offset_path(target_file_id); @@ -247,23 +285,33 @@ impl StaticFile { .append(true) .create(true) .open(&data_path)?; - if reset_file { - data_file.set_len(0)?; - } - if data_file.metadata()?.len() == 0 { data_file.write_all(&VERSION_RECORD)?; } - let offset = data_file.seek(SeekFrom::End(0))?; - write_record( - &mut data_file, - self.config.record_type, - payload_len, - &payload, - )?; + let mut offsets: Vec<(u64, u64)> = Vec::with_capacity(group.len()); + { + // BufWriter coalesces the 8-byte record headers + payloads into + // larger syscalls; cheap for one record, load-bearing for batches. + let mut writer = std::io::BufWriter::with_capacity(1 << 20, &mut data_file); + let mut cursor = writer.get_ref().metadata()?.len(); + for &(slot, value) in group { + if (value.len() as u64) > self.config.max_value_bytes { + return Err(Error::Invalid("record exceeds size limit".into())); + } + let payload_len = u32::try_from(value.len()) + .map_err(|_| Error::Invalid("record too large".into()))?; + offsets.push((slot, cursor)); + writer.write_all(&self.config.record_type)?; + writer.write_all(&payload_len.to_le_bytes())?; + writer.write_all(&RECORD_RESERVED)?; + writer.write_all(value)?; + cursor += 8 + value.len() as u64; + } + writer.flush()?; + } let data_len = data_file.seek(SeekFrom::End(0))?; - // Data and offset files must hit disk before the config commit marker. + // Data must hit disk before offsets, both before config commit. data_file.sync_all()?; let mut off_file = OpenOptions::new() @@ -272,165 +320,21 @@ impl StaticFile { .create(true) .truncate(false) .open(&off_path)?; - if reset_file { - off_file.set_len(0)?; - } if off_file.metadata()?.len() < OFFSET_FILE_LEN { off_file.set_len(OFFSET_FILE_LEN)?; } - off_file.seek(SeekFrom::Start(off_pos))?; - off_file.write_all(&offset.to_le_bytes())?; - off_file.sync_all()?; - - // Atomic config update is the commit point. - self.write_config(Some(slot), data_len)?; - *highest_written_slot = Some(slot); - - Ok(()) - } - - /// Append `items` with one fsync per file (data + offset), not per slot. - /// Whole batch is durable on return — the same caller-visible contract as - /// `put` — but with O(1) syncs per underlying file instead of O(n) per - /// item. - /// - /// Walks `items` once, grouping them by `file_id`. For each group, opens - /// the data file and offset file once, appends every record's bytes - /// (collecting `(slot, offset)` pairs in memory), writes the offset - /// table, fsyncs both files, then commits via `write_config`. Idempotent - /// re-put of `items[0]` at `highest_written_slot` is honored as in `put`. - pub fn put_batch(&self, items: Vec<(u64, Vec)>) -> Result<()> { - if items.is_empty() { - return Ok(()); - } - - // Validate ascending order up front (cheap, catches caller bugs). - for w in items.windows(2) { - if w[1].0 <= w[0].0 { - return Err(Error::Invalid( - "put_batch slots must be strictly ascending".into(), - )); - } - } - - let mut highest_written_slot = self.highest_written_slot.lock(); - let mut iter = items.into_iter().peekable(); - - // Idempotent re-put: if the first item is exactly highest_written_slot - // with matching bytes, drop it from the batch. - if let (Some(highest), Some((first_slot, _))) = (*highest_written_slot, iter.peek()) { - if *first_slot < highest { - return Err(Error::Invalid( - "put_batch out of order vs highest_written_slot".into(), - )); - } - if *first_slot == highest { - let (slot, value) = iter.next().expect("peeked"); - let existing = self - .read_record(slot)? - .ok_or_else(|| Error::Invalid("missing record at highest slot".into()))?; - if existing != value { - return Err(Error::Invalid("re-put with mismatched value".into())); - } - } - } - - // Group remaining items by file_id, write each group with a single - // fsync per file. - let mut last_slot: Option = None; - let mut last_data_len: u64 = 0; - while iter.peek().is_some() { - let target_file_id = file_id(iter.peek().expect("peeked").0); - let mut group: Vec<(u64, Vec)> = Vec::new(); - while let Some(&(slot, _)) = iter.peek() { - if file_id(slot) != target_file_id { - break; - } - group.push(iter.next().expect("peeked")); - } - - let reset_file = (*highest_written_slot).map(file_id) != Some(target_file_id); - let data_path = self.data_path(target_file_id); - let off_path = self.offset_path(target_file_id); - - // Data file: append all records, then fsync once. - let mut data_file = OpenOptions::new() - .read(true) - .append(true) - .create(true) - .open(&data_path)?; - if reset_file { - data_file.set_len(0)?; - } - if data_file.metadata()?.len() == 0 { - data_file.write_all(&VERSION_RECORD)?; - } - // BufWriter coalesces the small-record header writes (8 bytes) and - // the small payloads into larger syscalls. - let mut offsets: Vec<(u64, u64)> = Vec::with_capacity(group.len()); - { - let mut writer = std::io::BufWriter::with_capacity(1 << 20, &mut data_file); - let mut cursor = writer.get_ref().metadata()?.len(); - for (slot, value) in &group { - let payload: std::borrow::Cow<'_, [u8]> = if self.config.compression { - compress_record(value)?.into() - } else { - value.as_slice().into() - }; - let payload_len = u32::try_from(payload.len()) - .map_err(|_| Error::Invalid("record too large".into()))?; - offsets.push((*slot, cursor)); - // Inline `write_record` to avoid the `&mut File` -> BufWriter mismatch. - writer.write_all(&self.config.record_type)?; - writer.write_all(&payload_len.to_le_bytes())?; - writer.write_all(&0u16.to_le_bytes())?; - writer.write_all(&payload)?; - cursor += 8 + payload.len() as u64; - } - writer.flush()?; - } - let data_len = data_file.seek(SeekFrom::End(0))?; - data_file.sync_all()?; - - // Offset file: open, ensure full size, write all offsets in seek+write - // pairs (8 bytes each), then fsync once. - let mut off_file = OpenOptions::new() - .read(true) - .write(true) - .create(true) - .truncate(false) - .open(&off_path)?; - if reset_file { - off_file.set_len(0)?; - } - if off_file.metadata()?.len() < OFFSET_FILE_LEN { - off_file.set_len(OFFSET_FILE_LEN)?; - } - for (slot, offset) in &offsets { - off_file.seek(SeekFrom::Start(offset_position(*slot)))?; - off_file.write_all(&offset.to_le_bytes())?; - } - off_file.sync_all()?; - - // Track final slot/data_len for the single config commit at end of batch. - if let Some((s, _)) = group.last() { - last_slot = Some(*s); - last_data_len = data_len; - } - *highest_written_slot = last_slot; - } - - // Single atomic config commit covering the whole batch. - if let Some(s) = last_slot { - self.write_config(Some(s), last_data_len)?; + for (slot, offset) in &offsets { + off_file.seek(SeekFrom::Start(offset_position(*slot)))?; + off_file.write_all(&offset.to_le_bytes())?; } + off_file.sync_all()?; - Ok(()) + Ok(data_len) } /// Read a record at `slot` without consulting the writer mutex. Used by /// callers that already hold the lock (`put` for the idempotency check). - fn read_record(&self, slot: u64) -> Result>> { + fn read_record(&self, slot: u64) -> Result>, Error> { let file_id = file_id(slot); let offset = self.read_offset(file_id, slot)?; if offset == 0 { @@ -443,25 +347,51 @@ impl StaticFile { let mut header = [0; 8]; data_file.read_exact(&mut header)?; - if header[0..2] != self.config.record_type || header[6..8] != [0, 0] { + if header[0..2] != self.config.record_type || header[6..8] != RECORD_RESERVED { return Err(Error::Invalid("invalid record header".into())); } - let len = u32::from_le_bytes([header[2], header[3], header[4], header[5]]) as usize; + let len = u32::from_le_bytes([header[2], header[3], header[4], header[5]]); + if u64::from(len) > self.config.max_value_bytes { + return Err(Error::Invalid("record exceeds size limit".into())); + } + + let len = len as usize; let mut payload = vec![0; len]; data_file.read_exact(&mut payload)?; + Ok(Some(payload)) + } - if self.config.compression { - decompress_record(&payload, self.config.max_value_bytes).map(Some) + /// Restore files to the committed marker and remove future crash leftovers. + fn heal_to_marker( + &self, + highest_written_slot: Option, + current_data_len: u64, + ) -> Result<(), Error> { + if let Some(slot) = highest_written_slot { + self.heal_current_file(slot, current_data_len)?; + self.remove_uncommitted_files(Some(file_id(slot))) } else { - if (payload.len() as u64) > self.config.max_value_bytes { - return Err(Error::Invalid("record exceeds size limit".into())); - } - Ok(Some(payload)) + self.remove_uncommitted_files(None) } } - fn heal_current_file(&self, slot: u64, current_data_len: u64) -> Result<()> { + /// Re-read `column.conf` and restore files to whatever marker is durable. + fn heal_from_config(&self) -> Result, Error> { + let (highest, data_len, on_disk) = read_config(&self.config_path())?; + if on_disk != self.config { + return Err(Error::Invalid(format!( + "config mismatch: on-disk {on_disk:?}, build {:?}", + self.config + ))); + } + self.heal_to_marker(highest, data_len)?; + Ok(highest) + } + + /// Truncate the committed file back to the config marker after a crash. + /// Data beyond `current_data_len` and offsets beyond `slot` are uncommitted. + fn heal_current_file(&self, slot: u64, current_data_len: u64) -> Result<(), Error> { let file_id = file_id(slot); let data_path = self.data_path(file_id); let data_file = OpenOptions::new().read(true).write(true).open(&data_path)?; @@ -501,7 +431,43 @@ impl StaticFile { Ok(()) } - fn write_config(&self, highest_written_slot: Option, current_data_len: u64) -> Result<()> { + /// Remove data/offset files newer than the committed file id. + /// `None` means the store is empty, so every data/offset file is removed. + fn remove_uncommitted_files(&self, max_file_id: Option) -> Result<(), Error> { + let mut removed = false; + for entry in fs::read_dir(&self.root_dir)? { + let entry = entry?; + let file_name = entry.file_name(); + let Some(file_name) = file_name.to_str() else { + continue; + }; + let Some(file_id) = static_file_id(file_name) else { + continue; + }; + if max_file_id.is_none_or(|max| file_id > max) { + let file_type = entry.file_type()?; + if !file_type.is_file() { + return Err(Error::Invalid(format!( + "static data path is not a file: {}", + entry.path().display() + ))); + } + fs::remove_file(entry.path())?; + removed = true; + } + } + if removed { + sync_dir(&self.root_dir)?; + } + Ok(()) + } + + /// Atomically publish the commit marker after data and offsets are durable. + fn write_config( + &self, + highest_written_slot: Option, + current_data_len: u64, + ) -> Result<(), Error> { atomic_write_config( &self.config_path(), &self.root_dir.join(CONFIG_TMP_FILE), @@ -512,9 +478,13 @@ impl StaticFile { ) } - fn read_offset(&self, file_id: u64, slot: u64) -> Result { + fn read_offset(&self, file_id: u64, slot: u64) -> Result { let off_path = self.offset_path(file_id); - let mut off_file = File::open(&off_path)?; + let mut off_file = match File::open(&off_path) { + Ok(file) => file, + Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(0), + Err(e) => return Err(e.into()), + }; let mut bytes = [0; 8]; off_file.seek(SeekFrom::Start(offset_position(slot)))?; off_file.read_exact(&mut bytes)?; @@ -536,7 +506,8 @@ impl StaticFile { } } -fn read_config(path: &Path) -> Result { +/// Returns (highest_written_slot, current_data_len, on-disk config). +fn read_config(path: &Path) -> Result<(Option, u64, Config), Error> { let bytes = fs::read(path)?; if bytes.len() != CONFIG_LEN || &bytes[0..8] != CONFIG_MAGIC { return Err(Error::Invalid("invalid config".into())); @@ -544,25 +515,16 @@ fn read_config(path: &Path) -> Result { let highest = u64::from_le_bytes(bytes[8..16].try_into().expect("slice length checked")); let current_data_len = u64::from_le_bytes(bytes[16..24].try_into().expect("slice length checked")); - let record_type = [bytes[24], bytes[25]]; - let compression = match bytes[26] { - COMPRESSION_NONE => false, - COMPRESSION_SNAPPY => true, - other => { - return Err(Error::Invalid(format!("unknown compression flag {other}"))); - } - }; - let max_value_bytes = - u64::from_le_bytes(bytes[28..36].try_into().expect("slice length checked")); - Ok(ConfigOnDisk { - highest_written_slot: (highest != EMPTY_SLOT).then_some(highest), + let config = Config::deserialize(&bytes[CONFIG_DATA_START..])?; + Ok(( + (highest != EMPTY_SLOT).then_some(highest), current_data_len, - record_type, - compression, - max_value_bytes, - }) + config, + )) } +/// Write `column.conf.tmp`, fsync it, rename over `column.conf`, then fsync +/// the directory so the rename is durable. fn atomic_write_config( config_path: &Path, tmp_path: &Path, @@ -570,19 +532,12 @@ fn atomic_write_config( highest_written_slot: Option, current_data_len: u64, config: &Config, -) -> Result<()> { +) -> Result<(), Error> { let mut bytes = [0u8; CONFIG_LEN]; bytes[0..8].copy_from_slice(CONFIG_MAGIC); bytes[8..16].copy_from_slice(&highest_written_slot.unwrap_or(EMPTY_SLOT).to_le_bytes()); bytes[16..24].copy_from_slice(¤t_data_len.to_le_bytes()); - bytes[24..26].copy_from_slice(&config.record_type); - bytes[26] = if config.compression { - COMPRESSION_SNAPPY - } else { - COMPRESSION_NONE - }; - bytes[27] = 0; - bytes[28..36].copy_from_slice(&config.max_value_bytes.to_le_bytes()); + bytes[CONFIG_DATA_START..].copy_from_slice(&config.serialize()); { let mut tmp = File::create(tmp_path)?; @@ -602,42 +557,43 @@ fn offset_position(slot: u64) -> u64 { (slot % SLOTS_PER_FILE) * OFFSET_SIZE } -fn compress_record(bytes: &[u8]) -> Result> { - let mut encoder = FrameEncoder::new(Vec::new()); - encoder.write_all(bytes).map_err(Error::Compression)?; - encoder.flush().map_err(Error::Compression)?; - Ok(encoder.get_ref().clone()) +fn static_file_id(file_name: &str) -> Option { + let suffix = file_name.strip_prefix(DATA_FILE_PREFIX)?; + let id = suffix.strip_suffix(".off").unwrap_or(suffix); + if id.is_empty() || !id.bytes().all(|b| b.is_ascii_digit()) { + return None; + } + id.parse().ok() } -fn write_record( - file: &mut File, - record_type: [u8; 2], - payload_len: u32, - payload: &[u8], -) -> Result<()> { - file.write_all(&record_type)?; - file.write_all(&payload_len.to_le_bytes())?; - file.write_all(&0u16.to_le_bytes())?; - file.write_all(payload)?; - Ok(()) -} +/// Convert an ascending slot list into consecutive groups for each data file. +fn group_by_file_id<'a>(items: &[(u64, &'a [u8])]) -> Vec> { + let mut groups = Vec::new(); + let mut current_file_id = None; + let mut current_group = Vec::new(); + + for (slot, value) in items { + let target_file_id = file_id(*slot); + if current_file_id == Some(target_file_id) { + current_group.push((*slot, *value)); + } else { + if let Some(file_id) = current_file_id { + groups.push((file_id, current_group)); + current_group = Vec::new(); + } + current_file_id = Some(target_file_id); + current_group.push((*slot, *value)); + } + } -fn decompress_record(bytes: &[u8], max_value_bytes: u64) -> Result> { - let decoder = FrameDecoder::new(bytes); - let mut limited = decoder.take(max_value_bytes + 1); - let mut decompressed = Vec::new(); - limited - .read_to_end(&mut decompressed) - .map_err(Error::Compression)?; - if decompressed.len() as u64 > max_value_bytes { - return Err(Error::Invalid( - "record exceeds decompressed size limit".into(), - )); + if let Some(file_id) = current_file_id { + groups.push((file_id, current_group)); } - Ok(decompressed) + + groups } -fn sync_dir(path: &Path) -> Result<()> { +fn sync_dir(path: &Path) -> Result<(), Error> { let dir = File::open(path)?; dir.sync_all()?; Ok(()) From a8a0ac1d421ff40d9ab65e8d083e32da80fd67a6 Mon Sep 17 00:00:00 2001 From: dapplion <35266934+dapplion@users.noreply.github.com> Date: Sun, 28 Jun 2026 15:23:17 +0200 Subject: [PATCH 4/7] static_file_storage: precompute batch offsets --- common/static_file_storage/src/lib.rs | 31 ++++++++++++++++++--------- 1 file changed, 21 insertions(+), 10 deletions(-) diff --git a/common/static_file_storage/src/lib.rs b/common/static_file_storage/src/lib.rs index d615f3fd75e..88bbc088621 100644 --- a/common/static_file_storage/src/lib.rs +++ b/common/static_file_storage/src/lib.rs @@ -289,24 +289,35 @@ impl StaticFile { data_file.write_all(&VERSION_RECORD)?; } - let mut offsets: Vec<(u64, u64)> = Vec::with_capacity(group.len()); + let start_offset = data_file.metadata()?.len(); + let record_sizes = group + .iter() + .map(|(_, value)| { + if (value.len() as u64) > self.config.max_value_bytes { + return Err(Error::Invalid("record exceeds size limit".into())); + } + u32::try_from(value.len()).map_err(|_| Error::Invalid("record too large".into())) + }) + .collect::, _>>()?; + let offsets = group + .iter() + .zip(&record_sizes) + .scan(start_offset, |cursor, ((slot, _), payload_len)| { + let offset = *cursor; + *cursor += 8 + u64::from(*payload_len); + Some((*slot, offset)) + }) + .collect::>(); + { // BufWriter coalesces the 8-byte record headers + payloads into // larger syscalls; cheap for one record, load-bearing for batches. let mut writer = std::io::BufWriter::with_capacity(1 << 20, &mut data_file); - let mut cursor = writer.get_ref().metadata()?.len(); - for &(slot, value) in group { - if (value.len() as u64) > self.config.max_value_bytes { - return Err(Error::Invalid("record exceeds size limit".into())); - } - let payload_len = u32::try_from(value.len()) - .map_err(|_| Error::Invalid("record too large".into()))?; - offsets.push((slot, cursor)); + for ((_, value), payload_len) in group.iter().zip(record_sizes) { writer.write_all(&self.config.record_type)?; writer.write_all(&payload_len.to_le_bytes())?; writer.write_all(&RECORD_RESERVED)?; writer.write_all(value)?; - cursor += 8 + value.len() as u64; } writer.flush()?; } From 9a94d4afd91c6b42257851820b48a176f9e33ac8 Mon Sep 17 00:00:00 2001 From: dapplion <35266934+dapplion@users.noreply.github.com> Date: Sun, 28 Jun 2026 15:24:36 +0200 Subject: [PATCH 5/7] static_file_storage: simplify offset collection --- common/static_file_storage/src/lib.rs | 27 +++++++++++---------------- 1 file changed, 11 insertions(+), 16 deletions(-) diff --git a/common/static_file_storage/src/lib.rs b/common/static_file_storage/src/lib.rs index 88bbc088621..71799001fd8 100644 --- a/common/static_file_storage/src/lib.rs +++ b/common/static_file_storage/src/lib.rs @@ -289,23 +289,13 @@ impl StaticFile { data_file.write_all(&VERSION_RECORD)?; } - let start_offset = data_file.metadata()?.len(); - let record_sizes = group - .iter() - .map(|(_, value)| { - if (value.len() as u64) > self.config.max_value_bytes { - return Err(Error::Invalid("record exceeds size limit".into())); - } - u32::try_from(value.len()).map_err(|_| Error::Invalid("record too large".into())) - }) - .collect::, _>>()?; + let mut cursor = data_file.metadata()?.len(); let offsets = group .iter() - .zip(&record_sizes) - .scan(start_offset, |cursor, ((slot, _), payload_len)| { - let offset = *cursor; - *cursor += 8 + u64::from(*payload_len); - Some((*slot, offset)) + .map(|(slot, value)| { + let offset = cursor; + cursor += 8 + value.len() as u64; + (*slot, offset) }) .collect::>(); @@ -313,7 +303,12 @@ impl StaticFile { // BufWriter coalesces the 8-byte record headers + payloads into // larger syscalls; cheap for one record, load-bearing for batches. let mut writer = std::io::BufWriter::with_capacity(1 << 20, &mut data_file); - for ((_, value), payload_len) in group.iter().zip(record_sizes) { + for (_, value) in group { + if (value.len() as u64) > self.config.max_value_bytes { + return Err(Error::Invalid("record exceeds size limit".into())); + } + let payload_len = u32::try_from(value.len()) + .map_err(|_| Error::Invalid("record too large".into()))?; writer.write_all(&self.config.record_type)?; writer.write_all(&payload_len.to_le_bytes())?; writer.write_all(&RECORD_RESERVED)?; From dd20de55c9a52da1977e53103be7fe7f24927561 Mon Sep 17 00:00:00 2001 From: dapplion <35266934+dapplion@users.noreply.github.com> Date: Sun, 28 Jun 2026 15:27:33 +0200 Subject: [PATCH 6/7] Revert "static_file_storage: simplify offset collection" This reverts commit 9a94d4afd91c6b42257851820b48a176f9e33ac8. --- common/static_file_storage/src/lib.rs | 27 ++++++++++++++++----------- 1 file changed, 16 insertions(+), 11 deletions(-) diff --git a/common/static_file_storage/src/lib.rs b/common/static_file_storage/src/lib.rs index 71799001fd8..88bbc088621 100644 --- a/common/static_file_storage/src/lib.rs +++ b/common/static_file_storage/src/lib.rs @@ -289,13 +289,23 @@ impl StaticFile { data_file.write_all(&VERSION_RECORD)?; } - let mut cursor = data_file.metadata()?.len(); + let start_offset = data_file.metadata()?.len(); + let record_sizes = group + .iter() + .map(|(_, value)| { + if (value.len() as u64) > self.config.max_value_bytes { + return Err(Error::Invalid("record exceeds size limit".into())); + } + u32::try_from(value.len()).map_err(|_| Error::Invalid("record too large".into())) + }) + .collect::, _>>()?; let offsets = group .iter() - .map(|(slot, value)| { - let offset = cursor; - cursor += 8 + value.len() as u64; - (*slot, offset) + .zip(&record_sizes) + .scan(start_offset, |cursor, ((slot, _), payload_len)| { + let offset = *cursor; + *cursor += 8 + u64::from(*payload_len); + Some((*slot, offset)) }) .collect::>(); @@ -303,12 +313,7 @@ impl StaticFile { // BufWriter coalesces the 8-byte record headers + payloads into // larger syscalls; cheap for one record, load-bearing for batches. let mut writer = std::io::BufWriter::with_capacity(1 << 20, &mut data_file); - for (_, value) in group { - if (value.len() as u64) > self.config.max_value_bytes { - return Err(Error::Invalid("record exceeds size limit".into())); - } - let payload_len = u32::try_from(value.len()) - .map_err(|_| Error::Invalid("record too large".into()))?; + for ((_, value), payload_len) in group.iter().zip(record_sizes) { writer.write_all(&self.config.record_type)?; writer.write_all(&payload_len.to_le_bytes())?; writer.write_all(&RECORD_RESERVED)?; From 3889325789c3f4a3facf9ba635ad6493296dd5c2 Mon Sep 17 00:00:00 2001 From: dapplion <35266934+dapplion@users.noreply.github.com> Date: Sun, 28 Jun 2026 15:30:21 +0200 Subject: [PATCH 7/7] static_file_storage: add critical behavior tests --- common/static_file_storage/src/lib.rs | 112 +++++++++++++++++++++----- 1 file changed, 91 insertions(+), 21 deletions(-) diff --git a/common/static_file_storage/src/lib.rs b/common/static_file_storage/src/lib.rs index 88bbc088621..aa09d32eb4f 100644 --- a/common/static_file_storage/src/lib.rs +++ b/common/static_file_storage/src/lib.rs @@ -289,35 +289,24 @@ impl StaticFile { data_file.write_all(&VERSION_RECORD)?; } - let start_offset = data_file.metadata()?.len(); - let record_sizes = group - .iter() - .map(|(_, value)| { - if (value.len() as u64) > self.config.max_value_bytes { - return Err(Error::Invalid("record exceeds size limit".into())); - } - u32::try_from(value.len()).map_err(|_| Error::Invalid("record too large".into())) - }) - .collect::, _>>()?; - let offsets = group - .iter() - .zip(&record_sizes) - .scan(start_offset, |cursor, ((slot, _), payload_len)| { - let offset = *cursor; - *cursor += 8 + u64::from(*payload_len); - Some((*slot, offset)) - }) - .collect::>(); - + let mut offsets = Vec::with_capacity(group.len()); { // BufWriter coalesces the 8-byte record headers + payloads into // larger syscalls; cheap for one record, load-bearing for batches. let mut writer = std::io::BufWriter::with_capacity(1 << 20, &mut data_file); - for ((_, value), payload_len) in group.iter().zip(record_sizes) { + let mut cursor = writer.get_ref().metadata()?.len(); + for &(slot, value) in group { + if (value.len() as u64) > self.config.max_value_bytes { + return Err(Error::Invalid("record exceeds size limit".into())); + } + let payload_len = u32::try_from(value.len()) + .map_err(|_| Error::Invalid("record too large".into()))?; + offsets.push((slot, cursor)); writer.write_all(&self.config.record_type)?; writer.write_all(&payload_len.to_le_bytes())?; writer.write_all(&RECORD_RESERVED)?; writer.write_all(value)?; + cursor += 8 + value.len() as u64; } writer.flush()?; } @@ -609,3 +598,84 @@ fn sync_dir(path: &Path) -> Result<(), Error> { dir.sync_all()?; Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + + const CONFIG: Config = Config { + record_type: [4, 0], + max_value_bytes: 1024, + }; + + fn open(dir: &tempfile::TempDir) -> StaticFile { + StaticFile::open(dir.path().to_path_buf(), CONFIG).unwrap() + } + + #[test] + fn round_trip_sparse_slots() { + let dir = tempfile::tempdir().unwrap(); + let store = open(&dir); + + store.put(100, b"a").unwrap(); + store.put(9000, b"b").unwrap(); + + assert_eq!(store.get(100).unwrap(), Some(b"a".to_vec())); + assert_eq!(store.get(101).unwrap(), None); + assert_eq!(store.get(9000).unwrap(), Some(b"b".to_vec())); + assert!(!store.contains(101).unwrap()); + } + + #[test] + fn reopen_preserves_committed_records() { + let dir = tempfile::tempdir().unwrap(); + open(&dir) + .put_batch(vec![(SLOTS_PER_FILE - 1, b"a"), (SLOTS_PER_FILE, b"b")]) + .unwrap(); + + let store = open(&dir); + assert_eq!(store.get(SLOTS_PER_FILE - 1).unwrap(), Some(b"a".to_vec())); + assert_eq!(store.get(SLOTS_PER_FILE).unwrap(), Some(b"b".to_vec())); + } + + #[test] + fn batch_retries_committed_prefix() { + let dir = tempfile::tempdir().unwrap(); + let store = open(&dir); + + store.put_batch(vec![(0, b"a"), (1, b"b")]).unwrap(); + store + .put_batch(vec![(0, b"a"), (1, b"b"), (2, b"c")]) + .unwrap(); + + assert_eq!(store.get(2).unwrap(), Some(b"c".to_vec())); + assert!(store.put_batch(vec![(1, b"wrong")]).is_err()); + } + + #[test] + fn open_removes_uncommitted_future_file() { + let dir = tempfile::tempdir().unwrap(); + open(&dir).put(0, b"a").unwrap(); + fs::write(dir.path().join("data_00001"), b"stale").unwrap(); + fs::write(dir.path().join("data_00001.off"), b"stale").unwrap(); + + let store = open(&dir); + assert!(!dir.path().join("data_00001").exists()); + assert!(!dir.path().join("data_00001.off").exists()); + + store.put(SLOTS_PER_FILE, b"b").unwrap(); + assert_eq!(store.get(SLOTS_PER_FILE).unwrap(), Some(b"b".to_vec())); + } + + #[test] + fn config_mismatch_errors() { + let dir = tempfile::tempdir().unwrap(); + open(&dir); + + let other = Config { + record_type: [5, 0], + ..CONFIG + }; + assert!(StaticFile::open(dir.path().to_path_buf(), other).is_err()); + } +}