From 94b2e015e094460d27f658bdd874faf33c743b01 Mon Sep 17 00:00:00 2001 From: Kim Altintop Date: Tue, 25 Aug 2026 15:29:06 +0200 Subject: [PATCH 1/7] commitlog: Provide byte offset for commit decode errors It is often not clear how much data is left in a segment when decoding fails. Fix this by obtaining the `stream_position` at the start of decoding. --- crates/commitlog/src/commit.rs | 38 ++++++++++++++++++------- crates/commitlog/src/repo/mod.rs | 10 +++++++ crates/commitlog/src/segment.rs | 8 +++--- crates/commitlog/src/stream/common.rs | 40 +++++++++++++++++++++++++-- crates/commitlog/src/stream/writer.rs | 2 +- 5 files changed, 80 insertions(+), 18 deletions(-) diff --git a/crates/commitlog/src/commit.rs b/crates/commitlog/src/commit.rs index d10f55894b6..4c32dde28ab 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: 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: 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( + mut reader: 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,20 +311,33 @@ 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)); @@ -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: 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/repo/mod.rs b/crates/commitlog/src/repo/mod.rs index af04ad918bb..48a86922915 100644 --- a/crates/commitlog/src/repo/mod.rs +++ b/crates/commitlog/src/repo/mod.rs @@ -57,6 +57,16 @@ pub trait SegmentLen: io::Seek { } } +pub trait SegmentPos { + 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..adfb673502e 100644 --- a/crates/commitlog/src/segment.rs +++ b/crates/commitlog/src/segment.rs @@ -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..56d69a59a43 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,11 @@ impl CommitBuf { bytes::Buf::chain(&self.header[..], &self.body[..]) } - pub fn as_reader(&self) -> impl io::Read + '_ { - io::Read::chain(&self.header[..], &self.body[..]) + pub fn as_reader(&self, virtual_pos: u64) -> impl io::Read + SegmentPos + '_ { + CommitBufReader { + inner: io::Read::chain(&self.header[..], &self.body[..]), + pos: virtual_pos, + } } pub fn filled_len(&self) -> usize { @@ -142,6 +148,34 @@ impl CommitBuf { } } +struct CommitBufReader<'a> { + inner: io::Chain<&'a [u8], &'a [u8]>, + pos: u64, +} + +impl<'a> CommitBufReader<'a> { + fn new(a: &'a [u8], b: &'a [u8]) -> Self { + Self { + inner: io::Read::chain(a, b), + pos: 0, + } + } +} + +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..d583ac73daf 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(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"); From 6d86e065a8f23cc4874f9b07cffd7c2dce3b78bc Mon Sep 17 00:00:00 2001 From: Kim Altintop Date: Tue, 25 Aug 2026 15:42:44 +0200 Subject: [PATCH 2/7] Include pos in `ChecksumMismatch` --- crates/commitlog/src/commit.rs | 2 +- crates/commitlog/src/error.rs | 6 ++++-- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/crates/commitlog/src/commit.rs b/crates/commitlog/src/commit.rs index 4c32dde28ab..b86f2c32fa0 100644 --- a/crates/commitlog/src/commit.rs +++ b/crates/commitlog/src/commit.rs @@ -340,7 +340,7 @@ impl StoredCommit { })?; if chk != crc { - return Err(invalid_data(ChecksumMismatch)); + return Err(invalid_data(ChecksumMismatch { commit_pos: pos })); } Ok(Some(Self { 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 { From 73c2ef18e721636da576d04d6205032b555c770f Mon Sep 17 00:00:00 2001 From: Kim Altintop Date: Wed, 26 Aug 2026 08:47:46 +0200 Subject: [PATCH 3/7] Lint --- crates/commitlog/src/stream/common.rs | 9 --------- 1 file changed, 9 deletions(-) diff --git a/crates/commitlog/src/stream/common.rs b/crates/commitlog/src/stream/common.rs index 56d69a59a43..e02d8f59528 100644 --- a/crates/commitlog/src/stream/common.rs +++ b/crates/commitlog/src/stream/common.rs @@ -153,15 +153,6 @@ struct CommitBufReader<'a> { pos: u64, } -impl<'a> CommitBufReader<'a> { - fn new(a: &'a [u8], b: &'a [u8]) -> Self { - Self { - inner: io::Read::chain(a, b), - pos: 0, - } - } -} - impl io::Read for CommitBufReader<'_> { fn read(&mut self, buf: &mut [u8]) -> io::Result { let n = self.inner.read(buf)?; From 010e0c9e647a76069fde69b6e40e077e1eff4257 Mon Sep 17 00:00:00 2001 From: Kim Altintop Date: Wed, 26 Aug 2026 08:55:50 +0200 Subject: [PATCH 4/7] Docs --- crates/commitlog/src/repo/mod.rs | 7 +++++++ crates/commitlog/src/stream/common.rs | 9 +++++++-- 2 files changed, 14 insertions(+), 2 deletions(-) diff --git a/crates/commitlog/src/repo/mod.rs b/crates/commitlog/src/repo/mod.rs index 48a86922915..59759bd17c8 100644 --- a/crates/commitlog/src/repo/mod.rs +++ b/crates/commitlog/src/repo/mod.rs @@ -57,7 +57,14 @@ pub trait SegmentLen: io::Seek { } } +/// Trait to obtain the current position in a segment. +/// +/// All type 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; } diff --git a/crates/commitlog/src/stream/common.rs b/crates/commitlog/src/stream/common.rs index e02d8f59528..929944e1141 100644 --- a/crates/commitlog/src/stream/common.rs +++ b/crates/commitlog/src/stream/common.rs @@ -136,10 +136,15 @@ impl CommitBuf { bytes::Buf::chain(&self.header[..], &self.body[..]) } - pub fn as_reader(&self, virtual_pos: u64) -> impl io::Read + SegmentPos + '_ { + /// 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: virtual_pos, + pos: segment_pos, } } From 3f7fa6be06b7c979edb538d1d993e6f2c5a7737f Mon Sep 17 00:00:00 2001 From: Kim Altintop Date: Wed, 26 Aug 2026 08:58:06 +0200 Subject: [PATCH 5/7] Typo --- crates/commitlog/src/repo/mod.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/commitlog/src/repo/mod.rs b/crates/commitlog/src/repo/mod.rs index 59759bd17c8..49e674c2d4b 100644 --- a/crates/commitlog/src/repo/mod.rs +++ b/crates/commitlog/src/repo/mod.rs @@ -59,7 +59,7 @@ pub trait SegmentLen: io::Seek { /// Trait to obtain the current position in a segment. /// -/// All type implementing [io::Seek] implement this trait, via +/// 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 { From 17addc39cc25577e330427a78936239509af9bca Mon Sep 17 00:00:00 2001 From: Kim Altintop Date: Wed, 26 Aug 2026 09:11:38 +0200 Subject: [PATCH 6/7] Rust trait coherence is awkward --- crates/commitlog/src/commit.rs | 8 ++++---- crates/commitlog/src/segment.rs | 8 ++++---- crates/commitlog/src/stream/writer.rs | 2 +- 3 files changed, 9 insertions(+), 9 deletions(-) diff --git a/crates/commitlog/src/commit.rs b/crates/commitlog/src/commit.rs index b86f2c32fa0..dfdbc6cc9d3 100644 --- a/crates/commitlog/src/commit.rs +++ b/crates/commitlog/src/commit.rs @@ -190,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)) } @@ -295,12 +295,12 @@ 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( - mut reader: R, + reader: &mut R, log_format_version: u8, ) -> io::Result> { let pos = reader.segment_pos()?; @@ -382,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)) } } diff --git a/crates/commitlog/src/segment.rs b/crates/commitlog/src/segment.rs index adfb673502e..67cb7be9ac8 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 { diff --git a/crates/commitlog/src/stream/writer.rs b/crates/commitlog/src/stream/writer.rs index d583ac73daf..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(bytes_written)) + 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"); From abee8d0876dc86cf9d87acfe0fd11c2999277749 Mon Sep 17 00:00:00 2001 From: Kim Altintop Date: Wed, 26 Aug 2026 09:16:08 +0200 Subject: [PATCH 7/7] Loosen trait bound --- crates/commitlog/src/segment.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/commitlog/src/segment.rs b/crates/commitlog/src/segment.rs index 67cb7be9ac8..b29adf156c6 100644 --- a/crates/commitlog/src/segment.rs +++ b/crates/commitlog/src/segment.rs @@ -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> {