diff --git a/crates/commitlog/src/commit.rs b/crates/commitlog/src/commit.rs index d10f55894b6..dfdbc6cc9d3 100644 --- a/crates/commitlog/src/commit.rs +++ b/crates/commitlog/src/commit.rs @@ -9,6 +9,7 @@ use spacetimedb_sats::buffer::{BufReader, Cursor, DecodeError}; use crate::{ error::ChecksumMismatch, payload::Decoder, + repo::SegmentPos, segment::{CHECKSUM_ALGORITHM_CRC32C, CHECKSUM_CRC32C_LEN}, Transaction, DEFAULT_LOG_FORMAT_VERSION, }; @@ -189,7 +190,7 @@ impl Commit { /// [`ChecksumMismatch`] is returned. /// /// To retain access to the checksum, use [`StoredCommit::decode`]. - pub fn decode(reader: R) -> io::Result> { + pub fn decode(reader: &mut R) -> io::Result> { let commit = StoredCommit::decode(reader)?; Ok(commit.map(Into::into)) } @@ -294,11 +295,15 @@ impl StoredCommit { /// Verifies the checksum of the commit. If it doesn't match, an error of /// kind [`io::ErrorKind::InvalidData`] with an inner error downcastable to /// [`ChecksumMismatch`] is returned. - pub fn decode(reader: R) -> io::Result> { + pub fn decode(reader: &mut R) -> io::Result> { Self::decode_internal(reader, DEFAULT_LOG_FORMAT_VERSION) } - pub(crate) fn decode_internal(reader: R, log_format_version: u8) -> io::Result> { + pub(crate) fn decode_internal( + reader: &mut R, + log_format_version: u8, + ) -> io::Result> { + let pos = reader.segment_pos()?; let mut reader = Crc32cReader::new(reader); let v = if log_format_version == 0 { @@ -306,23 +311,36 @@ impl StoredCommit { } else { Version::V1 }; - let Some(hdr) = Header::decode_internal(&mut reader, v)? else { + let Some(hdr) = Header::decode_internal(&mut reader, v).map_err(|e| { + io::Error::new( + e.kind(), + format!("commit at byte position {}: failed to decode commit header: {}", pos, e), + ) + })? + else { return Ok(None); }; let mut records = vec![0; hdr.len as usize]; reader.read_exact(&mut records).map_err(|e| { io::Error::new( e.kind(), - format!("failed to read {} bytes of commit payload: {}", hdr.len, e), + format!( + "commit at byte position {}: failed to read {} bytes of commit payload: {}", + pos, hdr.len, e + ), ) })?; let chk = reader.crc32c(); - let crc = decode_u32(reader.into_inner()) - .map_err(|e| io::Error::new(e.kind(), format!("failed to read checksum: {e}")))?; + let crc = decode_u32(reader.into_inner()).map_err(|e| { + io::Error::new( + e.kind(), + format!("commit at byte position{}: failed to read checksum: {}", pos, e), + ) + })?; if chk != crc { - return Err(invalid_data(ChecksumMismatch)); + return Err(invalid_data(ChecksumMismatch { commit_pos: pos })); } Ok(Some(Self { @@ -364,7 +382,7 @@ impl Metadata { /// Note that this decodes the commit due to checksum verification. /// Like [`StoredCommit::decode`], this method returns `None` if the reader /// is at EOF already. - pub fn extract(reader: R) -> io::Result> { + pub fn extract(reader: &mut R) -> io::Result> { StoredCommit::decode(reader).map(|maybe_commit| maybe_commit.map(Self::from)) } } @@ -423,7 +441,7 @@ mod tests { let mut buf = Vec::with_capacity(commit.encoded_len()); commit.write(&mut buf).unwrap(); - let commit2 = Commit::decode(&mut buf.as_slice()).unwrap().unwrap(); + let commit2 = Commit::decode(&mut io::Cursor::new(buf.as_slice())).unwrap().unwrap(); assert_eq!(commit, commit2); } @@ -476,7 +494,7 @@ mod tests { // so we get `ChecksumMismatch` not any other error. buf[pos] ^= mask.get(); - match Commit::decode(&mut buf.as_slice()) { + match Commit::decode(&mut io::Cursor::new(buf.as_slice())) { Err(e) => { assert_eq!(e.kind(), io::ErrorKind::InvalidData); e.into_inner() diff --git a/crates/commitlog/src/error.rs b/crates/commitlog/src/error.rs index 0d72303b649..018267008be 100644 --- a/crates/commitlog/src/error.rs +++ b/crates/commitlog/src/error.rs @@ -58,8 +58,10 @@ pub struct Append { /// /// Usually wrapped in another error, such as [`io::Error`]. #[derive(Debug, Error)] -#[error("checksum mismatch")] -pub struct ChecksumMismatch; +#[error("commit at byte position {commit_pos}: checksum mismatch")] +pub struct ChecksumMismatch { + pub(crate) commit_pos: u64, +} #[derive(Debug, Error)] pub enum SegmentMetadata { diff --git a/crates/commitlog/src/repo/mod.rs b/crates/commitlog/src/repo/mod.rs index af04ad918bb..49e674c2d4b 100644 --- a/crates/commitlog/src/repo/mod.rs +++ b/crates/commitlog/src/repo/mod.rs @@ -57,6 +57,23 @@ pub trait SegmentLen: io::Seek { } } +/// Trait to obtain the current position in a segment. +/// +/// All types implementing [io::Seek] implement this trait, via +/// [io::Seek::stream_position]. The trait exists so that types that can't +/// easily implement [io::Seek] can still provide position information. +pub trait SegmentPos { + /// Get the current position in the segment, like + /// [io::Seek::stream_position]. + fn segment_pos(&mut self) -> io::Result; +} + +impl SegmentPos for T { + fn segment_pos(&mut self) -> io::Result { + self.stream_position() + } +} + pub trait SegmentReader: io::BufRead + SegmentLen + Send + Sync { /// Whether the segment is considered immutable. /// diff --git a/crates/commitlog/src/segment.rs b/crates/commitlog/src/segment.rs index d00be4dd4e5..b29adf156c6 100644 --- a/crates/commitlog/src/segment.rs +++ b/crates/commitlog/src/segment.rs @@ -12,7 +12,7 @@ use crate::{ error, index::{IndexError, IndexFileMut}, payload::Encode, - repo::{TxOffset, TxOffsetIndex, TxOffsetIndexMut}, + repo::{SegmentPos, TxOffset, TxOffsetIndex, TxOffsetIndexMut}, Options, }; @@ -571,7 +571,7 @@ pub struct Commits { reader: R, } -impl Iterator for Commits { +impl Iterator for Commits { type Item = io::Result; fn next(&mut self) -> Option { @@ -580,7 +580,7 @@ impl Iterator for Commits { } #[cfg(test)] -impl Commits { +impl Commits { pub fn with_log_format_version(self) -> impl Iterator> { CommitsWithVersion { inner: self } } @@ -592,7 +592,7 @@ struct CommitsWithVersion { } #[cfg(test)] -impl Iterator for CommitsWithVersion { +impl Iterator for CommitsWithVersion { type Item = io::Result<(u8, StoredCommit)>; fn next(&mut self) -> Option { @@ -663,7 +663,7 @@ impl Metadata { reader.seek(SeekFrom::Start(sofar.size_in_bytes))?; - fn commit_meta( + fn commit_meta( reader: &mut R, sofar: &Metadata, ) -> Result, error::SegmentMetadata> { diff --git a/crates/commitlog/src/stream/common.rs b/crates/commitlog/src/stream/common.rs index efb71fb42f1..929944e1141 100644 --- a/crates/commitlog/src/stream/common.rs +++ b/crates/commitlog/src/stream/common.rs @@ -6,7 +6,10 @@ use std::{ use tokio::io::{AsyncBufRead, AsyncBufReadExt as _, AsyncRead, AsyncReadExt as _, AsyncSeek, AsyncWrite}; -use crate::{commit, repo::Repo}; +use crate::{ + commit, + repo::{Repo, SegmentPos}, +}; /// How to convert [`crate::repo::SegmentWriter`]s into async I/O types. pub trait IntoAsyncWriter { @@ -133,8 +136,16 @@ impl CommitBuf { bytes::Buf::chain(&self.header[..], &self.body[..]) } - pub fn as_reader(&self) -> impl io::Read + '_ { - io::Read::chain(&self.header[..], &self.body[..]) + /// View the [CommitBuf] as an [io::Read] for decoding. + /// + /// The returned type also implements [SegmentPos] based on the supplied + /// position of the buffer within a segment. This is used for error + /// reporting. + pub fn as_reader(&self, segment_pos: u64) -> impl io::Read + SegmentPos + '_ { + CommitBufReader { + inner: io::Read::chain(&self.header[..], &self.body[..]), + pos: segment_pos, + } } pub fn filled_len(&self) -> usize { @@ -142,6 +153,25 @@ impl CommitBuf { } } +struct CommitBufReader<'a> { + inner: io::Chain<&'a [u8], &'a [u8]>, + pos: u64, +} + +impl io::Read for CommitBufReader<'_> { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + let n = self.inner.read(buf)?; + self.pos += n as u64; + Ok(n) + } +} + +impl SegmentPos for CommitBufReader<'_> { + fn segment_pos(&mut self) -> io::Result { + Ok(self.pos) + } +} + pub(super) enum DidReadExact { All, Eof, diff --git a/crates/commitlog/src/stream/writer.rs b/crates/commitlog/src/stream/writer.rs index afc14a15c46..cd034566604 100644 --- a/crates/commitlog/src/stream/writer.rs +++ b/crates/commitlog/src/stream/writer.rs @@ -313,7 +313,7 @@ where ); stream.read_exact(&mut self.commit_buf.body).await?; // Decode the commit and verify its checksum. - let commit = StoredCommit::decode(self.commit_buf.as_reader()) + let commit = StoredCommit::decode(&mut self.commit_buf.as_reader(bytes_written)) .inspect_err(|e| warn!("failed to decode commit: {e}"))? .expect("commit decode cannot return `None` because we already decoded the header");