From 40d0a403512633d599ca24aa0b84a075ecab9661 Mon Sep 17 00:00:00 2001 From: Kim Altintop Date: Wed, 8 Jul 2026 10:58:19 +0200 Subject: [PATCH] commitlog: misc fixes (#5493) A few smaller fixes, while searching for a bigger problem. None of them actually fixes the issue at hand. # Testing Adds a test (that passes with or without this patch). --- crates/commitlog/src/commitlog.rs | 4 +- crates/commitlog/src/segment.rs | 29 ++++++++++--- crates/commitlog/tests/random_payload/mod.rs | 45 +++++++++++++++++--- 3 files changed, 67 insertions(+), 11 deletions(-) diff --git a/crates/commitlog/src/commitlog.rs b/crates/commitlog/src/commitlog.rs index dfb7cf3d3ec..9082000dcfd 100644 --- a/crates/commitlog/src/commitlog.rs +++ b/crates/commitlog/src/commitlog.rs @@ -1,6 +1,6 @@ use std::{ fmt::Debug, - io, + io::{self, Seek}, marker::PhantomData, mem, ops::{Range, RangeBounds}, @@ -748,6 +748,8 @@ fn reset_to_internal(repo: &impl Repo, segments: &[u64], offset: u64) -> io::Res } file.ftruncate(offset, byte_offset)?; + // We should be reopening the log anyway, but just in case. + file.seek(io::SeekFrom::End(0))?; // Some filesystems require fsync after ftruncate. file.fsync()?; break; diff --git a/crates/commitlog/src/segment.rs b/crates/commitlog/src/segment.rs index 9f6a4850727..d00be4dd4e5 100644 --- a/crates/commitlog/src/segment.rs +++ b/crates/commitlog/src/segment.rs @@ -498,7 +498,7 @@ pub fn seek_to_offset( debug_assert!(index_key <= start_tx_offset); // Check if the offset index is pointing to the right commit. - let hdr = validate_commit_header(&mut segment, byte_offset)?; + let hdr = validate_commit_at_byte_offset(&mut segment, byte_offset)?; if hdr.min_tx_offset == index_key { // Advance the segment Seek if expected commit is found. segment.seek(SeekFrom::Start(byte_offset)) @@ -513,20 +513,39 @@ pub fn seek_to_offset( /// Try to extract the commit header from the asked position without advancing seek. /// `IndexFileMut` fsync asynchoronously, which makes it important for reader to verify its entry -pub fn validate_commit_header( +fn validate_commit_at_byte_offset( mut reader: &mut Reader, byte_offset: u64, ) -> io::Result { let pos = reader.stream_position()?; reader.seek(SeekFrom::Start(byte_offset))?; - let hdr = commit::Header::decode(&mut reader) - .and_then(|hdr| hdr.ok_or_else(|| io::Error::new(ErrorKind::UnexpectedEof, "unexpected EOF"))); + let hdr_or_error = StoredCommit::decode(&mut reader).and_then(|maybe_commit| { + let StoredCommit { + min_tx_offset, + epoch, + n, + records, + .. + } = maybe_commit.ok_or_else(|| { + io::Error::new( + ErrorKind::UnexpectedEof, + format!("unexpected EOF while validating commit at byte offset {byte_offset}"), + ) + })?; + + Ok(commit::Header { + min_tx_offset, + epoch, + n, + len: records.len() as u32, + }) + }); // Restore the original position reader.seek(SeekFrom::Start(pos))?; - hdr + hdr_or_error } /// Pair of transaction offset and payload. diff --git a/crates/commitlog/tests/random_payload/mod.rs b/crates/commitlog/tests/random_payload/mod.rs index ff149443c13..f6366c99fef 100644 --- a/crates/commitlog/tests/random_payload/mod.rs +++ b/crates/commitlog/tests/random_payload/mod.rs @@ -78,6 +78,37 @@ fn resets() { } } +#[test] +fn reset_then_resume() { + enable_logging(); + + let root = tempdir().unwrap(); + let mut clog = Commitlog::open(CommitLogDir::from_path_unchecked(root.path()), <_>::default(), None).unwrap(); + + let payload = gen_payload(); + for i in 0..10 { + clog.commit([(i, payload)]).unwrap(); + } + clog.flush_and_sync().unwrap(); + + clog = clog.reset_to(4).unwrap(); + + let payload2 = gen_payload(); + for i in 5..10 { + clog.commit([(i, payload2)]).unwrap(); + } + clog.flush_and_sync().unwrap(); + + let txs = clog + .transactions(&payload::ArrayDecoder) + .map(Result::unwrap) + .collect::>(); + + assert_eq!(txs.len(), 10); + assert!(txs[..5].iter().all(|tx| tx.txdata == payload)); + assert!(txs[5..].iter().all(|tx| tx.txdata == payload2)); +} + /// Try to generate commitlogs that will be amenable to compression - /// random data doesn't compress well, so try and have there be repetition fn compressible_payloads() -> impl Iterator { @@ -221,10 +252,14 @@ fn resume_small_trailing_garbage() { } } + let last_segment_offset = { + let segments = repo.existing_offsets().unwrap(); + segments.last().copied().unwrap() + }; + // Add some extra bytes, less than the commit header length. let last_segment_size = { - let segments = repo.existing_offsets().unwrap(); - let mut last_segment = repo.open_segment_writer(segments.last().copied().unwrap()).unwrap(); + let mut last_segment = repo.open_segment_writer(last_segment_offset).unwrap(); last_segment.write_all(&[67u8; commit::Header::LEN - 1]).unwrap(); last_segment.flush().unwrap(); last_segment.sync_all().unwrap(); @@ -232,10 +267,10 @@ fn resume_small_trailing_garbage() { }; { let mut clog = commitlog::Generic::open(&repo, <_>::default()).unwrap(); - + // The segment list should be unchanged. + assert_eq!(&last_segment_offset, repo.existing_offsets().unwrap().last().unwrap()); // The extra bytes should have been truncated away. - let segments = repo.existing_offsets().unwrap(); - let mut last_segment = repo.open_segment_writer(segments.last().copied().unwrap()).unwrap(); + let mut last_segment = repo.open_segment_writer(last_segment_offset).unwrap(); assert_eq!( last_segment.segment_len().unwrap(), last_segment_size - (commit::Header::LEN - 1) as u64