fix(buffers): recover disk_v2 from durable checkpoints

changelog.d/disk_v2_buffer_size_recompute_on_restart.fix.md

@@ -0,0 +1,3 @@
+Fixed a `disk_v2` buffer accounting bug where the unread-bytes counter (`total_buffer_size`) could underflow on restart. Previously the buffer seeded the counter from the sum of on-disk data file sizes and then had the reader decrement it record-by-record while seeking to its persisted read position. Those two inputs were captured at different moments — the data files on disk versus the read position in the ledger, which are flushed on independent schedules — so a crash between their flushes could make the decrements exceed the seeded total. Because the counter is unsigned, the subtraction wrapped to a near-maximum value, making the buffer appear permanently full and wedging the writer (no further writes accepted, and recovery could stall). The reader now recomputes the unread total authoritatively at the end of its startup seek, from a single consistent snapshot — the total size of the data files still on disk minus the bytes it consumed reaching the resume position — so the counter can no longer underflow across a restart.
+
+authors: graphcareful

lib/vector-buffers/src/variants/disk_v2/common.rs

@@ -30,6 +30,8 @@ pub const MINIMUM_MAX_RECORD_SIZE: usize = align16(RECORD_HEADER_LEN + 1);
 // have it configured.
 pub const DEFAULT_FLUSH_INTERVAL: Duration = Duration::from_millis(500);
 
+pub const DEFAULT_DATA_FILE_CLEANUP_INTERVAL: Duration = Duration::from_secs(1);
+
 // Using 256KB as it aligns nicely with the I/O size exposed by major cloud providers.  This may not
 // be the underlying block size used by the OS, but it still aligns well with what will happen on
 // the "backend" for cloud providers, which is simply a useful default for when we want to look at
@@ -44,6 +46,29 @@ pub const MAX_FILE_ID: u16 = u16::MAX;
 #[cfg(test)]
 pub const MAX_FILE_ID: u16 = 6;
 
+pub(crate) fn data_file_id_in_range(file_id: u16, start: u16, end: u16) -> bool {
+    if start <= end {
+        (start..=end).contains(&file_id)
+    } else {
+        file_id >= start || file_id <= end
+    }
+}
+
+pub(crate) fn data_file_name(file_id: u16) -> String {
+    format!("buffer-data-{file_id}.dat")
+}
+
+pub(crate) fn parse_data_file_id(path: &Path) -> Option<u16> {
+    let file_name = path.file_name()?.to_str()?;
+    let id = file_name
+        .strip_prefix("buffer-data-")?
+        .strip_suffix(".dat")?
+        .parse()
+        .ok()?;
+
+    (id < MAX_FILE_ID).then_some(id)
+}
+
 // The alignment used by the record serializer.
 const SERIALIZER_ALIGNMENT: usize = 16;
 const MAX_ALIGNABLE_AMOUNT: usize = usize::MAX - SERIALIZER_ALIGNMENT;

lib/vector-buffers/src/variants/disk_v2/io.rs

@@ -1,4 +1,8 @@
-use std::{io, path::Path};
+use std::{
+    future::Future,
+    io,
+    path::{Path, PathBuf},
+};
 
 use tokio::{
     fs::OpenOptions,
@@ -84,7 +88,26 @@ pub trait Filesystem: Send + Sync {
     ///
     /// If an I/O error occurred when attempting to delete the file, an error variant will be
     /// returned describing the underlying error.
-    async fn delete_file(&self, path: &Path) -> io::Result<()>;
+    fn delete_file<'a>(
+        &'a self,
+        path: &'a Path,
+    ) -> impl Future<Output = io::Result<()>> + Send + 'a;
+
+    /// Lists files in a directory.
+    ///
+    /// # Errors
+    ///
+    /// If an I/O error occurred when attempting to list the directory, an error variant will be
+    /// returned describing the underlying error.
+    fn list_files<'a>(
+        &'a self,
+        path: &'a Path,
+    ) -> impl Future<Output = io::Result<Vec<PathBuf>>> + Send + 'a;
+
+    /// Returns whether the buffer should spawn its periodic stale data file cleanup task.
+    fn supports_background_cleanup(&self) -> bool {
+        true
+    }
 }
 
 pub trait AsyncFile: AsyncRead + AsyncWrite + Send + Sync {
@@ -96,6 +119,13 @@ pub trait AsyncFile: AsyncRead + AsyncWrite + Send + Sync {
     /// will be returned describing the underlying error.
     async fn metadata(&self) -> io::Result<Metadata>;
 
+    /// Truncates the underlying file to the specified size.
+    ///
+    /// # Errors
+    /// If `size` is greater than the current file size, or an I/O error occurred when attempting to
+    /// truncate the file, an error variant will be returned describing the underlying error.
+    async fn truncate(&self, size: u64) -> io::Result<()>;
+
     /// Attempts to synchronize all OS-internal data, and metadata, to disk.
     ///
     /// This function will attempt to ensure that all in-memory data reaches the filesystem before returning.
@@ -161,8 +191,30 @@ impl Filesystem for ProductionFilesystem {
         unsafe { memmap2::MmapMut::map_mut(&std_file) }
     }
 
-    async fn delete_file(&self, path: &Path) -> io::Result<()> {
-        tokio::fs::remove_file(path).await
+    fn delete_file<'a>(
+        &'a self,
+        path: &'a Path,
+    ) -> impl Future<Output = io::Result<()>> + Send + 'a {
+        tokio::fs::remove_file(path)
+    }
+
+    #[allow(clippy::manual_async_fn)]
+    fn list_files<'a>(
+        &'a self,
+        path: &'a Path,
+    ) -> impl Future<Output = io::Result<Vec<PathBuf>>> + Send + 'a {
+        async move {
+            let mut entries = tokio::fs::read_dir(path).await?;
+            let mut files = Vec::new();
+
+            while let Some(entry) = entries.next_entry().await? {
+                if entry.file_type().await?.is_file() {
+                    files.push(entry.path());
+                }
+            }
+
+            Ok(files)
+        }
     }
 }
 
@@ -221,6 +273,18 @@ impl AsyncFile for tokio::fs::File {
         })
     }
 
+    async fn truncate(&self, size: u64) -> io::Result<()> {
+        let current_size = self.metadata().await?.len();
+        if size > current_size {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidInput,
+                "cannot extend a file through the truncation API",
+            ));
+        }
+
+        self.set_len(size).await
+    }
+
     async fn sync_all(&self) -> io::Result<()> {
         self.sync_all().await
     }