diff --git a/crates/file_store/Cargo.toml b/crates/file_store/Cargo.toml index 8fbdc358de..8e14769b0e 100644 --- a/crates/file_store/Cargo.toml +++ b/crates/file_store/Cargo.toml @@ -16,8 +16,8 @@ workspace = true [dependencies] bdk_core = { path = "../core", version = "0.6.1", features = ["serde"]} -bincode = { version = "1" } -serde = { version = "1", features = ["derive"] } +postcard = { version = "1.1", default-features = false, features = ["use-std"] } +serde = { version = "1", default-features = false, features = ["derive"] } [dev-dependencies] tempfile = "3" diff --git a/crates/file_store/src/entry_iter.rs b/crates/file_store/src/entry_iter.rs index 8b284f1814..1c96d40f60 100644 --- a/crates/file_store/src/entry_iter.rs +++ b/crates/file_store/src/entry_iter.rs @@ -1,18 +1,18 @@ use crate::StoreError; -use bincode::Options; use std::{ fs::File, - io::{self, BufReader, Seek}, + io::{self, BufRead, BufReader, Read, Seek}, marker::PhantomData, }; -use crate::bincode_options; - /// Iterator over entries in a file store. /// /// Reads and returns an entry each time [`next`] is called. If an error occurs while reading the /// iterator will yield a `Result::Err(_)` instead and then `None` for the next call to `next`. /// +/// Each entry is stored as a `postcard`-encoded `u64` varint length prefix followed by that many +/// bytes of `postcard`-encoded data. +/// /// [`next`]: Self::next pub struct EntryIter<'t, T> { /// Buffered reader around the file @@ -44,31 +44,132 @@ where if self.finished { return None; } - (|| { - if let Some(start) = self.start_pos.take() { - self.db_file.seek(io::SeekFrom::Start(start))?; + match self.read_entry() { + Ok(entry) => entry.map(Ok), + Err(e) => { + self.finished = true; + Some(Err(e)) + } + } + } +} + +impl EntryIter<'_, T> +where + T: serde::de::DeserializeOwned, +{ + /// Reads the next entry, or `Ok(None)` on clean end-of-file. + /// + /// On error the file is rewound to the start of the failed entry, so it isn't left mid-entry. + fn read_entry(&mut self) -> Result, StoreError> { + if let Some(start) = self.start_pos.take() { + self.db_file.seek(io::SeekFrom::Start(start))?; + } + let pos_before_read = self.db_file.stream_position()?; + + // An empty buffer here is a clean end-of-file, not a torn entry. Done before the rewind + // scope below because a failed peek consumes nothing. + if self.db_file.fill_buf()?.is_empty() { + return Ok(None); + } + + let entry = self.read_frame(); + if entry.is_err() { + // Leave the file at the start of the failed entry. + self.db_file.seek(io::SeekFrom::Start(pos_before_read))?; + } + entry.map(Some) + } + + /// Reads a single frame. + /// + /// A frame is a `postcard` varint length prefix followed by that many bytes of + /// `postcard`-encoded data. + fn read_frame(&mut self) -> Result { + let len = self.read_len_prefix()?; + let payload_start = self.db_file.stream_position()?; + let payload = self.read_payload(len, payload_start)?; + decode_frame(&payload) + } + + /// Reads the frame length prefix. + /// + /// The varint length prefix is a `postcard`-encoded `u64`, at most 10 bytes, where the high bit + /// of each byte (0x80 mask) is the continuation flag. + fn read_len_prefix(&mut self) -> Result { + let mut buf = [0_u8; 10]; + + for (i, byte) in buf.iter_mut().enumerate() { + if self.db_file.read(std::slice::from_mut(byte))? == 0 { + // Prefix cut short by end-of-file: a torn entry. + return Err(StoreError::Decode( + postcard::Error::DeserializeUnexpectedEnd, + )); } - let pos_before_read = self.db_file.stream_position()?; - match bincode_options().deserialize_from(&mut self.db_file) { - Ok(changeset) => Ok(Some(changeset)), - Err(e) => { - self.finished = true; - let pos_after_read = self.db_file.stream_position()?; - // allow unexpected EOF if 0 bytes were read - if let bincode::ErrorKind::Io(inner) = &*e { - if inner.kind() == io::ErrorKind::UnexpectedEof - && pos_after_read == pos_before_read - { - return Ok(None); - } - } - self.db_file.seek(io::SeekFrom::Start(pos_before_read))?; - Err(StoreError::Bincode(*e)) - } + if *byte & 0x80 == 0 { + return postcard::from_bytes(&buf[..=i]).map_err(StoreError::Decode); } - })() - .transpose() + } + + // Continuation flag still set after 10 bytes: not a valid u64 varint. + Err(StoreError::Decode( + postcard::Error::DeserializeUnexpectedEnd, + )) + } + + /// Reads `len` payload bytes into a fresh buffer. + fn read_payload(&mut self, len: u64, payload_start: u64) -> Result, StoreError> { + let mut payload = Vec::new(); + // Reserve exactly `len` bytes up front. Fail fast on a corrupt, oversized length prefix. + // Avoids unnecessary reads and allocations. + let alloc_failed = match usize::try_from(len) { + Ok(len) => payload.try_reserve_exact(len).is_err(), + Err(_) => true, + }; + if alloc_failed { + return Err(self.alloc_failure_error(len, payload_start)); + } + + let bytes_read = (&mut self.db_file).take(len).read_to_end(&mut payload)?; + if bytes_read as u64 != len { + return Err(StoreError::Decode( + postcard::Error::DeserializeUnexpectedEnd, + )); + } + Ok(payload) + } + + /// Discover the kind of allocation error. + /// + /// This only runs after `len` has already failed to allocate, so it is a big number but not + /// necessarily corrupt. Here we distinguish whether it exceeds the bytes actually remaining + /// in the file (a decode error), or if it fits within the file but exceeds what this + /// machine can allocate right now (an environment failure, not a format one). + fn alloc_failure_error(&self, len: u64, payload_start: u64) -> StoreError { + let remaining = self + .db_file + .get_ref() + .metadata() + .map(|m| m.len().saturating_sub(payload_start)) + .unwrap_or(0); + if len > remaining { + StoreError::Decode(postcard::Error::DeserializeUnexpectedEnd) + } else { + StoreError::Io(io::Error::other("failed to allocate memory for entry")) + } + } +} + +/// Decodes one framed payload. +/// +/// The length prefix stays authoritative for framing, so bytes left over after decoding are +/// corruption, not a format `postcard` has a dedicated variant for. +fn decode_frame(payload: &[u8]) -> Result { + match postcard::take_from_bytes(payload) { + Ok((changeset, [])) => Ok(changeset), + Ok(_) => Err(StoreError::Decode(postcard::Error::SerdeDeCustom)), + Err(e) => Err(StoreError::Decode(e)), } } @@ -81,3 +182,88 @@ impl Drop for EntryIter<'_, T> { } } } + +#[cfg(test)] +#[cfg_attr(coverage_nightly, coverage(off))] +mod test { + use super::*; + + // A single `0x80` byte is a varint continuation flag with no terminating byte: the length + // prefix is cut short by end-of-file, i.e. a torn entry. + fn torn_prefix_file() -> tempfile::NamedTempFile { + let mut file = tempfile::NamedTempFile::new().unwrap(); + std::io::Write::write_all(file.as_file_mut(), &[0x80]).unwrap(); + file + } + + // The iterator yields the error for a torn length prefix, then is fused: subsequent calls to + // `next` return `None` (and the file is rewound to the start of the failed entry). + #[test] + fn next_returns_none_after_error() { + let mut file = torn_prefix_file(); + let mut iter = EntryIter::::new(0, file.as_file_mut()); + + match iter.next() { + Some(Err(StoreError::Decode(postcard::Error::DeserializeUnexpectedEnd))) => {} + unexpected => panic!("unexpected result: {unexpected:?}"), + } + assert_eq!(iter.db_file.stream_position().unwrap(), 0); + // subsequent calls to `next` return `None` + assert!(iter.next().is_none()); + // check twice + assert!(iter.next().is_none()); + } + + // A length prefix cut short by end-of-file is a torn entry, not a clean end-of-file. + #[test] + fn errors_on_truncated_length_prefix() { + let mut file = torn_prefix_file(); + let mut iter = EntryIter::::new(0, file.as_file_mut()); + + match iter.next() { + Some(Err(StoreError::Decode(postcard::Error::DeserializeUnexpectedEnd))) => {} + unexpected => panic!("unexpected result: {unexpected:?}"), + } + } + + // Ten bytes with the continuation flag still set is not a valid `u64` varint. + #[test] + fn errors_on_overlong_length_prefix() { + let mut file = tempfile::NamedTempFile::new().unwrap(); + std::io::Write::write_all(file.as_file_mut(), &[0xFF; 10]).unwrap(); + let mut iter = EntryIter::::new(0, file.as_file_mut()); + + match iter.next() { + Some(Err(StoreError::Decode(postcard::Error::DeserializeUnexpectedEnd))) => {} + unexpected => panic!("unexpected result: {unexpected:?}"), + } + } + + // A length prefix that fits within the file but is too large to allocate is an environment + // failure (`Io`), not a format one (`Decode`). + // + // `try_reserve_exact` allocates the full requested size, so a length larger than the + // machine's available memory fails. A sparse file of 1 TiB makes such a length fit within the + // file while still exceeding what any reasonable machine can allocate. Skipped on filesystems + // that cannot create large sparse files (e.g. tmpfs). + #[test] + fn errors_with_io_when_length_fits_but_allocation_fails() { + let mut file = tempfile::NamedTempFile::new().unwrap(); + let file_size = 1u64 << 40; // 1 TiB + assert!( + file.as_file_mut().set_len(file_size).is_ok(), + "Filesystem can't create a large sparse file (e.g. tmpfs); can't exercise this path." + ); + + // A length that fits within the file but is far too large to allocate. + let len = file_size - 1000; + std::io::Write::write_all(file.as_file_mut(), &postcard::to_allocvec(&len).unwrap()) + .unwrap(); + + let mut iter = EntryIter::::new(0, file.as_file_mut()); + match iter.next() { + Some(Err(StoreError::Io(_))) => {} + unexpected => panic!("unexpected result: {unexpected:?}"), + } + } +} diff --git a/crates/file_store/src/lib.rs b/crates/file_store/src/lib.rs index 3731d50309..0c91fc01ec 100644 --- a/crates/file_store/src/lib.rs +++ b/crates/file_store/src/lib.rs @@ -4,14 +4,9 @@ mod entry_iter; mod store; use std::io; -use bincode::{DefaultOptions, Options}; pub use entry_iter::*; pub use store::*; -pub(crate) fn bincode_options() -> impl bincode::Options { - DefaultOptions::new().with_varint_encoding() -} - /// Error that occurs due to problems encountered with the file. #[derive(Debug)] pub enum StoreError { @@ -19,8 +14,8 @@ pub enum StoreError { Io(io::Error), /// Magic bytes do not match what is expected. InvalidMagicBytes { got: Vec, expected: Vec }, - /// Failure to decode data from the file. - Bincode(bincode::ErrorKind), + /// Failure to decode an entry from the file. + Decode(postcard::Error), } impl core::fmt::Display for StoreError { @@ -34,7 +29,7 @@ impl core::fmt::Display for StoreError { match self { Self::Io(e) => write!(f, "io error while reading store file: {}", e), - Self::Bincode(e) => write!(f, "bincode error while decoding entry {}", e), + Self::Decode(e) => write!(f, "error while decoding store entry: {}", e), Self::InvalidMagicBytes { got, expected } => { write!(f, "invalid magic bytes: ")?; write!(f, "expected 0x")?; diff --git a/crates/file_store/src/store.rs b/crates/file_store/src/store.rs index 858b9d2cdf..f7992aa5a3 100644 --- a/crates/file_store/src/store.rs +++ b/crates/file_store/src/store.rs @@ -1,6 +1,5 @@ -use crate::{bincode_options, EntryIter, StoreError}; +use crate::{EntryIter, StoreError}; use bdk_core::Merge; -use bincode::Options; use std::{ fmt::{self, Debug}, fs::{File, OpenOptions}, @@ -60,7 +59,7 @@ where /// /// If there exist changesets in the file, [`load`] will try to aggregate them in /// a single changeset to verify their integrity. If aggregation fails - /// [`StoreErrorWithDump`] will be returned with the [`StoreError::Bincode`] error variant in + /// [`StoreErrorWithDump`] will be returned with the [`StoreError::Decode`] error variant in /// its error field and the aggregated changeset so far in the changeset field. /// /// To get a new working file store from this error use [`Store::create`] and [`Store::append`] @@ -178,7 +177,7 @@ where /// /// If there exist changesets in the file, [`dump`] will try to aggregate them in a single /// changeset. If aggregation fails [`StoreErrorWithDump`] will be returned with the - /// [`StoreError::Bincode`] error variant in its error field and the aggregated changeset so + /// [`StoreError::Decode`] error variant in its error field and the aggregated changeset so /// far in the changeset field. /// /// [`dump`]: Store::dump @@ -242,12 +241,11 @@ where return Ok(()); } - bincode_options() - .serialize_into(&mut self.db_file, changeset) - .map_err(|e| match *e { - bincode::ErrorKind::Io(error) => error, - unexpected_err => panic!("unexpected bincode error: {unexpected_err}"), - })?; + // Each entry is a `u64` varint length prefix followed by the `postcard`-encoded changeset. + let payload = postcard::to_allocvec(changeset).map_err(io::Error::other)?; + let mut frame = postcard::to_allocvec(&(payload.len() as u64)).map_err(io::Error::other)?; + frame.extend_from_slice(&payload); + self.db_file.write_all(&frame)?; Ok(()) } @@ -369,7 +367,7 @@ mod test { match Store::::load(&TEST_MAGIC_BYTES, file_path) { Err(StoreErrorWithDump { changeset, - error: StoreError::Bincode(_), + error: StoreError::Decode(_), }) => { assert_eq!(changeset, Some(Box::new(test_changesets))) } @@ -397,7 +395,7 @@ mod test { match store.dump() { Err(StoreErrorWithDump { changeset, - error: StoreError::Bincode(_), + error: StoreError::Decode(_), }) => { assert_eq!(changeset, Some(Box::new(test_changesets))) } @@ -474,7 +472,10 @@ mod test { TestChangeSet::from(["4".into(), "5".into(), "6".into()]), ]; let last_changeset = TestChangeSet::from(["7".into(), "8".into(), "9".into()]); - let last_changeset_bytes = bincode_options().serialize(&last_changeset).unwrap(); + let last_changeset_payload = postcard::to_allocvec(&last_changeset).unwrap(); + let mut last_changeset_bytes = + postcard::to_allocvec(&(last_changeset_payload.len() as u64)).unwrap(); + last_changeset_bytes.extend_from_slice(&last_changeset_payload); for short_write_len in 1..last_changeset_bytes.len() - 1 { let file_path = temp_dir.path().join(format!("{short_write_len}.dat")); diff --git a/crates/file_store/tests/test_zero_width.rs b/crates/file_store/tests/test_zero_width.rs new file mode 100644 index 0000000000..1cd5b29ea7 --- /dev/null +++ b/crates/file_store/tests/test_zero_width.rs @@ -0,0 +1,33 @@ +use bdk_file_store::Store; +use std::{sync::mpsc, time::Duration}; + +const MAGIC: &[u8] = b"bdk_test_magic"; + +/// Loads `Store::<()>` from a file containing `contents` and returns whether `load` terminated. +/// +/// `()` is used because its encoding is zero bytes, so decoding it never advances the file offset. +/// The load runs on a detached thread so a hang fails the assertion instead of the whole test run. +fn load_terminates(contents: &[u8]) -> bool { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("db"); + std::fs::write(&path, contents).unwrap(); + + let (tx, rx) = mpsc::channel(); + std::thread::spawn(move || { + let _ = tx.send(Store::<()>::load(MAGIC, &path).is_ok()); + }); + rx.recv_timeout(Duration::from_secs(5)).is_ok() +} + +#[test] +fn load_terminates_on_zero_width_changeset() { + let mut contents = MAGIC.to_vec(); + contents.push(0xff); + assert!(load_terminates(&contents), "load did not terminate"); +} + +// A file with no entries must still load, even for a zero-width changeset type. +#[test] +fn load_terminates_on_magic_only_zero_width_store() { + assert!(load_terminates(MAGIC), "load did not terminate"); +}