From 0cb2b970f6a28b035388e7463930905d1de052c1 Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Thu, 12 Nov 2026 17:54:31 -0500 Subject: [PATCH 1/3] refactor: frame AOF and snapshot records with a checked length --- crates/ember-core/src/shard/persistence.rs | 14 +- crates/ember-persistence/README.md | 39 +- crates/ember-persistence/src/aof/mod.rs | 322 +++------- crates/ember-persistence/src/aof/record.rs | 22 +- crates/ember-persistence/src/aof/tests.rs | 71 +-- crates/ember-persistence/src/encryption.rs | 37 +- crates/ember-persistence/src/format.rs | 117 +--- crates/ember-persistence/src/frame.rs | 365 ++++++++++++ crates/ember-persistence/src/lib.rs | 1 + crates/ember-persistence/src/recovery.rs | 131 +++-- crates/ember-persistence/src/snapshot.rs | 654 ++++++++------------- crates/ember-server/src/replication.rs | 13 +- 12 files changed, 853 insertions(+), 933 deletions(-) create mode 100644 crates/ember-persistence/src/frame.rs diff --git a/crates/ember-core/src/shard/persistence.rs b/crates/ember-core/src/shard/persistence.rs index d79386d8..a316f7c2 100644 --- a/crates/ember-core/src/shard/persistence.rs +++ b/crates/ember-core/src/shard/persistence.rs @@ -93,10 +93,10 @@ pub(super) fn handle_serialize_snapshot( ) -> ShardResponse { let entries: Vec = keyspace .iter_entries() - .map(|(key, value, expire_ms)| SnapEntry { + .map(|(key, value, ttl_ms)| SnapEntry { key: key.to_owned(), value: value_to_snap(value), - expire_ms, + expires_at_ms: expires_at(ttl_ms), }) .collect(); @@ -193,6 +193,14 @@ pub(super) fn handle_snapshot( } } +/// Turns the TTL left, or -1 for none, into the unix ms time a snapshot +/// stores, or 0 for none. +fn expires_at(ttl_ms: i64) -> u64 { + u64::try_from(ttl_ms).map_or(0, |ms| { + ember_persistence::aof::unix_now_ms().saturating_add(ms) + }) +} + /// Converts a `Value` reference to a `SnapValue` for serialization. pub(super) fn value_to_snap(value: &Value) -> SnapValue { match value { @@ -298,7 +306,7 @@ pub(super) fn write_snapshot( writer.write_entry(&SnapEntry { key: key.to_owned(), value: value_to_snap(value), - expire_ms: ttl_ms, + expires_at_ms: expires_at(ttl_ms), })?; count += 1; } diff --git a/crates/ember-persistence/README.md b/crates/ember-persistence/README.md index 91b5e4fd..aebec8d1 100644 --- a/crates/ember-persistence/README.md +++ b/crates/ember-persistence/README.md @@ -6,34 +6,37 @@ each shard gets its own persistence files (`shard-{id}.aof` and `shard-{id}.snap ## what's in here -- **aof** — append-only file writer/reader with CRC32 integrity checks. binary TLV format, configurable fsync (always, every-second, OS-managed). gracefully handles truncated records from mid-write crashes -- **snapshot** — point-in-time serialization of an entire shard's keyspace. writes to a `.tmp` file first, then atomic rename to prevent partial snapshots from corrupting existing data -- **recovery** — startup sequence: load snapshot, replay AOF tail, skip expired entries. handles corrupt files gracefully (logs warning, starts empty) -- **encryption** — optional AES-256-GCM encryption for AOF records and snapshot files (enabled with `--features encryption`). accepts a 32-byte key file (raw bytes or hex). existing plaintext files are read transparently and migrated on the next rewrite/snapshot -- **format** — low-level binary serialization helpers: length-prefixed bytes, integers, floats, checksums, header validation +- **aof**: the append-only file writer and reader. the writer queues records in memory and writes each batch with one call. a write that fails part way is cut back to the last whole record. fsync runs after every write, once a second, or when the OS decides +- **snapshot**: a point-in-time copy of one shard's keyspace. it is written to a `.tmp` file, fsynced, then renamed into place, so a crash while writing leaves the previous snapshot as it was +- **recovery**: the startup sequence. it loads the snapshot without the keys that have expired, then replays the AOF records written after it. a file that can't be read stops the start with an error and is left untouched. a partial record at the end of the AOF, left by a crash, is cut off +- **frame**: the file header and the record framing that both files share +- **encryption**: optional AES-256-GCM (`--features encryption`) with a 32-byte key file, as raw bytes or hex. each file is encrypted with its own key, derived from the master key and a salt in the file's header. plaintext files are still read, and are rewritten encrypted at startup +- **format**: binary helpers for length-prefixed bytes, integers, floats and checksums ## file formats -**AOF** — `[EAOF magic][version][record...]` where each record is `[tag][payload][crc32]` +every file starts with `[magic: 4B][version: 1B][flags: 1B]`, followed by a 16-byte salt when the encrypted flag is set. the current layout is version 5, and files in other versions are refused. records and snapshot entries are stored as frames: -supported record types: SET, DEL, EXPIRE, PERSIST, PEXPIRE, INCR, DECR, INCRBY, DECRBY, APPEND, RENAME, COPY, LPUSH, RPUSH, LPOP, RPOP, ZADD, ZREM, HSET, HDEL, HINCRBY, SADD, SREM (plus VADD, VREM, PROTO.SET, PROTO.REGISTER when compiled with those features) +```text +[len: 4B][!len: 4B][body][crc32 of body: 4B] +``` + +the second copy of the length, inverted, lets a reader tell a last frame cut short by a crash, which it drops, from a damaged length, which is an error. with encryption, the body is `[nonce: 12B][ciphertext with its 16-byte tag]`. -**snapshot (v2)** — `[ESNP magic][version][shard_id][entry_count][entries...][footer_crc32]` where entries are type-tagged (string=0, list=1, sorted set=2, hash=3, set=4). v1 snapshots (no type tags) are still readable. +**AOF** (`EAOF`): the header, then one frame per record. -**encrypted formats (v3)** — when an encryption key is configured, AOF records and snapshot entries are wrapped with AES-256-GCM: `[nonce (12 bytes)][ciphertext][auth tag (16 bytes)]`. the file headers and version bytes remain plaintext so the reader can detect encrypted files and select the right decoding path. plaintext v2 files are still readable even with a key configured. +supported record types: SET, DEL, EXPIRE, PERSIST, PEXPIRE, INCR, DECR, INCRBY, DECRBY, APPEND, RENAME, COPY, LPUSH, RPUSH, LPOP, RPOP, ZADD, ZREM, HSET, HDEL, HINCRBY, SADD, SREM (plus VADD, VREM, PROTO.SET, PROTO.REGISTER when compiled with those features) + +**snapshot** (`ESNP`): `[header][shard_id: 2B][one frame per entry][empty frame][entry count: 4B][crc32: 4B]`. each entry holds the key, a type-tagged value (string=0, list=1, sorted set=2, hash=3, set=4, protobuf=5, vector=6) and the unix time in milliseconds when it expires, or 0 for never. the trailing CRC covers the entry frames and identifies the snapshot: the AOF's checkpoint record names the snapshot it follows by this CRC. ## usage ```rust -use ember_persistence::aof::{AofWriter, AofRecord}; -use ember_persistence::snapshot::{SnapshotWriter, SnapshotReader, SnapEntry, SnapValue}; -use ember_persistence::recovery::recover_shard; - -// recovery is typically called by the shard on startup -let result = recover_shard(data_dir, shard_id); -for entry in result.entries { - // insert into keyspace... -} +use ember_persistence::recovery::{recover_shard, Recover}; + +// the shard implements `Recover` with its keyspace, so recovered entries +// and replayed records go through the same code as live writes +let result = recover_shard(data_dir, shard_id, &mut keyspace)?; ``` ## related crates diff --git a/crates/ember-persistence/src/aof/mod.rs b/crates/ember-persistence/src/aof/mod.rs index e8d6113f..3bed486a 100644 --- a/crates/ember-persistence/src/aof/mod.rs +++ b/crates/ember-persistence/src/aof/mod.rs @@ -1,20 +1,8 @@ //! Append-only file for recording mutations. //! -//! Each shard writes its own AOF file (`shard-{id}.aof`). Records are -//! written after successful mutations. The binary format uses a simple -//! tag + payload + CRC32 structure for each record. -//! -//! File layout: -//! ```text -//! [EAOF magic: 4B][version: 1B] -//! [record]* -//! ``` -//! -//! Record layout: -//! ```text -//! [tag: 1B][payload...][crc32: 4B] -//! ``` -//! The CRC32 covers the tag + payload bytes. +//! Each shard writes its own AOF file (`shard-{id}.aof`): a header, then one +//! frame per record, in the layout described in [`crate::frame`]. Records +//! are written after successful mutations. use std::fmt; use std::fs::{self, File, OpenOptions}; @@ -22,6 +10,7 @@ use std::io::{BufReader, Seek, Write}; use std::path::{Path, PathBuf}; use crate::format::{self, FormatError}; +use crate::frame::{self, Codec, Frame, Key}; mod record; @@ -49,6 +38,10 @@ pub enum FsyncPolicy { pub struct AofWriter { file: File, path: PathBuf, + /// Encodes records for this file. + codec: Codec, + /// The key new files are encrypted with. + key: Option, /// Encoded records not yet written to the file. pending: Vec, /// Length of the file up to the end of the last whole record. @@ -59,96 +52,57 @@ pub struct AofWriter { /// The existing file is in a different format than this writer writes. /// See [`AofWriter::needs_rewrite`]. needs_rewrite: bool, - #[cfg(feature = "encryption")] - encryption: Option, -} - -/// The master key, kept to start new files, and the cipher of the current one. -#[cfg(feature = "encryption")] -struct Encryption { - key: crate::encryption::EncryptionKey, - cipher: crate::encryption::FileCipher, } impl AofWriter { /// Opens (or creates) a plaintext AOF file. A new file gets a header; /// an existing one is appended to. pub fn open(path: impl Into) -> Result { - Self::open_versioned( - path.into(), - format::FORMAT_VERSION, - #[cfg(feature = "encryption")] - None, - ) + Self::open_with(path.into(), None) } - /// Opens (or creates) an encrypted AOF file using AES-256-GCM. A new - /// file gets a v4 header. + /// Opens (or creates) an AOF file encrypted with `key`. #[cfg(feature = "encryption")] - pub fn open_encrypted( - path: impl Into, - key: crate::encryption::EncryptionKey, - ) -> Result { - Self::open_versioned(path.into(), format::FORMAT_VERSION_ENCRYPTED, Some(key)) + pub fn open_encrypted(path: impl Into, key: Key) -> Result { + Self::open_with(path.into(), Some(key)) } - fn open_versioned( - path: PathBuf, - version: u8, - #[cfg(feature = "encryption")] key: Option, - ) -> Result { - let existing = fs::metadata(&path).map(|m| m.len() > 0).unwrap_or(false); + fn open_with(path: PathBuf, key: Option) -> Result { + let existing = fs::metadata(&path).is_ok_and(|m| m.len() > 0); let mut file = open_persistence_file(&path)?; - let mut writer = Vec::new(); - - // the file's current version, and for an encrypted file the - // cipher to keep appending with. a header that cannot be read - // counts as a different format. - #[cfg(feature = "encryption")] - let (current_version, cipher) = if existing { - match read_existing_header(&path, key.as_ref()) { - Ok((v, cipher)) => (Some(v), cipher), - Err(_) => (None, None), + + // an existing file is appended to with its own codec. one this + // writer can't append to, encrypted the other way or unreadable, + // must be rewritten first, which starts a new file with a new + // header; until then any codec will do + let mut header = Vec::new(); + let (codec, needs_rewrite) = if existing { + let found = File::open(&path).map_err(FormatError::from).and_then(|f| { + frame::read_header(&mut BufReader::new(f), format::AOF_MAGIC, key.as_ref()) + }); + match found { + Ok(codec) if codec.is_encrypted() == key.is_some() => (codec, false), + _ => ( + frame::write_header(&mut header, format::AOF_MAGIC, key.as_ref())?, + true, + ), } } else { - format::write_header_versioned(&mut writer, format::AOF_MAGIC, version)?; - let cipher = key - .as_ref() - .map(|k| k.write_new_file_salt(&mut writer)) - .transpose()?; - (Some(version), cipher) + let codec = frame::write_header(&mut header, format::AOF_MAGIC, key.as_ref())?; + file.write_all(&header)?; + (codec, false) }; - #[cfg(not(feature = "encryption"))] - let current_version = if existing { - File::open(&path) - .map_err(FormatError::from) - .and_then(|f| format::read_header(&mut BufReader::new(f), format::AOF_MAGIC)) - .ok() - } else { - format::write_header_versioned(&mut writer, format::AOF_MAGIC, version)?; - Some(version) - }; - // a new file's header goes straight to disk - file.write_all(&writer)?; let written_len = file.metadata()?.len(); - // a file that needs a rewrite is truncated before anything is - // written, which picks a new salt; until then any cipher will do - #[cfg(feature = "encryption")] - let encryption = key.map(|key| { - let cipher = cipher.unwrap_or_else(|| key.legacy_cipher()); - Encryption { key, cipher } - }); - Ok(Self { file, path, + codec, + key, pending: Vec::new(), written_len, torn: false, - needs_rewrite: current_version != Some(version), - #[cfg(feature = "encryption")] - encryption, + needs_rewrite, }) } @@ -163,12 +117,7 @@ impl AofWriter { /// Queues a record. It reaches the file on the next /// [`flush`](Self::flush) or [`sync`](Self::sync). pub fn write_record(&mut self, record: &AofRecord) -> Result<(), FormatError> { - write_encoded( - &mut self.pending, - record, - #[cfg(feature = "encryption")] - self.encryption.as_ref().map(|enc| &enc.cipher), - ) + encode_record(&self.codec, record, &mut self.pending) } /// Writes the queued records to the OS. On failure the file is cut @@ -215,34 +164,14 @@ impl AofWriter { _ => {} } let mut tmp_file = open_persistence_file(&tmp_path)?; - let mut tmp_writer = Vec::new(); - - // the new file's cipher. the old one stays in use until the rename - #[cfg(feature = "encryption")] - let cipher = match self.encryption { - Some(ref enc) => { - format::write_header_versioned( - &mut tmp_writer, - format::AOF_MAGIC, - format::FORMAT_VERSION_ENCRYPTED, - )?; - Some(enc.key.write_new_file_salt(&mut tmp_writer)?) - } - None => { - format::write_header(&mut tmp_writer, format::AOF_MAGIC)?; - None - } - }; - #[cfg(not(feature = "encryption"))] - format::write_header(&mut tmp_writer, format::AOF_MAGIC)?; - - write_encoded( - &mut tmp_writer, + let mut contents = Vec::new(); + let codec = frame::write_header(&mut contents, format::AOF_MAGIC, self.key.as_ref())?; + encode_record( + &codec, &AofRecord::Checkpoint { snapshot_crc }, - #[cfg(feature = "encryption")] - cipher.as_ref(), + &mut contents, )?; - tmp_file.write_all(&tmp_writer)?; + tmp_file.write_all(&contents)?; tmp_file.sync_all()?; // atomic rename: old AOF is replaced only after new file is durable. @@ -251,17 +180,21 @@ impl AofWriter { fs::rename(&tmp_path, &self.path)?; sync_parent_dir(&self.path); self.file = tmp_file; - self.written_len = tmp_writer.len() as u64; + self.codec = codec; + self.written_len = contents.len() as u64; self.torn = false; - #[cfg(feature = "encryption")] - if let (Some(enc), Some(cipher)) = (self.encryption.as_mut(), cipher) { - enc.cipher = cipher; - } self.needs_rewrite = false; Ok(()) } } +/// Appends `record` to `out` as one frame. +fn encode_record(codec: &Codec, record: &AofRecord, out: &mut Vec) -> Result<(), FormatError> { + let plain = record.to_bytes()?; + frame::write_frame(out, &codec.seal(&plain)?)?; + Ok(()) +} + /// A file that records are appended to. A trait so tests can make writes /// fail part way. trait AppendFile: Write { @@ -319,114 +252,43 @@ pub(crate) fn sync_parent_dir(path: &Path) { } } -/// Writes one record. With a cipher: `[nonce: 12B][len: 4B][ciphertext]`. -/// Otherwise the v2 format: `[tag+payload][crc32: 4B]`. -fn write_encoded( - w: &mut impl Write, - record: &AofRecord, - #[cfg(feature = "encryption")] cipher: Option<&crate::encryption::FileCipher>, -) -> Result<(), FormatError> { - let payload = record.to_bytes()?; - - #[cfg(feature = "encryption")] - if let Some(cipher) = cipher { - let (nonce, ciphertext) = cipher.encrypt(&payload)?; - w.write_all(&nonce)?; - format::write_len(w, ciphertext.len())?; - w.write_all(&ciphertext)?; - return Ok(()); - } - - let checksum = format::crc32(&payload); - w.write_all(&payload)?; - format::write_u32(w, checksum)?; - Ok(()) -} - -/// Reads the header of an existing AOF: its version and, when it is -/// encrypted and `key` is given, the cipher for its records. -#[cfg(feature = "encryption")] -fn read_existing_header( - path: &Path, - key: Option<&crate::encryption::EncryptionKey>, -) -> Result<(u8, Option), FormatError> { - let mut reader = BufReader::new(File::open(path)?); - let version = format::read_header(&mut reader, format::AOF_MAGIC)?; - let cipher = match key { - Some(key) if format::is_encrypted(version) => { - Some(key.read_file_cipher(version, &mut reader)?) - } - _ => None, - }; - Ok((version, cipher)) -} - /// Reader for iterating over AOF records. pub struct AofReader { reader: BufReader, - /// Format version from the file header. v2 = plaintext, v3/v4 = encrypted. - version: u8, - /// End offset of the last complete record read. See [`AofReader::valid_len`]. + codec: Codec, + /// End offset of the last whole record read. See [`AofReader::valid_len`]. valid_len: u64, - /// Set for an encrypted file. - #[cfg(feature = "encryption")] - cipher: Option, } impl fmt::Debug for AofReader { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("AofReader") - .field("version", &self.version) + .field("encrypted", &self.codec.is_encrypted()) .finish() } } impl AofReader { - /// Opens an AOF file and validates the header. + /// Opens a plaintext AOF file and reads the header. pub fn open(path: impl AsRef) -> Result { - let reader = Self::open_with_key( - path, - #[cfg(feature = "encryption")] - None, - )?; - if format::is_encrypted(reader.version) { - return Err(FormatError::EncryptionRequired); - } - Ok(reader) + Self::open_with(path.as_ref(), None) } - /// Opens an AOF file with an encryption key for decrypting v3/v4 records. - /// - /// Also handles v2 (plaintext) files — the key is simply unused, - /// allowing transparent migration. + /// Opens an AOF file that may be encrypted with `key`. Plaintext files + /// are read too. #[cfg(feature = "encryption")] - pub fn open_encrypted( - path: impl AsRef, - key: crate::encryption::EncryptionKey, - ) -> Result { - Self::open_with_key(path, Some(key)) + pub fn open_encrypted(path: impl AsRef, key: Key) -> Result { + Self::open_with(path.as_ref(), Some(&key)) } - fn open_with_key( - path: impl AsRef, - #[cfg(feature = "encryption")] encryption_key: Option, - ) -> Result { - let mut reader = BufReader::new(File::open(path.as_ref())?); - let version = format::read_header(&mut reader, format::AOF_MAGIC)?; - #[cfg(feature = "encryption")] - let cipher = match encryption_key { - Some(key) if format::is_encrypted(version) => { - Some(key.read_file_cipher(version, &mut reader)?) - } - _ => None, - }; + pub(crate) fn open_with(path: &Path, key: Option<&Key>) -> Result { + let mut reader = BufReader::new(File::open(path)?); + let codec = frame::read_header(&mut reader, format::AOF_MAGIC, key)?; let valid_len = reader.stream_position()?; Ok(Self { reader, - version, + codec, valid_len, - #[cfg(feature = "encryption")] - cipher, }) } @@ -434,24 +296,15 @@ impl AofReader { /// /// Returns `Ok(None)` at end of file. A record cut short by a crash /// mid-write also returns `Ok(None)`; [`valid_len`](Self::valid_len) - /// then tells the caller where the complete records end. + /// then tells the caller where the whole records end. pub fn read_record(&mut self) -> Result, FormatError> { - #[cfg(feature = "encryption")] - let result = if format::is_encrypted(self.version) { - self.read_encrypted_record() - } else { - self.read_v2_record() - }; - #[cfg(not(feature = "encryption"))] - let result = self.read_v2_record(); - - match result { - Ok(record) => { + match frame::read_frame(&mut self.reader)? { + Frame::End | Frame::Torn => Ok(None), + Frame::Body(body) => { + let record = AofRecord::from_bytes(&self.codec.open(body)?)?; self.valid_len = self.reader.stream_position()?; Ok(Some(record)) } - Err(FormatError::UnexpectedEof) => Ok(None), - Err(e) => Err(e), } } @@ -461,41 +314,6 @@ impl AofReader { pub fn valid_len(&self) -> u64 { self.valid_len } - - /// Reads a v2 (plaintext) record: tag + payload + crc32. - fn read_v2_record(&mut self) -> Result { - let mut crc_reader = format::CrcReader::new(&mut self.reader); - let record = AofRecord::decode(&mut crc_reader)?; - let computed = crc_reader.finalize(); - let stored = format::read_u32(&mut self.reader)?; - format::verify_crc32_values(computed, stored)?; - Ok(record) - } - - /// Reads an encrypted record: nonce + len + ciphertext. - #[cfg(feature = "encryption")] - fn read_encrypted_record(&mut self) -> Result { - let cipher = self - .cipher - .as_ref() - .ok_or(FormatError::EncryptionRequired)?; - - let mut nonce = [0u8; crate::encryption::NONCE_SIZE]; - format::read_exact(&mut self.reader, &mut nonce)?; - - let ct_len = format::read_u32(&mut self.reader)? as usize; - if ct_len > format::MAX_FIELD_LEN { - return Err(FormatError::Io(std::io::Error::new( - std::io::ErrorKind::InvalidData, - format!("encrypted record length {ct_len} exceeds maximum"), - ))); - } - let mut ciphertext = vec![0u8; ct_len]; - format::read_exact(&mut self.reader, &mut ciphertext)?; - - let plaintext = cipher.decrypt(&nonce, &ciphertext)?; - AofRecord::from_bytes(&plaintext) - } } /// Opens a persistence file with create+append and restrictive permissions. diff --git a/crates/ember-persistence/src/aof/record.rs b/crates/ember-persistence/src/aof/record.rs index 4e284621..5cac2d39 100644 --- a/crates/ember-persistence/src/aof/record.rs +++ b/crates/ember-persistence/src/aof/record.rs @@ -713,19 +713,21 @@ impl AofRecord { Ok(buf) } - /// Deserializes a record from its binary payload (tag byte + fields, no CRC). - /// - /// The format is the same as `to_bytes()`. CRC validation is the caller's - /// responsibility. + /// Deserializes a record from the bytes `to_bytes()` produced. Bytes + /// left over after the record are an error. pub fn from_bytes(data: &[u8]) -> Result { - Self::decode(&mut io::Cursor::new(data)) + let mut cursor = io::Cursor::new(data); + let record = Self::decode(&mut cursor)?; + if cursor.position() != data.len() as u64 { + return Err(FormatError::InvalidData( + "aof record has trailing bytes".into(), + )); + } + Ok(record) } - /// Decodes one record (tag byte + fields, no CRC) from a stream. - /// - /// This is the only record decoder. The AOF reader runs it over a - /// [`format::CrcReader`] so the checksum covers exactly the bytes decoded. - pub(super) fn decode(cursor: &mut impl io::Read) -> Result { + /// Decodes one record (tag byte + fields) from a stream. + fn decode(cursor: &mut impl io::Read) -> Result { let tag = format::read_u8(cursor)?; match tag { TAG_SET => { diff --git a/crates/ember-persistence/src/aof/tests.rs b/crates/ember-persistence/src/aof/tests.rs index c89ba50a..10f55ee8 100644 --- a/crates/ember-persistence/src/aof/tests.rs +++ b/crates/ember-persistence/src/aof/tests.rs @@ -925,53 +925,6 @@ mod encrypted { AofRecord::Del { key: key.into() } } - /// Writes a v3 AOF, which encrypts with the master key directly. - fn write_v3(path: &Path, key: &EncryptionKey, records: &[AofRecord]) -> Result { - let mut file = Vec::new(); - format::write_header_versioned( - &mut file, - format::AOF_MAGIC, - format::FORMAT_VERSION_ENCRYPTED_V3, - )?; - let cipher = key.legacy_cipher(); - for record in records { - let (nonce, ciphertext) = cipher.encrypt(&record.to_bytes()?)?; - file.extend_from_slice(&nonce); - format::write_len(&mut file, ciphertext.len())?; - file.extend_from_slice(&ciphertext); - } - fs::write(path, file)?; - Ok(()) - } - - #[test] - fn v3_file_is_still_read_and_needs_rewrite() -> Result { - let dir = temp_dir(); - let path = dir.path().join("v3.aof"); - let key = test_key(); - write_v3(&path, &key, &[del("a"), del("b")])?; - - let mut reader = AofReader::open_encrypted(&path, key.clone())?; - assert_eq!(reader.read_record()?, Some(del("a"))); - assert_eq!(reader.read_record()?, Some(del("b"))); - assert_eq!(reader.read_record()?, None); - - let mut writer = AofWriter::open_encrypted(&path, key.clone())?; - assert!(writer.needs_rewrite()); - writer.truncate(7)?; - writer.write_record(&del("c"))?; - writer.sync()?; - - let mut reader = AofReader::open_encrypted(&path, key)?; - assert_eq!(reader.version, format::FORMAT_VERSION_ENCRYPTED); - assert_eq!( - reader.read_record()?, - Some(AofRecord::Checkpoint { snapshot_crc: 7 }) - ); - assert_eq!(reader.read_record()?, Some(del("c"))); - Ok(()) - } - #[test] fn reopened_v4_file_keeps_its_salt() -> Result { let dir = temp_dir(); @@ -1129,3 +1082,27 @@ fn appends_after_a_truncate_land_in_the_new_file() -> Result { assert!(!path.with_extension("aof.tmp").exists()); Ok(()) } + +#[test] +fn a_file_in_an_older_layout_is_refused_and_rewritten() -> Result { + let dir = temp_dir(); + let path = dir.path().join("old.aof"); + // a version 2 header, the layout before frames + fs::write(&path, b"EAOF\x02")?; + assert!(matches!( + AofReader::open(&path), + Err(FormatError::UnsupportedVersion(2)) + )); + + let mut writer = AofWriter::open(&path)?; + assert!(writer.needs_rewrite()); + writer.truncate(7)?; + writer.write_record(&AofRecord::Del { key: "k".into() })?; + drop(writer); + let mut reader = AofReader::open(&path)?; + assert_eq!( + reader.read_record()?, + Some(AofRecord::Checkpoint { snapshot_crc: 7 }) + ); + Ok(()) +} diff --git a/crates/ember-persistence/src/encryption.rs b/crates/ember-persistence/src/encryption.rs index 961c3a0b..ff838d03 100644 --- a/crates/ember-persistence/src/encryption.rs +++ b/crates/ember-persistence/src/encryption.rs @@ -4,11 +4,11 @@ //! random 12-byte nonce. AES-GCM provides authenticated encryption — a //! tampered ciphertext is detected immediately rather than producing garbage. //! -//! Each v4 file starts with a random salt, and its records are encrypted -//! with a key derived from the master key and that salt. Random nonces -//! stay far from GCM's collision limits because no key encrypts more than -//! one file, and a record copied from one file into another fails to -//! decrypt. v3 files, which use the master key directly, are still read. +//! Each encrypted file's header holds a random salt, and its records are +//! encrypted with a key derived from the master key and that salt. Random +//! nonces stay far from GCM's collision limits because no key encrypts more +//! than one file, and a record copied from one file into another fails to +//! decrypt. //! //! This module is only compiled when the `encryption` feature is enabled. @@ -31,7 +31,7 @@ pub const NONCE_SIZE: usize = 12; /// AES-256-GCM authentication tag size in bytes. pub const TAG_SIZE: usize = 16; -/// Size of the random salt at the start of each v4 file. +/// Size of the random salt in each encrypted file's header. pub const SALT_SIZE: usize = 16; /// A 256-bit master key for AES-256-GCM. @@ -118,7 +118,7 @@ impl EncryptionKey { &self.bytes } - /// Returns the cipher for a v4 file with this salt. Its key is derived + /// Returns the cipher for a file with this salt. Its key is derived /// from the master key and the salt with HKDF-SHA256. pub fn file_cipher(&self, salt: &[u8; SALT_SIZE]) -> FileCipher { let mut subkey = Zeroizing::new([0u8; 32]); @@ -128,11 +128,6 @@ impl EncryptionKey { FileCipher::new(&subkey) } - /// Returns the cipher for a v3 file, which used the master key directly. - pub fn legacy_cipher(&self) -> FileCipher { - FileCipher::new(&self.bytes) - } - /// Picks a salt for a new file, writes it to `w` and returns the file's /// cipher. pub fn write_new_file_salt(&self, w: &mut impl Write) -> io::Result { @@ -142,16 +137,9 @@ impl EncryptionKey { Ok(self.file_cipher(&salt)) } - /// Returns the cipher for an existing file with header `version`. For a - /// v4 file, reads the salt from `r`, which must be positioned at it. - pub fn read_file_cipher( - &self, - version: u8, - r: &mut impl Read, - ) -> Result { - if version == format::FORMAT_VERSION_ENCRYPTED_V3 { - return Ok(self.legacy_cipher()); - } + /// Reads a file's salt from `r`, which must be positioned at it, and + /// returns the file's cipher. + pub fn read_file_cipher(&self, r: &mut impl Read) -> Result { let mut salt = [0u8; SALT_SIZE]; format::read_exact(r, &mut salt)?; Ok(self.file_cipher(&salt)) @@ -217,7 +205,6 @@ mod tests { other.decrypt(&nonce, &ciphertext), Err(FormatError::DecryptionFailed) )); - assert!(key.legacy_cipher().decrypt(&nonce, &ciphertext).is_err()); } #[test] @@ -229,9 +216,7 @@ mod tests { .unwrap() .encrypt(b"data") .unwrap(); - let cipher = key - .read_file_cipher(format::FORMAT_VERSION_ENCRYPTED, &mut file.as_slice()) - .unwrap(); + let cipher = key.read_file_cipher(&mut file.as_slice()).unwrap(); assert_eq!(cipher.decrypt(&nonce, &ciphertext).unwrap(), b"data"); } diff --git a/crates/ember-persistence/src/format.rs b/crates/ember-persistence/src/format.rs index fad43ece..c06c71e1 100644 --- a/crates/ember-persistence/src/format.rs +++ b/crates/ember-persistence/src/format.rs @@ -2,6 +2,7 @@ //! //! Provides TLV-style encoding primitives, CRC32 checksums, and magic //! byte constants. All multi-byte integers are stored in little-endian. +//! The file header and record framing are in [`crate::frame`]. use std::io::{self, Read, Write}; @@ -14,27 +15,6 @@ pub const AOF_MAGIC: &[u8; 4] = b"EAOF"; /// Magic bytes for the snapshot file header. pub const SNAP_MAGIC: &[u8; 4] = b"ESNP"; -/// Current unencrypted format version. -/// -/// v1: original format (strings only) -/// v2: type-tagged entries (string, list, sorted set, hash, set) -pub const FORMAT_VERSION: u8 = 2; - -/// Format version for encrypted files (requires the `encryption` feature). -/// -/// v4: per-record AES-256-GCM with a key derived from the master key and a -/// random salt stored in the file header. -pub const FORMAT_VERSION_ENCRYPTED: u8 = 4; - -/// The earlier encrypted format, which used the master key for every file. -/// Still read; files in it are rewritten as v4. -pub const FORMAT_VERSION_ENCRYPTED_V3: u8 = 3; - -/// Whether files with this header version hold encrypted records. -pub fn is_encrypted(version: u8) -> bool { - version == FORMAT_VERSION_ENCRYPTED || version == FORMAT_VERSION_ENCRYPTED_V3 -} - /// Errors that can occur when reading or writing persistence formats. #[derive(Debug, Error)] pub enum FormatError { @@ -73,43 +53,6 @@ pub fn crc32(data: &[u8]) -> u32 { h.finalize() } -/// Reader adapter that feeds every byte it returns into a CRC32 hasher. -/// -/// A record can then be decoded straight from the stream and its checksum -/// compared afterwards, with no second parser to rebuild the record bytes. -pub struct CrcReader { - inner: R, - hasher: Hasher, -} - -impl CrcReader { - pub fn new(inner: R) -> Self { - Self { - inner, - hasher: Hasher::new(), - } - } - - /// Returns the checksum of every byte read so far. - pub fn finalize(self) -> u32 { - self.hasher.finalize() - } - - /// Returns the hasher, to combine with a running checksum over several - /// records. - pub fn into_hasher(self) -> Hasher { - self.hasher - } -} - -impl Read for CrcReader { - fn read(&mut self, buf: &mut [u8]) -> io::Result { - let n = self.inner.read(buf)?; - self.hasher.update(&buf[..n]); - Ok(n) - } -} - // --------------------------------------------------------------------------- // write helpers // --------------------------------------------------------------------------- @@ -248,42 +191,6 @@ pub fn read_exact(r: &mut impl Read, buf: &mut [u8]) -> Result<(), FormatError> }) } -/// Writes a file header: magic bytes + version byte. -pub fn write_header(w: &mut impl Write, magic: &[u8; 4]) -> io::Result<()> { - w.write_all(magic)?; - write_u8(w, FORMAT_VERSION) -} - -/// Writes a file header with an explicit version byte. -pub fn write_header_versioned(w: &mut impl Write, magic: &[u8; 4], version: u8) -> io::Result<()> { - w.write_all(magic)?; - write_u8(w, version) -} - -/// The maximum format version this build can read. -/// -/// When the `encryption` feature is compiled in, v3 and v4 (encrypted) -/// files are supported. Without the feature, only v1 and v2 are accepted. -#[cfg(feature = "encryption")] -const MAX_READABLE_VERSION: u8 = FORMAT_VERSION_ENCRYPTED; -#[cfg(not(feature = "encryption"))] -const MAX_READABLE_VERSION: u8 = FORMAT_VERSION; - -/// Reads and validates a file header. Returns an error if magic doesn't -/// match or version is unsupported. Returns the format version. -pub fn read_header(r: &mut impl Read, expected_magic: &[u8; 4]) -> Result { - let mut magic = [0u8; 4]; - read_exact(r, &mut magic)?; - if &magic != expected_magic { - return Err(FormatError::InvalidMagic); - } - let version = read_u8(r)?; - if version == 0 || version > MAX_READABLE_VERSION { - return Err(FormatError::UnsupportedVersion(version)); - } - Ok(version) -} - /// Verifies that `data` matches the expected CRC32 checksum. pub fn verify_crc32(data: &[u8], expected: u32) -> Result<(), FormatError> { let actual = crc32(data); @@ -408,28 +315,6 @@ mod tests { assert_eq!(read_bytes(&mut Cursor::new(&buf)).unwrap(), b""); } - #[test] - fn header_round_trip() { - let mut buf = Vec::new(); - write_header(&mut buf, AOF_MAGIC).unwrap(); - read_header(&mut Cursor::new(&buf), AOF_MAGIC).unwrap(); - } - - #[test] - fn header_wrong_magic() { - let mut buf = Vec::new(); - write_header(&mut buf, AOF_MAGIC).unwrap(); - let err = read_header(&mut Cursor::new(&buf), SNAP_MAGIC).unwrap_err(); - assert!(matches!(err, FormatError::InvalidMagic)); - } - - #[test] - fn header_wrong_version() { - let buf = vec![b'E', b'A', b'O', b'F', 99]; - let err = read_header(&mut Cursor::new(&buf), AOF_MAGIC).unwrap_err(); - assert!(matches!(err, FormatError::UnsupportedVersion(99))); - } - #[test] fn crc32_deterministic() { let a = crc32(b"test data"); diff --git a/crates/ember-persistence/src/frame.rs b/crates/ember-persistence/src/frame.rs new file mode 100644 index 00000000..ca924fc9 --- /dev/null +++ b/crates/ember-persistence/src/frame.rs @@ -0,0 +1,365 @@ +//! The layout shared by AOF and snapshot files. +//! +//! Each file starts with a header: +//! ```text +//! [magic: 4B][version: 1B][flags: 1B][salt: 16B, only when encrypted] +//! ``` +//! Records and snapshot entries are then stored as frames: +//! ```text +//! [len: 4B][!len: 4B][body: len B][crc32 of body: 4B] +//! ``` +//! Without encryption the body is the record's bytes. With it, the body is +//! `[nonce: 12B][ciphertext]`, encrypted with a key derived from the master +//! key and the file's salt. All integers are little-endian. +//! +//! The inverted copy of the length tells a damaged length apart from a +//! frame cut short. A crash while appending leaves a prefix of one frame: +//! either fewer than 8 bytes, or a length that checks out and more bytes +//! than are left. Some filesystems instead leave the end of the file +//! zeroed. Readers report both as [`Frame::Torn`]. Any other damage, a +//! length that fails its check or a body that fails its CRC, is an error. + +use std::borrow::Cow; +use std::io::{self, Read}; + +use crate::format::{self, FormatError}; + +/// The file layout version. Files in any other layout are refused. +pub const VERSION: u8 = 5; + +/// Header flag: frame bodies are encrypted. +const FLAG_ENCRYPTED: u8 = 1; + +/// The key for encrypted files. A type with no values in builds without +/// the `encryption` feature, so code can take an `Option<&Key>` either way. +#[cfg(feature = "encryption")] +pub use crate::encryption::EncryptionKey as Key; +#[cfg(not(feature = "encryption"))] +#[derive(Debug, Clone)] +pub enum Key {} + +/// How a file's frame bodies are encoded. +pub enum Codec { + Plain, + /// Boxed: the cipher's key schedule is large. + #[cfg(feature = "encryption")] + Encrypted(Box), +} + +impl Codec { + pub fn is_encrypted(&self) -> bool { + !matches!(self, Codec::Plain) + } + + /// Encodes a record or entry as a frame body. + pub fn seal<'a>(&self, plain: &'a [u8]) -> Result, FormatError> { + match self { + Codec::Plain => Ok(Cow::Borrowed(plain)), + #[cfg(feature = "encryption")] + Codec::Encrypted(cipher) => { + let (nonce, ciphertext) = cipher.encrypt(plain)?; + let mut body = Vec::with_capacity(nonce.len() + ciphertext.len()); + body.extend_from_slice(&nonce); + body.extend_from_slice(&ciphertext); + Ok(Cow::Owned(body)) + } + } + } + + /// Decodes a frame body back into the record or entry bytes. + pub fn open(&self, body: Vec) -> Result, FormatError> { + match self { + Codec::Plain => Ok(body), + #[cfg(feature = "encryption")] + Codec::Encrypted(cipher) => { + use crate::encryption::NONCE_SIZE; + if body.len() < NONCE_SIZE { + return Err(FormatError::DecryptionFailed); + } + let (nonce, ciphertext) = body.split_at(NONCE_SIZE); + let nonce: &[u8; NONCE_SIZE] = nonce.try_into().expect("split at NONCE_SIZE"); + cipher.decrypt(nonce, ciphertext) + } + } + } +} + +/// Writes a header for a new file and returns the codec for its frames. +/// With a key, the file is encrypted under a fresh salt. +pub fn write_header( + out: &mut Vec, + magic: &[u8; 4], + key: Option<&Key>, +) -> Result { + out.extend_from_slice(magic); + out.push(VERSION); + match key { + None => { + out.push(0); + Ok(Codec::Plain) + } + #[cfg(feature = "encryption")] + Some(key) => { + out.push(FLAG_ENCRYPTED); + Ok(Codec::Encrypted(Box::new(key.write_new_file_salt(out)?))) + } + #[cfg(not(feature = "encryption"))] + Some(never) => match *never {}, + } +} + +/// Reads a file header and returns the codec for its frames. An encrypted +/// file needs `key`; a plaintext one is read with or without it. +pub fn read_header( + r: &mut impl Read, + magic: &[u8; 4], + #[allow(unused_variables)] key: Option<&Key>, +) -> Result { + let mut found = [0u8; 4]; + format::read_exact(r, &mut found)?; + if &found != magic { + return Err(FormatError::InvalidMagic); + } + let version = format::read_u8(r)?; + if version != VERSION { + return Err(FormatError::UnsupportedVersion(version)); + } + let flags = format::read_u8(r)?; + if flags & !FLAG_ENCRYPTED != 0 { + return Err(FormatError::InvalidData(format!( + "unknown header flags {flags:#04x}" + ))); + } + if flags & FLAG_ENCRYPTED == 0 { + return Ok(Codec::Plain); + } + #[cfg(feature = "encryption")] + if let Some(key) = key { + return Ok(Codec::Encrypted(Box::new(key.read_file_cipher(r)?))); + } + Err(FormatError::EncryptionRequired) +} + +/// Appends `body` to `out` as one frame. +pub fn write_frame(out: &mut Vec, body: &[u8]) -> io::Result<()> { + let len = u32::try_from(body.len()).map_err(|_| { + io::Error::new( + io::ErrorKind::InvalidInput, + format!("record of {} bytes exceeds u32::MAX", body.len()), + ) + })?; + out.extend_from_slice(&len.to_le_bytes()); + out.extend_from_slice(&(!len).to_le_bytes()); + out.extend_from_slice(body); + out.extend_from_slice(&format::crc32(body).to_le_bytes()); + Ok(()) +} + +/// What [`read_frame`] found. +#[derive(Debug, PartialEq, Eq)] +pub enum Frame { + /// A whole frame's body. + Body(Vec), + /// The end of the file, at a frame boundary. + End, + /// The rest of the file is part of one frame or zeros, as a crash + /// while appending leaves it. + Torn, +} + +/// Reads the next frame. +pub fn read_frame(r: &mut impl Read) -> Result { + let mut head = [0u8; 8]; + let got = read_up_to(r, &mut head)?; + if got == 0 { + return Ok(Frame::End); + } + if got < head.len() { + return Ok(Frame::Torn); + } + let len = u32::from_le_bytes(head[..4].try_into().expect("4 bytes")); + let check = u32::from_le_bytes(head[4..].try_into().expect("4 bytes")); + if check != !len { + if head == [0; 8] && rest_is_zero(r)? { + return Ok(Frame::Torn); + } + return Err(FormatError::InvalidData( + "frame length failed its check".into(), + )); + } + if len as usize > format::MAX_FIELD_LEN { + return Err(FormatError::InvalidData(format!( + "frame length {len} exceeds the maximum" + ))); + } + + let mut body = vec![0u8; len as usize]; + if read_up_to(r, &mut body)? < body.len() { + return Ok(Frame::Torn); + } + let mut crc = [0u8; 4]; + if read_up_to(r, &mut crc)? < crc.len() { + return Ok(Frame::Torn); + } + format::verify_crc32(&body, u32::from_le_bytes(crc))?; + Ok(Frame::Body(body)) +} + +/// Fills `buf` as far as the input allows and returns how many bytes it +/// read. +fn read_up_to(r: &mut impl Read, buf: &mut [u8]) -> Result { + let mut filled = 0; + while filled < buf.len() { + match r.read(&mut buf[filled..]) { + Ok(0) => break, + Ok(n) => filled += n, + Err(e) if e.kind() == io::ErrorKind::Interrupted => {} + Err(e) => return Err(e.into()), + } + } + Ok(filled) +} + +/// Whether everything left in `r` is zero bytes. +fn rest_is_zero(r: &mut impl Read) -> Result { + let mut buf = [0u8; 4096]; + loop { + let n = read_up_to(r, &mut buf)?; + if buf[..n].iter().any(|&b| b != 0) { + return Ok(false); + } + if n < buf.len() { + return Ok(true); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn frames(bodies: &[&[u8]]) -> Vec { + let mut out = Vec::new(); + for body in bodies { + write_frame(&mut out, body).unwrap(); + } + out + } + + fn read_all(mut data: &[u8]) -> Result, FormatError> { + let mut found = Vec::new(); + loop { + let frame = read_frame(&mut data)?; + let last = !matches!(frame, Frame::Body(_)); + found.push(frame); + if last { + return Ok(found); + } + } + } + + fn body(b: &[u8]) -> Frame { + Frame::Body(b.to_vec()) + } + + #[test] + fn frames_round_trip() { + let data = frames(&[b"one", b"", b"three"]); + assert_eq!( + read_all(&data).unwrap(), + [body(b"one"), body(b""), body(b"three"), Frame::End] + ); + } + + #[test] + fn every_cut_of_the_last_frame_reads_as_torn() { + let data = frames(&[b"first", b"second"]); + let first_len = frames(&[b"first"]).len(); + for cut in first_len + 1..data.len() { + assert_eq!( + read_all(&data[..cut]).unwrap(), + [body(b"first"), Frame::Torn], + "cut at {cut}" + ); + } + } + + #[test] + fn a_zeroed_tail_reads_as_torn() { + let mut data = frames(&[b"first"]); + data.extend_from_slice(&[0; 100]); + assert_eq!(read_all(&data).unwrap(), [body(b"first"), Frame::Torn]); + } + + #[test] + fn a_damaged_length_is_an_error_not_a_torn_tail() { + // a length that claims more than the file holds must not cut off + // the frames after it + let mut data = frames(&[b"first", b"second", b"third"]); + let second = frames(&[b"first"]).len(); + data[second + 3] = 0x7f; + assert!(read_all(&data).is_err()); + } + + #[test] + fn zeros_followed_by_data_are_an_error() { + let mut data = frames(&[b"first"]); + data.extend_from_slice(&[0; 8]); + data.extend_from_slice(&frames(&[b"second"])); + assert!(read_all(&data).is_err()); + } + + #[test] + fn a_damaged_body_fails_its_crc() { + let mut data = frames(&[b"first", b"second"]); + let last = data.len() - 5; + data[last] ^= 1; + assert!(matches!( + read_all(&data), + Err(FormatError::ChecksumMismatch { .. }) + )); + } + + #[test] + fn header_round_trips() { + let mut out = Vec::new(); + assert!(!write_header(&mut out, b"TEST", None) + .unwrap() + .is_encrypted()); + assert!(!read_header(&mut out.as_slice(), b"TEST", None) + .unwrap() + .is_encrypted()); + assert!(matches!( + read_header(&mut out.as_slice(), b"ELSE", None), + Err(FormatError::InvalidMagic) + )); + out[4] = VERSION - 1; + assert!(matches!( + read_header(&mut out.as_slice(), b"TEST", None), + Err(FormatError::UnsupportedVersion(_)) + )); + } + + #[cfg(feature = "encryption")] + #[test] + fn encrypted_bodies_round_trip_and_need_the_key() { + let key = Key::from_bytes([7; 32]); + let mut out = Vec::new(); + let codec = write_header(&mut out, b"TEST", Some(&key)).unwrap(); + let sealed = codec.seal(b"secret").unwrap().into_owned(); + assert_ne!(sealed, b"secret"); + + let reopened = read_header(&mut out.as_slice(), b"TEST", Some(&key)).unwrap(); + assert_eq!(reopened.open(sealed.clone()).unwrap(), b"secret"); + assert!(matches!( + read_header(&mut out.as_slice(), b"TEST", None), + Err(FormatError::EncryptionRequired) + )); + let wrong = Key::from_bytes([8; 32]); + let other = read_header(&mut out.as_slice(), b"TEST", Some(&wrong)).unwrap(); + assert!(matches!( + other.open(sealed), + Err(FormatError::DecryptionFailed) + )); + } +} diff --git a/crates/ember-persistence/src/lib.rs b/crates/ember-persistence/src/lib.rs index 379082c1..255c8c6e 100644 --- a/crates/ember-persistence/src/lib.rs +++ b/crates/ember-persistence/src/lib.rs @@ -7,6 +7,7 @@ pub mod aof; #[cfg(feature = "encryption")] pub mod encryption; pub mod format; +pub mod frame; pub mod manifest; pub mod recovery; pub mod snapshot; diff --git a/crates/ember-persistence/src/recovery.rs b/crates/ember-persistence/src/recovery.rs index 5b4c4e11..4dda4053 100644 --- a/crates/ember-persistence/src/recovery.rs +++ b/crates/ember-persistence/src/recovery.rs @@ -35,16 +35,9 @@ use tracing::warn; use crate::aof::{self, AofReader, AofRecord}; use crate::format::FormatError; +use crate::frame::Key; use crate::snapshot::{self, SnapValue, SnapshotReader}; -/// Type alias for an optional encryption key reference. When the -/// `encryption` feature is disabled, this is always `Option<&()>` — -/// always `None` — and all encryption branches compile away. -#[cfg(feature = "encryption")] -type EncryptionKeyRef<'a> = &'a crate::encryption::EncryptionKey; -#[cfg(not(feature = "encryption"))] -type EncryptionKeyRef<'a> = &'a (); - /// Receives the state that recovery reads back. pub trait Recover { /// Stores a key from the snapshot. `ttl` is the time it has left. @@ -106,7 +99,7 @@ pub fn recover_shard_encrypted( fn recover_shard_impl( data_dir: &Path, shard_id: u16, - #[allow(unused_variables)] key: Option>, + key: Option<&Key>, target: &mut impl Recover, ) -> Result { let mut result = RecoveryResult::default(); @@ -158,16 +151,9 @@ type LoadedEntry = (String, SnapValue, Option); fn load_snapshot( path: &Path, expected_shard_id: u16, - #[allow(unused_variables)] encryption_key: Option>, + key: Option<&Key>, ) -> Result<(Vec, u32), FormatError> { - #[cfg(feature = "encryption")] - let mut reader = if let Some(key) = encryption_key { - SnapshotReader::open_encrypted(path, key.clone())? - } else { - SnapshotReader::open(path)? - }; - #[cfg(not(feature = "encryption"))] - let mut reader = SnapshotReader::open(path)?; + let mut reader = SnapshotReader::open_with(path, key)?; if reader.shard_id != expected_shard_id { return Err(FormatError::InvalidData(format!( @@ -176,21 +162,14 @@ fn load_snapshot( ))); } - // entries store the TTL left when the snapshot was written. the file's - // modification time is that moment, so the time since then has passed - // for every key too. - let written_ago_ms = std::fs::metadata(path) - .and_then(|m| m.modified()) - .ok() - .and_then(|t| t.elapsed().ok()) - .map_or(0, |d| d.as_millis().min(i64::MAX as u128) as i64); - let mut entries = Vec::new(); while let Some(entry) = reader.read_entry()? { - let ttl = match entry.expire_ms { - ms if ms < 0 => None, - ms if ms <= written_ago_ms => continue, - ms => Some(Duration::from_millis((ms - written_ago_ms) as u64)), + let ttl = match entry.expires_at_ms { + 0 => None, + at => match aof::ms_until(at) { + Some(ms) => Some(Duration::from_millis(ms)), + None => continue, // expired since the snapshot was written + }, }; entries.push((entry.key, entry.value, ttl)); } @@ -214,7 +193,7 @@ struct AofScan { fn scan_aof( path: &Path, snapshot_crc: Option, - key: Option>, + key: Option<&Key>, ) -> Result { let mut reader = open_aof(path, key)?; let (mut index, mut start, mut stale, mut anchored) = (0, 0, false, false); @@ -248,7 +227,7 @@ fn scan_aof( fn replay_aof( path: &Path, start: usize, - key: Option>, + key: Option<&Key>, target: &mut impl Recover, ) -> Result { let mut reader = open_aof(path, key)?; @@ -276,15 +255,8 @@ fn always_replayed(#[allow(unused_variables)] record: &AofRecord) -> bool { false } -fn open_aof( - path: &Path, - #[allow(unused_variables)] key: Option>, -) -> Result { - #[cfg(feature = "encryption")] - if let Some(key) = key { - return AofReader::open_encrypted(path, key.clone()); - } - AofReader::open(path) +fn open_aof(path: &Path, key: Option<&Key>) -> Result { + AofReader::open_with(path, key) } /// Cuts off a partial record left at the end of the AOF by a crash. @@ -356,15 +328,16 @@ mod tests { .collect() } - /// Writes a snapshot of shard 0 holding `keys` and returns its CRC. - fn write_snapshot(dir: &Path, keys: &[(&str, i64)]) -> u32 { + /// Writes a snapshot of shard 0 holding `keys`, each with the unix ms + /// time it expires at (0 for never), and returns its CRC. + fn write_snapshot(dir: &Path, keys: &[(&str, u64)]) -> u32 { let mut writer = SnapshotWriter::create(snapshot::snapshot_path(dir, 0), 0).unwrap(); - for &(key, expire_ms) in keys { + for &(key, expires_at_ms) in keys { writer .write_entry(&SnapEntry { key: key.into(), value: SnapValue::String(Bytes::from("v")), - expire_ms, + expires_at_ms, }) .unwrap(); } @@ -405,7 +378,7 @@ mod tests { #[test] fn snapshot_then_aof() { let dir = temp_dir(); - write_snapshot(dir.path(), &[("a", -1), ("b", 60_000)]); + write_snapshot(dir.path(), &[("a", 0), ("b", aof::unix_now_ms() + 60_000)]); write_aof(dir.path(), &[del("a")]); let (target, result) = recover(dir.path()); @@ -419,20 +392,25 @@ mod tests { } #[test] - fn snapshot_ttls_count_the_time_since_it_was_written() { + fn keys_that_expired_since_the_snapshot_are_left_out() { + // the expiry is a point in time, so a snapshot restored from a + // backup, whatever its file time, can't bring back expired keys let dir = temp_dir(); - write_snapshot(dir.path(), &[("short", 60_000), ("forever", -1)]); - // the snapshot was taken an hour ago - let hour_ago = std::time::SystemTime::now() - Duration::from_secs(3600); - std::fs::File::options() - .write(true) - .open(snapshot::snapshot_path(dir.path(), 0)) - .unwrap() - .set_modified(hour_ago) - .unwrap(); + let now = aof::unix_now_ms(); + write_snapshot( + dir.path(), + &[ + ("gone", now - 1_000), + ("forever", 0), + ("later", now + 60_000), + ], + ); let (target, _) = recover(dir.path()); - assert_eq!(target.restored, [("forever".to_string(), None)]); + let keys: Vec<_> = target.restored.iter().map(|(k, _)| k.as_str()).collect(); + assert_eq!(keys, ["forever", "later"]); + let later_ttl = target.restored[1].1.unwrap(); + assert!(later_ttl > Duration::from_secs(50) && later_ttl <= Duration::from_secs(60)); } #[test] @@ -509,6 +487,43 @@ mod tests { assert_eq!(recover(dir.path()).0.applied, [del("a"), del("c")]); } + #[test] + fn a_zeroed_tail_is_cut_off() { + // some filesystems leave zeros where a crash cut off a write + use std::io::Write; + + let dir = temp_dir(); + write_aof(dir.path(), &[del("a")]); + std::fs::OpenOptions::new() + .append(true) + .open(aof::aof_path(dir.path(), 0)) + .unwrap() + .write_all(&[0; 64]) + .unwrap(); + + assert_eq!(recover(dir.path()).0.applied, [del("a")]); + write_aof(dir.path(), &[del("b")]); + assert_eq!(recover(dir.path()).0.applied, [del("a"), del("b")]); + } + + #[test] + fn a_damaged_length_mid_file_fails_recovery() { + // a length claiming more than the file holds looks like a torn + // tail. its check copy tells it apart, so the records after it + // aren't cut off + let dir = temp_dir(); + let ends = write_aof(dir.path(), &[del("a"), del("b"), del("c")]); + let path = aof::aof_path(dir.path(), 0); + let mut bytes = std::fs::read(&path).unwrap(); + bytes[ends[0] as usize + 3] = 0x7f; // the second record's length + std::fs::write(&path, &bytes).unwrap(); + + let mut target = Recorded::default(); + let err = recover_shard(dir.path(), 0, &mut target).unwrap_err(); + assert!(matches!(err, RecoveryError::Aof { .. }), "{err}"); + assert_eq!(std::fs::read(&path).unwrap(), bytes); + } + #[test] fn aof_after_its_snapshot_is_replayed() { let dir = temp_dir(); diff --git a/crates/ember-persistence/src/snapshot.rs b/crates/ember-persistence/src/snapshot.rs index 8fd4068e..48749823 100644 --- a/crates/ember-persistence/src/snapshot.rs +++ b/crates/ember-persistence/src/snapshot.rs @@ -5,24 +5,25 @@ //! file first and are atomically renamed on completion — this ensures //! a partial/crashed snapshot never corrupts the existing `.snap` file. //! -//! File layout: +//! File layout, with the header and frames described in [`crate::frame`]: //! ```text -//! [ESNP magic: 4B][version: 1B][shard_id: 2B][entry_count: 4B] -//! [entries...] -//! [footer_crc32: 4B] +//! [header][shard_id: 2B] +//! [one frame per entry] +//! [an empty frame][entry_count: 4B][crc32 of the entry frame bodies: 4B] //! ``` //! -//! Each entry (v2, type-tagged): +//! Each entry: //! ```text -//! [key_len: 4B][key][type_tag: 1B][type-specific payload][expire_ms: 8B] +//! [key_len: 4B][key][type_tag: 1B][type-specific payload][expires_at_ms: 8B] //! ``` //! //! Type tags: 0=string, 1=list, 2=sorted set, 3=hash, 4=set, 5=protobuf, -//! 6=vector. -//! `expire_ms` is the TTL left when the snapshot was written, in -//! milliseconds, or -1 for no expiry. Recovery subtracts the time since the -//! file was last modified. -//! v1 entries (no type tag) are still readable for backward compatibility. +//! 6=vector. `expires_at_ms` is the unix time in milliseconds at which the +//! key expires, or 0 for no expiry, so a snapshot restored later doesn't +//! bring back keys that expired since. +//! +//! The trailing CRC identifies the snapshot: the AOF's checkpoint record +//! names the snapshot it follows by it. use std::collections::{HashMap, HashSet, VecDeque}; use std::fs::{self, File, OpenOptions}; @@ -32,6 +33,7 @@ use std::path::{Path, PathBuf}; use bytes::Bytes; use crate::format::{self, FormatError}; +use crate::frame::{self, Codec, Frame, Key}; /// Type tags for snapshot entries. const TYPE_STRING: u8 = 0; @@ -207,8 +209,8 @@ pub enum SnapValue { pub struct SnapEntry { pub key: String, pub value: SnapValue, - /// Remaining TTL in milliseconds, or -1 for no expiration. - pub expire_ms: i64, + /// Unix time in milliseconds at which the key expires, or 0 for none. + pub expires_at_ms: u64, } impl SnapEntry { @@ -252,7 +254,7 @@ impl SnapEntry { 1 + LEN_PREFIX + type_name.len() + LEN_PREFIX + data.len() } }; - // key + value + expire_ms (i64 = 8 bytes) + // key + value + expires_at_ms key_size + value_size + 8 } } @@ -339,29 +341,31 @@ fn write_snap_value(buf: &mut Vec, value: &SnapValue) -> Result<(), FormatEr Ok(()) } -/// Encodes an entry as `[key][type_tag][payload][expire_ms]`. +/// Encodes an entry as `[key][type_tag][payload][expires_at_ms]`. fn encode_entry(entry: &SnapEntry) -> Result, FormatError> { let mut buf = Vec::with_capacity(entry.estimated_size()); format::write_bytes(&mut buf, entry.key.as_bytes())?; write_snap_value(&mut buf, &entry.value)?; - format::write_i64(&mut buf, entry.expire_ms)?; + buf.extend_from_slice(&entry.expires_at_ms.to_le_bytes()); Ok(buf) } -/// Decodes an entry written by [`encode_entry`]. Format version 1 has no -/// type tag, since every value was a string. -fn decode_entry(r: &mut impl io::Read, version: u8) -> Result { - let key = read_snap_string(r, "key")?; - let value = if version == 1 { - SnapValue::String(Bytes::from(format::read_bytes(r)?)) - } else { - parse_snap_value(r)? - }; - let expire_ms = format::read_i64(r)?; +/// Decodes an entry written by [`encode_entry`]. +fn decode_entry(data: &[u8]) -> Result { + let mut r = io::Cursor::new(data); + let key = read_snap_string(&mut r, "key")?; + let value = parse_snap_value(&mut r)?; + let mut expires_at_ms = [0u8; 8]; + format::read_exact(&mut r, &mut expires_at_ms)?; + if r.position() != data.len() as u64 { + return Err(FormatError::InvalidData( + "snapshot entry has trailing bytes".into(), + )); + } Ok(SnapEntry { key, value, - expire_ms, + expires_at_ms: u64::from_le_bytes(expires_at_ms), }) } @@ -371,82 +375,153 @@ pub fn deserialize_snap_value(data: &[u8]) -> Result { parse_snap_value(&mut cursor) } +/// Writes a snapshot to any output: to a file through [`SnapshotWriter`], +/// or to memory for a replica. +struct Encoder { + out: W, + codec: Codec, + /// Running CRC over the entry frame bodies, for the trailer. + hasher: crc32fast::Hasher, + count: u32, + frame: Vec, +} + +impl Encoder { + fn new(mut out: W, shard_id: u16, key: Option<&Key>) -> Result { + let mut header = Vec::new(); + let codec = frame::write_header(&mut header, format::SNAP_MAGIC, key)?; + header.extend_from_slice(&shard_id.to_le_bytes()); + out.write_all(&header)?; + Ok(Self { + out, + codec, + hasher: crc32fast::Hasher::new(), + count: 0, + frame: Vec::new(), + }) + } + + fn write_entry(&mut self, entry: &SnapEntry) -> Result<(), FormatError> { + let plain = encode_entry(entry)?; + let body = self.codec.seal(&plain)?; + self.hasher.update(&body); + self.frame.clear(); + frame::write_frame(&mut self.frame, &body)?; + self.out.write_all(&self.frame)?; + self.count = self.count.checked_add(1).ok_or_else(|| { + FormatError::InvalidData("snapshot has more than u32::MAX entries".into()) + })?; + Ok(()) + } + + /// Writes the trailer and returns the output and the snapshot's CRC. + fn finish(mut self) -> Result<(W, u32), FormatError> { + let crc = self.hasher.finalize(); + let mut trailer = Vec::new(); + frame::write_frame(&mut trailer, &[])?; + trailer.extend_from_slice(&self.count.to_le_bytes()); + trailer.extend_from_slice(&crc.to_le_bytes()); + self.out.write_all(&trailer)?; + Ok((self.out, crc)) + } +} + +/// Reads a snapshot written by [`Encoder`]. +struct Decoder { + input: R, + codec: Codec, + shard_id: u16, + hasher: crc32fast::Hasher, + count: u32, + /// The CRC, once the trailer has been read and checked. + crc: Option, +} + +impl Decoder { + fn new(mut input: R, key: Option<&Key>) -> Result { + let codec = frame::read_header(&mut input, format::SNAP_MAGIC, key)?; + let shard_id = format::read_u16(&mut input)?; + Ok(Self { + input, + codec, + shard_id, + hasher: crc32fast::Hasher::new(), + count: 0, + crc: None, + }) + } + + /// Reads the next entry, or `None` after the last one, once the + /// trailer has checked out. + fn read_entry(&mut self) -> Result, FormatError> { + if self.crc.is_some() { + return Ok(None); + } + match frame::read_frame(&mut self.input)? { + // a snapshot is renamed into place only once it is complete + Frame::End | Frame::Torn => Err(FormatError::UnexpectedEof), + Frame::Body(body) if body.is_empty() => { + let count = format::read_u32(&mut self.input)?; + let stored = format::read_u32(&mut self.input)?; + if count != self.count { + return Err(FormatError::InvalidData(format!( + "snapshot holds {} entries but its trailer says {count}", + self.count + ))); + } + let crc = self.hasher.clone().finalize(); + format::verify_crc32_values(crc, stored)?; + self.crc = Some(crc); + Ok(None) + } + Frame::Body(body) => { + self.hasher.update(&body); + self.count += 1; + Ok(Some(decode_entry(&self.codec.open(body)?)?)) + } + } + } + + /// Reads any entries left and returns the snapshot's CRC. + fn finish(mut self) -> Result { + while self.read_entry()?.is_some() {} + Ok(self.crc.expect("set when read_entry returns None")) + } +} + /// Writes a complete snapshot to disk. /// /// Entries are written to a temporary file first, then atomically -/// renamed to the final path. The caller provides an iterator over -/// the entries to write. +/// renamed to the final path by [`finish`](Self::finish). pub struct SnapshotWriter { final_path: PathBuf, tmp_path: PathBuf, - writer: BufWriter, - /// Running CRC over all entry bytes for the footer checksum. - hasher: crc32fast::Hasher, - count: u32, - /// Set to `true` after a successful `finish()`. Used by the `Drop` - /// impl to clean up incomplete temp files. - finished: bool, - #[cfg(feature = "encryption")] - cipher: Option, + /// Taken by `finish()`. Until then, `Drop` removes the temp file. + encoder: Option>>, } impl SnapshotWriter { /// Creates a new snapshot writer. The file won't appear at `path` /// until [`Self::finish`] is called successfully. pub fn create(path: impl Into, shard_id: u16) -> Result { - let final_path = path.into(); - let (tmp_path, writer) = Self::open_tmp(&final_path)?; - let mut writer = BufWriter::new(writer); - - format::write_header(&mut writer, format::SNAP_MAGIC)?; - format::write_u16(&mut writer, shard_id)?; - format::write_u32(&mut writer, 0)?; - - Ok(Self { - final_path, - tmp_path, - writer, - hasher: crc32fast::Hasher::new(), - count: 0, - finished: false, - #[cfg(feature = "encryption")] - cipher: None, - }) + Self::create_with(path.into(), shard_id, None) } - /// Creates a new encrypted snapshot writer. + /// Creates a new snapshot writer that encrypts with `key`. #[cfg(feature = "encryption")] pub fn create_encrypted( path: impl Into, shard_id: u16, - key: crate::encryption::EncryptionKey, + key: Key, ) -> Result { - let final_path = path.into(); - let (tmp_path, file) = Self::open_tmp(&final_path)?; - let mut writer = BufWriter::new(file); - - format::write_header_versioned( - &mut writer, - format::SNAP_MAGIC, - format::FORMAT_VERSION_ENCRYPTED, - )?; - format::write_u16(&mut writer, shard_id)?; - format::write_u32(&mut writer, 0)?; - let cipher = key.write_new_file_salt(&mut writer)?; - - Ok(Self { - final_path, - tmp_path, - writer, - hasher: crc32fast::Hasher::new(), - count: 0, - finished: false, - cipher: Some(cipher), - }) + Self::create_with(path.into(), shard_id, Some(&key)) } - /// Opens the temp file for writing. - fn open_tmp(final_path: &Path) -> Result<(PathBuf, File), FormatError> { + fn create_with( + final_path: PathBuf, + shard_id: u16, + key: Option<&Key>, + ) -> Result { let tmp_path = final_path.with_extension("snap.tmp"); let mut opts = OpenOptions::new(); opts.write(true).create(true).truncate(true); @@ -456,80 +531,41 @@ impl SnapshotWriter { opts.mode(0o600); } let file = opts.open(&tmp_path)?; - Ok((tmp_path, file)) + let encoder = Encoder::new(BufWriter::new(file), shard_id, key)?; + Ok(Self { + final_path, + tmp_path, + encoder: Some(encoder), + }) } /// Writes a single entry to the snapshot. - /// - /// When encrypted, each entry is written as `[nonce: 12B][len: 4B][ciphertext]`. - /// The footer CRC covers the encrypted bytes (nonce + len + ciphertext). pub fn write_entry(&mut self, entry: &SnapEntry) -> Result<(), FormatError> { - let buf = encode_entry(entry)?; - - #[cfg(feature = "encryption")] - if let Some(ref cipher) = self.cipher { - let (nonce, ciphertext) = cipher.encrypt(&buf)?; - let ct_len = u32::try_from(ciphertext.len()).map_err(|_| { - io::Error::new( - io::ErrorKind::InvalidInput, - "encrypted record exceeds u32::MAX bytes", - ) - })?; - // footer CRC covers the encrypted envelope - self.hasher.update(&nonce); - let ct_len_bytes = ct_len.to_le_bytes(); - self.hasher.update(&ct_len_bytes); - self.hasher.update(&ciphertext); - self.writer.write_all(&nonce)?; - format::write_u32(&mut self.writer, ct_len)?; - self.writer.write_all(&ciphertext)?; - self.count += 1; - return Ok(()); - } - - self.hasher.update(&buf); - self.writer.write_all(&buf)?; - self.count += 1; - Ok(()) + self.encoder + .as_mut() + .expect("taken only by finish") + .write_entry(entry) } - /// Finalizes the snapshot: writes the footer CRC, flushes, and - /// atomically renames the temp file to the final path. Returns the - /// footer CRC, which identifies this snapshot. + /// Finalizes the snapshot: writes the trailer, fsyncs, and atomically + /// renames the temp file to the final path. Returns the snapshot's CRC, + /// which identifies it. pub fn finish(mut self) -> Result { - // write footer CRC — clone the hasher so we don't move out of self - let checksum = self.hasher.clone().finalize(); - format::write_u32(&mut self.writer, checksum)?; - self.writer.flush()?; - self.writer.get_ref().sync_all()?; - - // rewrite the header with the correct entry count. - // open a second handle for the seek — the BufWriter is already - // flushed and synced above. - { - use std::io::{Seek, SeekFrom}; - let mut file = fs::OpenOptions::new().write(true).open(&self.tmp_path)?; - // header: 4 (magic) + 1 (version) + 2 (shard_id) = 7 bytes. - // an encrypted file's salt comes after the count. - file.seek(SeekFrom::Start(7))?; - format::write_u32(&mut file, self.count)?; - file.sync_all()?; - } + let encoder = self.encoder.take().expect("taken only by finish"); + let (out, crc) = encoder.finish()?; + let file = out.into_inner().map_err(|e| e.into_error())?; + file.sync_all()?; - // atomic rename fs::rename(&self.tmp_path, &self.final_path)?; - // fsync the parent directory so the rename is durable across crashes crate::aof::sync_parent_dir(&self.final_path); - - self.finished = true; - Ok(checksum) + Ok(crc) } } impl Drop for SnapshotWriter { fn drop(&mut self) { - if !self.finished { + if self.encoder.is_some() { // best-effort cleanup of incomplete temp file let _ = fs::remove_file(&self.tmp_path); } @@ -538,224 +574,64 @@ impl Drop for SnapshotWriter { /// Reads entries from a snapshot file. pub struct SnapshotReader { - reader: BufReader, + decoder: Decoder>, pub shard_id: u16, - pub entry_count: u32, - read_so_far: u32, - hasher: crc32fast::Hasher, - /// Format version — v1 has no type tags, v2 has type-tagged entries, - /// v3 and v4 are encrypted. - version: u8, - /// Set for an encrypted file. - #[cfg(feature = "encryption")] - cipher: Option, } impl SnapshotReader { - /// Opens a snapshot file and reads the header. + /// Opens a plaintext snapshot file and reads the header. pub fn open(path: impl AsRef) -> Result { - let file = File::open(path.as_ref())?; - let mut reader = BufReader::new(file); - - let version = format::read_header(&mut reader, format::SNAP_MAGIC)?; - - if format::is_encrypted(version) { - return Err(FormatError::EncryptionRequired); - } - - let shard_id = format::read_u16(&mut reader)?; - let entry_count = format::read_u32(&mut reader)?; - - Ok(Self { - reader, - shard_id, - entry_count, - read_so_far: 0, - hasher: crc32fast::Hasher::new(), - version, - #[cfg(feature = "encryption")] - cipher: None, - }) + Self::open_with(path.as_ref(), None) } - /// Opens a snapshot file with an encryption key for decrypting v3/v4 entries. - /// - /// Also handles v1/v2 (plaintext) files — the key is simply unused. + /// Opens a snapshot file that may be encrypted with `key`. Plaintext + /// files are read too. #[cfg(feature = "encryption")] - pub fn open_encrypted( - path: impl AsRef, - key: crate::encryption::EncryptionKey, - ) -> Result { - let file = File::open(path.as_ref())?; - let mut reader = BufReader::new(file); - - let version = format::read_header(&mut reader, format::SNAP_MAGIC)?; - let shard_id = format::read_u16(&mut reader)?; - let entry_count = format::read_u32(&mut reader)?; - let cipher = format::is_encrypted(version) - .then(|| key.read_file_cipher(version, &mut reader)) - .transpose()?; + pub fn open_encrypted(path: impl AsRef, key: Key) -> Result { + Self::open_with(path.as_ref(), Some(&key)) + } + pub(crate) fn open_with(path: &Path, key: Option<&Key>) -> Result { + let decoder = Decoder::new(BufReader::new(File::open(path)?), key)?; Ok(Self { - reader, - shard_id, - entry_count, - read_so_far: 0, - hasher: crc32fast::Hasher::new(), - version, - cipher, + shard_id: decoder.shard_id, + decoder, }) } - /// Reads the next entry. Returns `None` when all entries have been read. + /// Reads the next entry. Returns `None` after the last one. pub fn read_entry(&mut self) -> Result, FormatError> { - if self.read_so_far >= self.entry_count { - return Ok(None); - } - - #[cfg(feature = "encryption")] - if format::is_encrypted(self.version) { - return self.read_encrypted_entry(); - } - - self.read_plaintext_entry() - } - - /// Reads a plaintext (v1/v2) entry. - fn read_plaintext_entry(&mut self) -> Result, FormatError> { - // the footer CRC covers the entry bytes as stored - let mut reader = format::CrcReader::new(&mut self.reader); - let entry = decode_entry(&mut reader, self.version)?; - self.hasher.combine(&reader.into_hasher()); - self.read_so_far += 1; - Ok(Some(entry)) + self.decoder.read_entry() } - /// Reads an encrypted (v3/v4) entry: nonce + len + ciphertext. - /// Decrypts to get the same bytes as a plaintext entry, then parses. - #[cfg(feature = "encryption")] - fn read_encrypted_entry(&mut self) -> Result, FormatError> { - use std::io::Read as _; - - let cipher = self - .cipher - .as_ref() - .ok_or(FormatError::EncryptionRequired)?; - - let mut nonce = [0u8; crate::encryption::NONCE_SIZE]; - self.reader - .read_exact(&mut nonce) - .map_err(|e| match e.kind() { - io::ErrorKind::UnexpectedEof => FormatError::UnexpectedEof, - _ => FormatError::Io(e), - })?; - - let ct_len = format::read_u32(&mut self.reader)? as usize; - if ct_len > format::MAX_FIELD_LEN { - return Err(FormatError::Io(io::Error::new( - io::ErrorKind::InvalidData, - format!("encrypted entry length {ct_len} exceeds maximum"), - ))); - } - - let mut ciphertext = vec![0u8; ct_len]; - self.reader - .read_exact(&mut ciphertext) - .map_err(|e| match e.kind() { - io::ErrorKind::UnexpectedEof => FormatError::UnexpectedEof, - _ => FormatError::Io(e), - })?; - - // footer CRC covers the encrypted envelope - self.hasher.update(&nonce); - let ct_len_bytes = (ct_len as u32).to_le_bytes(); - self.hasher.update(&ct_len_bytes); - self.hasher.update(&ciphertext); - - let plaintext = cipher.decrypt(&nonce, &ciphertext)?; - let entry = decode_entry(&mut io::Cursor::new(&plaintext), format::FORMAT_VERSION)?; - self.read_so_far += 1; - Ok(Some(entry)) - } - - /// Verifies the footer CRC32 after all entries have been read, and - /// returns it. Must be called after reading all entries. + /// Checks the trailer after the last entry and returns the snapshot's + /// CRC. pub fn verify_footer(self) -> Result { - let expected = self.hasher.finalize(); - let mut reader = self.reader; - let stored = format::read_u32(&mut reader)?; - format::verify_crc32_values(expected, stored)?; - Ok(stored) + self.decoder.finish() } } -/// Serializes snapshot entries to an in-memory byte buffer. -/// -/// The format is identical to the file-based snapshot so the bytes can be -/// sent over the network and loaded with [`read_snapshot_from_bytes`]. -/// Only unencrypted v2 format is produced — encryption is not used for the -/// in-memory replication path. +/// Serializes snapshot entries to bytes, in the file format without +/// encryption, to send a replica its initial state. Load them with +/// [`read_snapshot_from_bytes`]. pub fn write_snapshot_bytes(shard_id: u16, entries: &[SnapEntry]) -> Result, FormatError> { - use std::io::{Seek, SeekFrom, Write as _}; - - let mut buf = io::Cursor::new(Vec::::new()); - let mut hasher = crc32fast::Hasher::new(); - - format::write_header(&mut buf, format::SNAP_MAGIC)?; - format::write_u16(&mut buf, shard_id)?; - // remember where the entry count lives so we can patch it at the end - let count_pos = buf.position(); - format::write_u32(&mut buf, 0u32)?; - - let mut count = 0u32; + let mut encoder = Encoder::new(Vec::new(), shard_id, None)?; for entry in entries { - let entry_bytes = encode_entry(entry)?; - hasher.update(&entry_bytes); - buf.write_all(&entry_bytes)?; - count += 1; + encoder.write_entry(entry)?; } - - // patch entry count in the header - let end_pos = buf.position(); - buf.seek(SeekFrom::Start(count_pos))?; - format::write_u32(&mut buf, count)?; - buf.seek(SeekFrom::Start(end_pos))?; - - // footer CRC - let checksum = hasher.finalize(); - format::write_u32(&mut buf, checksum)?; - - Ok(buf.into_inner()) + Ok(encoder.finish()?.0) } -/// Deserializes snapshot entries from bytes produced by [`write_snapshot_bytes`]. -/// -/// Returns the shard ID from the header and all live entries. Validates the -/// footer CRC to detect transmission errors. +/// Deserializes snapshot entries from bytes produced by +/// [`write_snapshot_bytes`], checking the trailer. Returns the shard ID from +/// the header and all entries. pub fn read_snapshot_from_bytes(data: &[u8]) -> Result<(u16, Vec), FormatError> { - let mut r = io::Cursor::new(data); - let mut hasher = crc32fast::Hasher::new(); - - let version = format::read_header(&mut r, format::SNAP_MAGIC)?; - if version != format::FORMAT_VERSION { - return Err(FormatError::UnsupportedVersion(version)); - } - let shard_id = format::read_u16(&mut r)?; - let entry_count = format::read_u32(&mut r)?; - - let mut entries = Vec::with_capacity(entry_count.min(65536) as usize); - for _ in 0..entry_count { - let mut reader = format::CrcReader::new(&mut r); - entries.push(decode_entry(&mut reader, version)?); - hasher.combine(&reader.into_hasher()); + let mut decoder = Decoder::new(data, None)?; + let mut entries = Vec::new(); + while let Some(entry) = decoder.read_entry()? { + entries.push(entry); } - - // verify footer CRC - let expected = hasher.finalize(); - let stored = format::read_u32(&mut r)?; - format::verify_crc32_values(expected, stored)?; - - Ok((shard_id, entries)) + Ok((decoder.shard_id, entries)) } /// Returns the snapshot file path for a given shard in a data directory. @@ -785,7 +661,6 @@ mod tests { let reader = SnapshotReader::open(&path)?; assert_eq!(reader.shard_id, 0); - assert_eq!(reader.entry_count, 0); reader.verify_footer()?; Ok(()) } @@ -799,17 +674,17 @@ mod tests { SnapEntry { key: "hello".into(), value: SnapValue::String(Bytes::from("world")), - expire_ms: -1, + expires_at_ms: 0, }, SnapEntry { key: "ttl".into(), value: SnapValue::String(Bytes::from("expiring")), - expire_ms: 5000, + expires_at_ms: 5000, }, SnapEntry { key: "empty".into(), value: SnapValue::String(Bytes::new()), - expire_ms: -1, + expires_at_ms: 0, }, ]; @@ -823,7 +698,6 @@ mod tests { let mut reader = SnapshotReader::open(&path)?; assert_eq!(reader.shard_id, 7); - assert_eq!(reader.entry_count, 3); let mut got = Vec::new(); while let Some(entry) = reader.read_entry()? { @@ -844,7 +718,7 @@ mod tests { writer.write_entry(&SnapEntry { key: "k".into(), value: SnapValue::String(Bytes::from("v")), - expire_ms: -1, + expires_at_ms: 0, })?; writer.finish()?; } @@ -875,7 +749,7 @@ mod tests { writer.write_entry(&SnapEntry { key: "original".into(), value: SnapValue::String(Bytes::from("data")), - expire_ms: -1, + expires_at_ms: 0, })?; writer.finish()?; } @@ -886,7 +760,7 @@ mod tests { writer.write_entry(&SnapEntry { key: "new".into(), value: SnapValue::String(Bytes::from("partial")), - expire_ms: -1, + expires_at_ms: 0, })?; // drop without finish — simulates a crash drop(writer); @@ -911,7 +785,7 @@ mod tests { let entry = SnapEntry { key: "expires".into(), value: SnapValue::String(Bytes::from("soon")), - expire_ms: 42_000, + expires_at_ms: 42_000, }; { @@ -922,7 +796,7 @@ mod tests { let mut reader = SnapshotReader::open(&path)?; let got = reader.read_entry()?.unwrap(); - assert_eq!(got.expire_ms, 42_000); + assert_eq!(got.expires_at_ms, 42_000); reader.verify_footer()?; Ok(()) } @@ -941,12 +815,12 @@ mod tests { SnapEntry { key: "mylist".into(), value: SnapValue::List(deque), - expire_ms: -1, + expires_at_ms: 0, }, SnapEntry { key: "mystr".into(), value: SnapValue::String(Bytes::from("val")), - expire_ms: 1000, + expires_at_ms: 1000, }, ]; @@ -981,12 +855,12 @@ mod tests { (200.0, "bob".into()), (150.0, "charlie".into()), ]), - expire_ms: -1, + expires_at_ms: 0, }, SnapEntry { key: "mystr".into(), value: SnapValue::String(Bytes::from("val")), - expire_ms: 1000, + expires_at_ms: 1000, }, ]; @@ -1025,23 +899,23 @@ mod tests { writer.write_entry(&SnapEntry { key: "a".into(), value: SnapValue::String(Bytes::from("1")), - expire_ms: -1, + expires_at_ms: 0, })?; writer.write_entry(&SnapEntry { key: "b".into(), value: SnapValue::String(Bytes::from("2")), - expire_ms: 5000, + expires_at_ms: 5000, })?; writer.finish()?; } - // truncate the file mid-way through the second entry + // truncate the file mid-way through the second entry, which ends + // where the 20-byte trailer starts let data = fs::read(&path)?; - let truncated = &data[..data.len() - 20]; + let truncated = &data[..data.len() - 25]; fs::write(&path, truncated)?; let mut reader = SnapshotReader::open(&path)?; - assert_eq!(reader.entry_count, 2); // first entry should still be readable let first = reader.read_entry()?; @@ -1056,6 +930,26 @@ mod tests { Ok(()) } + #[test] + fn a_snapshot_missing_its_trailer_is_refused() -> Result { + let dir = temp_dir(); + let path = dir.path().join("no-trailer.snap"); + let mut writer = SnapshotWriter::create(&path, 0)?; + writer.write_entry(&SnapEntry { + key: "a".into(), + value: SnapValue::String(Bytes::from("1")), + expires_at_ms: 0, + })?; + writer.finish()?; + let data = fs::read(&path)?; + fs::write(&path, &data[..data.len() - 20])?; + + let mut reader = SnapshotReader::open(&path)?; + assert!(reader.read_entry()?.is_some()); + assert!(reader.read_entry().is_err()); + Ok(()) + } + #[cfg(feature = "vector")] #[test] fn vector_entries_round_trip() -> Result { @@ -1075,7 +969,7 @@ mod tests { ("doc2".into(), vec![0.4, 0.5, 0.6]), ], }, - expire_ms: -1, + expires_at_ms: 0, }]; { @@ -1112,7 +1006,7 @@ mod tests { dim: 128, elements: vec![], }, - expire_ms: 5000, + expires_at_ms: 5000, }]; { @@ -1141,41 +1035,6 @@ mod tests { EncryptionKey::from_bytes([0x42; 32]) } - #[test] - fn v3_snapshot_is_still_read() -> Result { - let dir = temp_dir(); - let path = dir.path().join("v3.snap"); - let key = test_key(); - let entry = SnapEntry { - key: "k".into(), - value: SnapValue::String(Bytes::from("v")), - expire_ms: -1, - }; - - // a v3 file encrypts with the master key and has no salt - let mut file = Vec::new(); - format::write_header_versioned( - &mut file, - format::SNAP_MAGIC, - format::FORMAT_VERSION_ENCRYPTED_V3, - )?; - format::write_u16(&mut file, 3)?; - format::write_u32(&mut file, 1)?; - let (nonce, ciphertext) = key.legacy_cipher().encrypt(&encode_entry(&entry)?)?; - let mut envelope = nonce.to_vec(); - format::write_len(&mut envelope, ciphertext.len())?; - envelope.extend_from_slice(&ciphertext); - file.extend_from_slice(&envelope); - format::write_u32(&mut file, format::crc32(&envelope))?; - fs::write(&path, file)?; - - let mut reader = SnapshotReader::open_encrypted(&path, key)?; - assert_eq!(reader.shard_id, 3); - assert_eq!(reader.read_entry()?, Some(entry)); - reader.verify_footer()?; - Ok(()) - } - #[test] fn encrypted_snapshot_round_trip() -> Result { let dir = temp_dir(); @@ -1186,12 +1045,12 @@ mod tests { SnapEntry { key: "hello".into(), value: SnapValue::String(Bytes::from("world")), - expire_ms: -1, + expires_at_ms: 0, }, SnapEntry { key: "ttl".into(), value: SnapValue::String(Bytes::from("expiring")), - expire_ms: 5000, + expires_at_ms: 5000, }, ]; @@ -1205,7 +1064,6 @@ mod tests { let mut reader = SnapshotReader::open_encrypted(&path, key)?; assert_eq!(reader.shard_id, 7); - assert_eq!(reader.entry_count, 2); let mut got = Vec::new(); while let Some(entry) = reader.read_entry()? { @@ -1228,7 +1086,7 @@ mod tests { writer.write_entry(&SnapEntry { key: "k".into(), value: SnapValue::String(Bytes::from("v")), - expire_ms: -1, + expires_at_ms: 0, })?; writer.finish()?; } @@ -1250,7 +1108,7 @@ mod tests { writer.write_entry(&SnapEntry { key: "k".into(), value: SnapValue::String(Bytes::from("v")), - expire_ms: -1, + expires_at_ms: 0, })?; writer.finish()?; } @@ -1273,7 +1131,7 @@ mod tests { writer.write_entry(&SnapEntry { key: "k".into(), value: SnapValue::String(Bytes::from("v")), - expire_ms: -1, + expires_at_ms: 0, })?; writer.finish()?; } @@ -1304,27 +1162,27 @@ mod tests { SnapEntry { key: "str".into(), value: SnapValue::String(Bytes::from("val")), - expire_ms: -1, + expires_at_ms: 0, }, SnapEntry { key: "list".into(), value: SnapValue::List(deque), - expire_ms: 1000, + expires_at_ms: 1000, }, SnapEntry { key: "zset".into(), value: SnapValue::SortedSet(vec![(1.0, "a".into()), (2.0, "b".into())]), - expire_ms: -1, + expires_at_ms: 0, }, SnapEntry { key: "hash".into(), value: SnapValue::Hash(hash), - expire_ms: -1, + expires_at_ms: 0, }, SnapEntry { key: "set".into(), value: SnapValue::Set(set), - expire_ms: -1, + expires_at_ms: 0, }, ]; diff --git a/crates/ember-server/src/replication.rs b/crates/ember-server/src/replication.rs index 369bdfe3..cff02aba 100644 --- a/crates/ember-server/src/replication.rs +++ b/crates/ember-server/src/replication.rs @@ -691,11 +691,14 @@ impl ReplicationClient { })?; for entry in entries { - // expire_ms is the time left, or negative for no expiry - let ttl_ms = match entry.expire_ms { - ms if ms < 0 => 0, - 0 => continue, // expired while the snapshot was in flight - ms => ms as u64, + // the time the key expires at, or 0 for no expiry. RESTORE + // takes the time left, with 0 for none + let ttl_ms = match entry.expires_at_ms { + 0 => 0, + at => match ember_persistence::aof::ms_until(at) { + Some(ms) => ms, + None => continue, // expired while the snapshot was in flight + }, }; let data = snapshot::serialize_snap_value(&entry.value) .map_err(|e| std::io::Error::other(format!("snapshot value encode failed: {e}")))?; From 24a9a3bd44d79ef3b45e15f5cafa1ef6b83ea5b6 Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Thu, 12 Nov 2026 17:54:31 -0500 Subject: [PATCH 2/3] docs: describe the version 5 file layout --- ARCHITECTURE.md | 2 +- CHANGELOG.md | 4 ++++ docs/troubleshooting.md | 4 ++-- 3 files changed, 7 insertions(+), 3 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index e03e091a..65bf668c 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -171,7 +171,7 @@ persistence is per-shard: - each shard has its own snapshot file - recovery can run shard-by-shard in parallel -the AOF format is binary and append-only. every record has a stable tag plus a CRC32 checksum. snapshots are also binary, include a header and footer checksum, and are written by streaming the live keyspace once. +the AOF and snapshot files share one layout: a header with a version and flags, then one frame per record or entry. a frame holds its length twice, the second time inverted, then the body and a CRC32 of the body. that lets recovery tell a last frame cut short by a crash, which it drops, from damage elsewhere, which stops the start. snapshots store each key's expiry as a unix time, so a snapshot restored later doesn't bring back keys that expired in between. snapshots are written by streaming the live keyspace once. both snapshot writes and AOF rewrites use the usual safe pattern: diff --git a/CHANGELOG.md b/CHANGELOG.md index 3399a4be..8c2bb095 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,10 @@ ## unreleased +### changed +- new layout for AOF and snapshot files, version 5. every record and entry is a frame with its length stored twice and a CRC, so recovery tells a record cut short by a crash from damage in the middle of the file. a header flag marks encryption. files from earlier versions are no longer read +- snapshots store each key's expiry as a unix time. they stored the TTL left and recovery subtracted the time since the file's mtime, so a snapshot restored from a backup brought back keys that had expired + ### removed - concurrent mode: `--concurrent`, `EMBER_CONCURRENT`, the `concurrent` config key and the helm `ember.concurrent` value. the default mode is faster, serving about 35% more requests without pipelining and 2.6× more with a pipeline of 16 on a 32-core machine. concurrent mode also left strings out of the AOF and replication, rejected collection commands, and ran SET NX/XX and INCR as separate read and write steps (#446) - raft and the `openraft` dependency. raft only ever ran on the bootstrap node, and gossip with config epochs already decided slot ownership and failover, as in Redis Cluster. `raft-port-offset` and `--cluster-raft-port-offset` are still accepted and ignored, so existing configs keep working (#443) diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index c92295cc..bc8955f5 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -128,8 +128,8 @@ startup time is proportional to the size of the data being replayed. a large AOF a shard couldn't read one of its files, so the server stopped without changing any of them. the log line names the file and the reason. - `file is encrypted but no encryption key was provided` or `decryption failed`: start with the `encryption-key-file` that wrote the files -- `unsupported format version`: the files need a newer build, or one with the `encryption` feature -- `crc32 mismatch`, `invalid magic bytes` or `unknown record tag`: the file is damaged. restore it from a backup, or move it out of `data-dir` to start without it. moving it out loses the data it holds +- `unsupported format version`: another version of ember wrote the files in a layout this one doesn't read. start that version again, or move the files out of `data-dir` to start without them +- `crc32 mismatch`, `frame length failed its check`, `invalid magic bytes` or `unknown record tag`: the file is damaged. restore it from a backup, or move it out of `data-dir` to start without it. moving it out loses the data it holds --- From c0ba0045e1580a646132e0adbf0eae6688540f1c Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Thu, 12 Nov 2026 18:01:19 -0500 Subject: [PATCH 3/3] docs: link the changelog entries to their PR --- CHANGELOG.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8c2bb095..dd2f6ba0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,8 +3,8 @@ ## unreleased ### changed -- new layout for AOF and snapshot files, version 5. every record and entry is a frame with its length stored twice and a CRC, so recovery tells a record cut short by a crash from damage in the middle of the file. a header flag marks encryption. files from earlier versions are no longer read -- snapshots store each key's expiry as a unix time. they stored the TTL left and recovery subtracted the time since the file's mtime, so a snapshot restored from a backup brought back keys that had expired +- new layout for AOF and snapshot files, version 5. every record and entry is a frame with its length stored twice and a CRC, so recovery tells a record cut short by a crash from damage in the middle of the file. a header flag marks encryption. files from earlier versions are no longer read (#450) +- snapshots store each key's expiry as a unix time. they stored the TTL left and recovery subtracted the time since the file's mtime, so a snapshot restored from a backup brought back keys that had expired (#450) ### removed - concurrent mode: `--concurrent`, `EMBER_CONCURRENT`, the `concurrent` config key and the helm `ember.concurrent` value. the default mode is faster, serving about 35% more requests without pipelining and 2.6× more with a pipeline of 16 on a 32-core machine. concurrent mode also left strings out of the AOF and replication, rejected collection commands, and ran SET NX/XX and INCR as separate read and write steps (#446)