commitlog: Provide byte offset for commit decode errors

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<R: Read>(reader: R) -> io::Result<Option<Self>> {
+    pub fn decode<R: Read + SegmentPos>(reader: &mut R) -> io::Result<Option<Self>> {
         let commit = StoredCommit::decode(reader)?;
         Ok(commit.map(Into::into))
     }
@@ -294,35 +295,52 @@ 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<R: Read>(reader: R) -> io::Result<Option<Self>> {
+    pub fn decode<R: Read + SegmentPos>(reader: &mut R) -> io::Result<Option<Self>> {
         Self::decode_internal(reader, DEFAULT_LOG_FORMAT_VERSION)
     }
 
-    pub(crate) fn decode_internal<R: Read>(reader: R, log_format_version: u8) -> io::Result<Option<Self>> {
+    pub(crate) fn decode_internal<R: Read + SegmentPos>(
+        reader: &mut R,
+        log_format_version: u8,
+    ) -> io::Result<Option<Self>> {
+        let pos = reader.segment_pos()?;
         let mut reader = Crc32cReader::new(reader);
 
         let v = if log_format_version == 0 {
             Version::V0
         } 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<R: io::Read>(reader: R) -> io::Result<Option<Self>> {
+    pub fn extract<R: io::Read + SegmentPos>(reader: &mut R) -> io::Result<Option<Self>> {
         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()

crates/commitlog/src/error.rs

@@ -58,8 +58,10 @@ pub struct Append<T> {
 ///
 /// 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 {

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<u64>;
+}
+
+impl<T: io::Seek> SegmentPos for T {
+    fn segment_pos(&mut self) -> io::Result<u64> {
+        self.stream_position()
+    }
+}
+
 pub trait SegmentReader: io::BufRead + SegmentLen + Send + Sync {
     /// Whether the segment is considered immutable.
     ///

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<R> {
     reader: R,
 }
 
-impl<R: io::BufRead> Iterator for Commits<R> {
+impl<R: io::BufRead + SegmentPos> Iterator for Commits<R> {
     type Item = io::Result<StoredCommit>;
 
     fn next(&mut self) -> Option<Self::Item> {
@@ -580,7 +580,7 @@ impl<R: io::BufRead> Iterator for Commits<R> {
 }
 
 #[cfg(test)]
-impl<R: io::BufRead> Commits<R> {
+impl<R: io::BufRead + SegmentPos> Commits<R> {
     pub fn with_log_format_version(self) -> impl Iterator<Item = io::Result<(u8, StoredCommit)>> {
         CommitsWithVersion { inner: self }
     }
@@ -592,7 +592,7 @@ struct CommitsWithVersion<R> {
 }
 
 #[cfg(test)]
-impl<R: io::BufRead> Iterator for CommitsWithVersion<R> {
+impl<R: io::BufRead + SegmentPos> Iterator for CommitsWithVersion<R> {
     type Item = io::Result<(u8, StoredCommit)>;
 
     fn next(&mut self) -> Option<Self::Item> {
@@ -663,7 +663,7 @@ impl Metadata {
 
         reader.seek(SeekFrom::Start(sofar.size_in_bytes))?;
 
-        fn commit_meta<R: io::Read>(
+        fn commit_meta<R: io::Read + SegmentPos>(
             reader: &mut R,
             sofar: &Metadata,
         ) -> Result<Option<commit::Metadata>, error::SegmentMetadata> {

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,15 +136,42 @@ 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 {
         self.header.len() + self.body.len()
     }
 }
 
+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<usize> {
+        let n = self.inner.read(buf)?;
+        self.pos += n as u64;
+        Ok(n)
+    }
+}
+
+impl SegmentPos for CommitBufReader<'_> {
+    fn segment_pos(&mut self) -> io::Result<u64> {
+        Ok(self.pos)
+    }
+}
+
 pub(super) enum DidReadExact {
     All,
     Eof,

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");