commitlog: Improve error context

crates/commitlog/src/commitlog.rs

@@ -61,7 +61,25 @@ impl<R: Repo, T> Generic<R, T> {
         }
         let head = if let Some(last) = tail.pop() {
             debug!("resuming last segment: {last}");
+            // Resume the last segment for writing, or create a new segment
+            // starting from the last good commit + 1.
             repo::resume_segment_writer(&repo, opts, last)?.or_else(|meta| {
+                // The first commit in the last segment being corrupt is an
+                // edge case: we'd try to start a new segment with an offset
+                // equal to the already existing one, which would fail.
+                //
+                // We cannot just skip it either, as we don't know the reason
+                // for the corruption (there could be more, potentially
+                // recoverable commits in the segment).
+                //
+                // Thus, provide some context about what is wrong and refuse to
+                // start.
+                if meta.tx_range.is_empty() {
+                    return Err(io::Error::new(
+                        io::ErrorKind::InvalidData,
+                        format!("repo {}: first commit in resumed segment {} is corrupt", repo, last),
+                    ));
+                }
                 tail.push(meta.tx_range.start);
                 repo::create_segment_writer(&repo, opts, meta.max_epoch, meta.tx_range.end)
             })?

crates/commitlog/src/repo/fs.rs

@@ -1,3 +1,4 @@
+use std::fmt;
 use std::fs::{self, File};
 use std::io;
 use std::sync::Arc;
@@ -76,6 +77,12 @@ impl Fs {
     }
 }
 
+impl fmt::Display for Fs {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}", self.root.display())
+    }
+}
+
 impl SegmentLen for File {}
 
 impl FileLike for NamedTempFile {
@@ -101,11 +108,18 @@ impl Repo for Fs {
             .or_else(|e| {
                 if e.kind() == io::ErrorKind::AlreadyExists {
                     debug!("segment {offset} already exists");
+                    // If the segment is completely empty, we can resume writing.
                     let file = self.open_segment_writer(offset)?;
                     if file.metadata()?.len() == 0 {
                         debug!("segment {offset} is empty");
                         return Ok(file);
                     }
+
+                    // Otherwise, provide some context.
+                    return Err(io::Error::new(
+                        io::ErrorKind::AlreadyExists,
+                        format!("repo {}: segment {} already exists and is non-empty", self, offset),
+                    ));
                 }
 
                 Err(e)
@@ -131,7 +145,10 @@ impl Repo for Fs {
 
     fn remove_segment(&self, offset: u64) -> io::Result<()> {
         let _ = self.remove_offset_index(offset).map_err(|e| {
-            warn!("failed to remove offset index for segment {offset}, error: {e}");
+            warn!(
+                "repo {}: failed to remove offset index for segment {}: {}",
+                self, offset, e
+            );
         });
         fs::remove_file(self.segment_path(offset))
     }

crates/commitlog/src/repo/mem.rs

@@ -1,6 +1,6 @@
 use std::{
     collections::{btree_map, BTreeMap},
-    io,
+    fmt, io,
     sync::{Arc, RwLock, RwLockWriteGuard},
 };
 
@@ -213,6 +213,12 @@ impl Memory {
     }
 }
 
+impl fmt::Display for Memory {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str("<memory>")
+    }
+}
+
 impl Repo for Memory {
     type SegmentWriter = Segment;
     type SegmentReader = io::BufReader<Segment>;

crates/commitlog/src/repo/mod.rs

@@ -1,4 +1,4 @@
-use std::io;
+use std::{fmt, io};
 
 use log::{debug, warn};
 
@@ -59,7 +59,10 @@ impl<T: FileLike + io::Read + io::Write + SegmentLen + Send + Sync> SegmentWrite
 ///
 /// This is mainly an internal trait to allow testing against an in-memory
 /// representation.
-pub trait Repo: Clone {
+///
+/// The [fmt::Display] should provide context about the location of the repo,
+/// e.g. the root directory for a filesystem-based implementation.
+pub trait Repo: Clone + fmt::Display {
     /// The type of log segments managed by this repo, which must behave like a file.
     type SegmentWriter: SegmentWriter + 'static;
     type SegmentReader: SegmentReader + 'static;

crates/commitlog/src/tests/partial.rs

@@ -1,18 +1,20 @@
 use std::{
     cmp,
-    fmt::Debug,
+    fmt::{self, Debug},
     io::{self, Seek as _, SeekFrom},
-    iter::repeat,
+    iter::{self, repeat},
+    num::NonZeroU16,
     sync::RwLockWriteGuard,
 };
 
 use log::debug;
+use pretty_assertions::assert_matches;
 
 use crate::{
     commitlog, error, payload,
     repo::{self, Repo, SegmentLen},
-    segment::FileLike,
-    tests::helpers::enable_logging,
+    segment::{self, FileLike},
+    tests::helpers::{enable_logging, fill_log_with},
     Commit, Encode, Options, DEFAULT_LOG_FORMAT_VERSION,
 };
 
@@ -140,6 +142,42 @@ fn overwrite_reopen() {
     );
 }
 
+/// Edge case surfaced in production:
+///
+/// If the first commit in the last segment is corrupt, creating a new segment
+/// would fail because the `tx_range` is the same as the corrupt segment.
+///
+/// We don't automatically recover from that, but test that `open` returns an
+/// error providing some context.
+#[test]
+fn first_commit_in_last_segment_corrupt() {
+    enable_logging();
+
+    let repo = repo::Memory::new();
+    let options = Options {
+        max_segment_size: 512,
+        max_records_in_commit: NonZeroU16::new(1).unwrap(),
+        ..<_>::default()
+    };
+    {
+        let mut log = commitlog::Generic::open(repo.clone(), options).unwrap();
+        fill_log_with(&mut log, iter::once([b'x'; 64]).cycle().take(9));
+    }
+    let segments = repo.existing_offsets().unwrap();
+    assert_eq!(2, segments.len(), "repo should contain 2 segments");
+
+    {
+        let last_segment = repo.open_segment_writer(*segments.last().unwrap()).unwrap();
+        let mut data = last_segment.buf_mut();
+        data[segment::Header::LEN + 1..].fill(0);
+    }
+
+    assert_matches!(
+        commitlog::Generic::<_, [u8; 64]>::open(repo, options),
+        Err(e) if e.kind() == io::ErrorKind::InvalidData,
+    );
+}
+
 fn open_log<T>(repo: ShortMem) -> commitlog::Generic<ShortMem, T> {
     commitlog::Generic::open(
         repo,
@@ -229,6 +267,12 @@ impl ShortMem {
     }
 }
 
+impl fmt::Display for ShortMem {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        fmt::Display::fmt(&self.inner, f)
+    }
+}
+
 impl Repo for ShortMem {
     type SegmentWriter = ShortSegment;
     type SegmentReader = io::BufReader<repo::mem::Segment>;