commitlog: Truncate segment before resuming (#5116)

`read_exact` doesn't distinguish between EOF and not enough bytes to
fill the given buffer. We may thus consider a segment valid to be
resumed, but leave in trailing bytes (less than the commit header
length). That can cause silent data loss, because appending more data
will render the segment corrupt: a restart will then start a new
segment, leaving anything after the trailing bytes unreachable.

To fix this, truncate the segment to the size determined by
`Metadata::extract` before resuming writes.

# Expected complexity level and risk

2

# Testing

Added a test

Author: kim

crates/commitlog/src/repo/fs.rs

@@ -1,6 +1,6 @@
 use std::fmt;
 use std::fs::{self, File};
-use std::io;
+use std::io::{self, Seek};
 use std::sync::Arc;
 
 use log::{debug, warn};
@@ -277,7 +277,7 @@ impl Repo for Fs {
         // The segment file either does not exist, or is of length zero.
         // Write the header to a temporary file and atomically move it into place.
         let mut tmp = tempfile::Builder::new().make_in(&self.root.0, |tmp_path| {
-            File::options().read(true).append(true).create_new(true).open(tmp_path)
+            File::options().read(true).write(true).create_new(true).open(tmp_path)
         })?;
         header.write(&mut tmp)?;
         tmp.as_file_mut().sync_all()?;
@@ -292,7 +292,11 @@ impl Repo for Fs {
     }
 
     fn open_segment_writer(&self, offset: u64) -> io::Result<Self::SegmentWriter> {
-        File::options().read(true).append(true).open(self.segment_path(offset))
+        // NOTE: We previously used `O_APPEND`, but Windows demands `write(true)`
+        // so that we can truncate trailing garbage in [super::resume_segment_writer].
+        let mut file = File::options().read(true).write(true).open(self.segment_path(offset))?;
+        file.seek(io::SeekFrom::End(0))?;
+        Ok(file)
     }
 
     fn segment_file_path(&self, offset: u64) -> Option<String> {

crates/commitlog/src/repo/mod.rs

@@ -8,7 +8,7 @@ use spacetimedb_fs_utils::compression::Zstd;
 pub use spacetimedb_fs_utils::compression::{CompressOnce, CompressionStats};
 
 use crate::{
-    commit::Commit,
+    commit::{self, Commit},
     error,
     index::{IndexFile, IndexFileMut},
     segment::{self, FileLike, Header, Metadata, OffsetIndexWriter, Reader, Writer},
@@ -375,16 +375,24 @@ pub fn resume_segment_writer<R: Repo>(
     if reader.sealed() {
         Ok(ResumedSegment::Sealed(meta))
     } else {
-        let Metadata {
-            header: _,
-            tx_range,
-            size_in_bytes,
-            max_epoch,
-            max_commit_offset: _,
-            max_commit: _,
-        } = meta;
         let mut writer = repo.open_segment_writer(offset)?;
+        // Ensure that the segment's size is exactly what we determined.
+        //
+        // When `Metadata` encounters EOF, it could be that there actually are
+        // trailing bytes in the segment, but less than the commit header
+        // length. This is difficult to detect due to the use of `read_exact`,
+        // so ensure we remove any trailing bytes.
+        //
+        // To be extra cautious, check that no more than the header length
+        // bytes are left over before truncating the segment. This is an assert
+        // because it would be a bug in `Metadata::extract`.
+        assert!(
+            writer.segment_len()? < meta.size_in_bytes + commit::Header::LEN as u64,
+            "{repo}: trailing bytes exceed commit header length in segment {offset}"
+        );
+        writer.ftruncate(meta.tx_range.end, meta.size_in_bytes)?;
         // Ensure we have enough space for this segment.
+        //
         // The segment could have been created without the `fallocate` feature
         // enabled, so we call this here again to ensure writes can't fail due
         // to ENOSPC.
@@ -394,15 +402,15 @@ pub fn resume_segment_writer<R: Repo>(
 
         Ok(ResumedSegment::Resumed(Writer {
             commit: Commit {
-                min_tx_offset: tx_range.end,
+                min_tx_offset: meta.tx_range.end,
                 n: 0,
                 records: Vec::new(),
-                epoch: max_epoch,
+                epoch: meta.max_epoch,
             },
             inner: io::BufWriter::new(writer),
 
-            min_tx_offset: tx_range.start,
-            bytes_written: size_in_bytes,
+            min_tx_offset: meta.tx_range.start,
+            bytes_written: meta.size_in_bytes,
 
             offset_index_head: create_offset_index_writer(repo, offset, opts),
         }))

crates/commitlog/src/segment.rs

@@ -46,7 +46,12 @@ impl Header {
 
     pub fn decode<R: io::Read>(mut read: R) -> io::Result<Self> {
         let mut buf = [0; Self::LEN];
-        read.read_exact(&mut buf)?;
+        read.read_exact(&mut buf).map_err(|e| {
+            io::Error::new(
+                e.kind(),
+                format!("failed to read segment header ({} bytes): {}", Self::LEN, e),
+            )
+        })?;
 
         if !buf.starts_with(&MAGIC) {
             return Err(io::Error::new(
@@ -612,6 +617,7 @@ impl Metadata {
         mut reader: R,
         offset_index: Option<&TxOffsetIndex>,
     ) -> Result<Self, error::SegmentMetadata> {
+        reader.seek(io::SeekFrom::Start(0))?;
         let header = Header::decode(&mut reader)?;
         Self::with_header(min_tx_offset, header, reader, offset_index)
     }

crates/commitlog/tests/random_payload/mod.rs

@@ -1,7 +1,9 @@
+use std::io::Write;
+
 use log::info;
-use spacetimedb_commitlog::repo::Repo;
+use spacetimedb_commitlog::repo::{Repo, SegmentLen};
 use spacetimedb_commitlog::tests::helpers::enable_logging;
-use spacetimedb_commitlog::{commitlog, payload, repo, Commitlog, Options};
+use spacetimedb_commitlog::{commit, commitlog, payload, repo, Commitlog, Options};
 use spacetimedb_paths::server::CommitLogDir;
 use spacetimedb_paths::FromPathUnchecked;
 use tempfile::tempdir;
@@ -196,3 +198,60 @@ fn resume_empty_segment() {
         }
     }
 }
+
+/// Tests that resuming a segment that has trailing bytes smaller than a
+/// commitlog header causes those trailing bytes to be removed.
+///
+/// Regression test for https://github.com/clockworklabs/SpacetimeDB/pull/5116
+#[test]
+fn resume_small_trailing_garbage() {
+    enable_logging();
+
+    let root = tempdir().unwrap();
+    let path = CommitLogDir::from_path_unchecked(root.path());
+
+    let repo = repo::Fs::new(path, None).unwrap();
+    // Write some data.
+    {
+        let mut clog = commitlog::Generic::open(&repo, <_>::default()).unwrap();
+        for (i, payload) in compressible_payloads().take(1024).enumerate() {
+            clog.commit([(i as u64, payload)]).unwrap();
+            clog.flush().unwrap();
+            clog.sync();
+        }
+    }
+
+    // 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();
+        last_segment.write_all(&[67u8; commit::Header::LEN - 1]).unwrap();
+        last_segment.flush().unwrap();
+        last_segment.sync_all().unwrap();
+        last_segment.segment_len().unwrap()
+    };
+    {
+        let mut clog = commitlog::Generic::open(&repo, <_>::default()).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();
+        assert_eq!(
+            last_segment.segment_len().unwrap(),
+            last_segment_size - (commit::Header::LEN - 1) as u64
+        );
+
+        // Add some more data.
+        for (i, payload) in compressible_payloads()
+            .take(1024)
+            .enumerate()
+            .map(|(offset, payload)| (offset + 1024, payload))
+        {
+            clog.commit([(i as u64, payload)]).unwrap();
+            clog.flush().unwrap();
+            clog.sync();
+        }
+
+        assert_eq!(2048, clog.commits_from(0).map(Result::unwrap).count());
+    }
+}

crates/snapshot/src/lib.rs

@@ -1492,14 +1492,39 @@ impl FileOrDirPath<'_> {
     /// On Windows, only the file needs to be synced, and it's even an error to
     /// sync a directory. Passing in [Self::Dir] is thus a no-op on Windows.
     fn sync_all(&self) -> io::Result<()> {
-        #[cfg(target_os = "windows")]
-        if let Self::Dir(_) = self {
-            return Ok(());
+        match self {
+            #[cfg(target_os = "windows")]
+            Self::Dir(path) => Ok(()),
+            #[cfg(not(target_os = "windows"))]
+            Self::Dir(path) => File::open(path)
+                .map_err(|e| {
+                    io::Error::new(
+                        e.kind(),
+                        format!("failed to open directory {} for fsync: {}", path.display(), e),
+                    )
+                })?
+                .sync_all()
+                .map_err(|e| io::Error::new(e.kind(), format!("failed to fsync directory {}: {}", path.display(), e))),
+            Self::File(path) => {
+                File::options()
+                    .read(true)
+                    // Windows needs the file to be writable for `sync_all` to work.
+                    // Set all the open options explicitly, just for visibility.
+                    .write(true)
+                    .truncate(false)
+                    .create(false)
+                    .append(false)
+                    .open(path)
+                    .map_err(|e| {
+                        io::Error::new(
+                            e.kind(),
+                            format!("failed to open file {} for fsync: {}", path.display(), e),
+                        )
+                    })?
+                    .sync_all()
+                    .map_err(|e| io::Error::new(e.kind(), format!("failed to fsync file {}: {}", path.display(), e)))
+            }
         }
-        let (Self::File(path) | Self::Dir(path)) = self;
-        File::open(path)
-            .and_then(|fd| fd.sync_all())
-            .map_err(|e| io::Error::new(e.kind(), format!("failed to fsync {}: {}", path.display(), e)))
     }
 }