commitlog: Improve error context (#3506)

The commitlog creates new segments atomically, returning EEXIST if the
segment already exists. This is to break a retry loop in case the
filesystem becomes unwritable.

This error did not contain any context about what does not exist, so
this patch adds some.

Also, an unhandled edge case has been discovered:

When opening an existing log, the commitlog will try to resume the last
segment for writing. If it finds a corrupt commit in that segment, it
won't resume, but instead create a new segment at the corrupt commit's
offset + 1.

However, if the first commit in the last segment is corrupted, the
offset will be that of the last segment -- trying to start a new segment
will thus fail with EEXIST.

Without additional recovery mechanisms, it is not obvious what to do in
this case: the segment could contain valid data after the initial
commit, so we certainly don't want to throw it away.

Instead, we now detect this case and return `InvalidData` with some
context.

# Expected complexity level and risk

1

# Testing

- [ ] A (regression) test is included

Author: kim

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