Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion crates/file_store/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ workspace = true

[dependencies]
bdk_core = { path = "../core", version = "0.6.1", features = ["serde"]}
bincode = { version = "1" }
postcard = { version = "1.1", default-features = false, features = ["use-std"] }
serde = { version = "1", features = ["derive"] }

[dev-dependencies]
Expand Down
104 changes: 77 additions & 27 deletions crates/file_store/src/entry_iter.rs
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -44,31 +44,81 @@ where
if self.finished {
return None;
}
(|| {
if let Some(start) = self.start_pos.take() {
self.db_file.seek(io::SeekFrom::Start(start))?;
}
let entry = self.read_entry().transpose();
// stop after the end of the file or the first error
if !matches!(entry, Some(Ok(_))) {
self.finished = true;
}
entry
}
}

impl<T> EntryIter<'_, T>
where
T: serde::de::DeserializeOwned,
{
/// Reads the next entry, or `Ok(None)` on a clean end-of-file.
///
/// If the entry cannot be read the file is rewound to where it started, so the position is
/// never left in the middle of an entry.
fn read_entry(&mut self) -> Result<Option<T>, 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()?;

// no bytes left: this is the end of the file, not a torn entry
if self.db_file.fill_buf()?.is_empty() {
return Ok(None);
}

let entry = self.read_frame();
if entry.is_err() {
self.db_file.seek(io::SeekFrom::Start(pos_before_read))?;
}
entry.map(Some)
}

/// Reads one length-prefixed frame and decodes it.
///
/// A frame that ends early (end-of-file) or does not decode is [`StoreError::Decode`]. Any
/// other failure to read is reported as [`StoreError::Io`].
fn read_frame(&mut self) -> Result<T, StoreError> {
let len = self.read_len_prefix()?;

// `take` + `read_to_end` only allocates what the file actually holds, so a corrupt, huge
// length prefix cannot trigger a huge allocation.
let mut payload = Vec::new();
(&mut self.db_file).take(len).read_to_end(&mut payload)?;
if payload.len() as u64 != len {
return Err(StoreError::Decode(
postcard::Error::DeserializeUnexpectedEnd,
));
}

// The length prefix is authoritative, so bytes left over after decoding are corruption.
match postcard::take_from_bytes(&payload) {
Ok((entry, [])) => Ok(entry),
Ok(_) => Err(StoreError::Decode(postcard::Error::SerdeDeCustom)),
Err(e) => Err(StoreError::Decode(e)),
}
}

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))
}
/// Reads the `postcard` varint `u64` that holds the length of the frame's payload.
fn read_len_prefix(&mut self) -> Result<u64, StoreError> {
// a `u64` varint is at most 10 bytes; the high bit of a byte flags that another follows
let mut buf = [0_u8; 10];
for i in 0..buf.len() {
if self.db_file.read(&mut buf[i..=i])? == 0 {
return Err(StoreError::Decode(
postcard::Error::DeserializeUnexpectedEnd,
));
}
if buf[i] & 0x80 == 0 {
return postcard::from_bytes(&buf[..=i]).map_err(StoreError::Decode);
}
})()
.transpose()
}
Err(StoreError::Decode(postcard::Error::DeserializeBadVarint))
}
}

Expand Down
9 changes: 2 additions & 7 deletions crates/file_store/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand All @@ -20,7 +15,7 @@ pub enum StoreError {
/// Magic bytes do not match what is expected.
InvalidMagicBytes { got: Vec<u8>, expected: Vec<u8> },
/// Failure to decode data from the file.
Bincode(bincode::ErrorKind),
Decode(postcard::Error),
}

impl core::fmt::Display for StoreError {
Expand All @@ -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, "decode error while decoding entry: {}", e),
Self::InvalidMagicBytes { got, expected } => {
write!(f, "invalid magic bytes: ")?;
write!(f, "expected 0x")?;
Expand Down
Loading