fix(buffers): fsync the disk_v2 data directory on file creation and its newly-created ancestors

changelog.d/disk_v2_data_dir_ancestor_fsync.fix.md

@@ -0,0 +1,3 @@
+Fixed a durability gap in `disk_v2` buffer creation. On first startup the buffer creates its data directory chain (`<base>/buffer/v2/<id>`) with `create_dir_all`, but that does not make the new directory entries durable in their parents (fsyncing a directory persists its contents, not its own entry). A power loss shortly after startup could therefore drop the freshly created buffer directories -- and an already-acknowledged data file inside them -- even though each data file's own directory is fsynced when the file is created. The buffer now fsyncs every directory it creates, from the buffer directory's parent up to the first pre-existing ancestor, before any write can be acknowledged.
+
+authors: graphcareful

changelog.d/disk_v2_data_file_directory_fsync.fix.md

@@ -0,0 +1,3 @@
+Fixed a durability gap in the `disk_v2` buffer that could permanently lose end-to-end acknowledged records across a crash. When the buffer created a new data file it synced the file's contents to disk, but never synced the parent data directory, so the file's directory entry was not made durable. After a crash (for example a `SIGKILL` or power loss) the newly created file could disappear entirely on reopen even though its contents had been fsynced — taking with it records the buffer had already acknowledged to its upstream source, which the source therefore would not retransmit. The buffer now fsyncs the data directory immediately after creating a new data file, so the file is durably present before any records written into it can be acknowledged.
+
+authors: graphcareful

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

@@ -85,6 +85,19 @@ 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<()>;
+
+    /// Durably persists the directory entries (file creations and deletions) of `path`.
+    ///
+    /// Synchronizing a file's contents with [`AsyncFile::sync_all`] does not make that file's
+    /// *directory entry* durable; the parent directory must itself be synchronized. Without this,
+    /// a newly created data file can disappear after a crash even though its contents were synced,
+    /// losing records the buffer may have already acknowledged to its upstream source.
+    ///
+    /// # Errors
+    ///
+    /// If an I/O error occurred while synchronizing the directory, an error variant will be
+    /// returned describing the underlying error.
+    async fn sync_directory(&self, path: &Path) -> io::Result<()>;
 }
 
 pub trait AsyncFile: AsyncRead + AsyncWrite + Send + Sync {
@@ -164,6 +177,66 @@ impl Filesystem for ProductionFilesystem {
     async fn delete_file(&self, path: &Path) -> io::Result<()> {
         tokio::fs::remove_file(path).await
     }
+
+    async fn sync_directory(&self, path: &Path) -> io::Result<()> {
+        // On Unix, a directory's entries are made durable by opening the directory and issuing an
+        // `fsync` against its descriptor.
+        #[cfg(unix)]
+        {
+            let directory = tokio::fs::File::open(path).await?;
+            match directory.sync_all().await {
+                Ok(()) => Ok(()),
+                // Some filesystems (certain FUSE, overlay, or network mounts) do not support fsync
+                // against a directory descriptor and return `Unsupported`/`InvalidInput`. That
+                // leaves a newly created data file's directory entry non-durable across a crash,
+                // but it is not a reason to fail an otherwise-working buffer, so warn once and
+                // continue rather than propagating the error to the write/init path.
+                Err(e)
+                    if matches!(
+                        e.kind(),
+                        io::ErrorKind::Unsupported | io::ErrorKind::InvalidInput
+                    ) =>
+                {
+                    use std::sync::Once;
+
+                    static WARN_ONCE: Once = Once::new();
+                    WARN_ONCE.call_once(|| {
+                        tracing::warn!(
+                            error = %e,
+                            "This filesystem does not support synchronizing a directory; the disk \
+                             buffer cannot make new data-file directory entries durable, so a crash \
+                             may lose a newly created data file and any records already acknowledged \
+                             from it."
+                        );
+                    });
+                    Ok(())
+                }
+                Err(e) => Err(e),
+            }
+        }
+        // On non-Unix platforms there is no portable way to durably persist a directory's entries
+        // (e.g. on Windows a directory cannot simply be opened and flushed like a regular file), so
+        // we cannot make a newly created data file's directory entry durable here. The buffer
+        // therefore remains exposed to losing a freshly created data file -- and the records it has
+        // acknowledged from it -- across a crash on these platforms. Warn once so the gap is
+        // observable rather than silent.
+        #[cfg(not(unix))]
+        {
+            use std::sync::Once;
+
+            static WARN_ONCE: Once = Once::new();
+            WARN_ONCE.call_once(|| {
+                tracing::warn!(
+                    "The disk buffer cannot durably synchronize its data directory on this \
+                     platform; a crash may lose a newly created data file and any records already \
+                     acknowledged from it."
+                );
+            });
+
+            let _ = path;
+            Ok(())
+        }
+    }
 }
 
 /// Builds a set of `OpenOptions` for opening a file as readable/writable.

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

@@ -363,6 +363,11 @@ where
         self.get_data_file_path(self.get_current_reader_file_id())
     }
 
+    /// Gets the buffer's data directory.
+    pub(super) fn data_dir(&self) -> &std::path::Path {
+        &self.config.data_dir
+    }
+
     /// Gets the current writer data file path.
     pub fn get_current_writer_data_file_path(&self) -> PathBuf {
         self.get_data_file_path(self.get_current_writer_file_id())
@@ -584,11 +589,57 @@ where
         config: DiskBufferConfig<FS>,
         usage_handle: BufferUsageHandle,
     ) -> Result<Ledger<FS>, LedgerLoadCreateError> {
-        // Create our containing directory if it doesn't already exist.
+        // Create our containing directory (and any missing ancestors) if it doesn't already exist.
+        //
+        // `create_dir_all` creates the directories but does not make their *entries* durable in
+        // their parents: fsyncing a directory persists the files and subdirectories it contains,
+        // not the directory's own entry in its parent. The buffer's data directory is a freshly
+        // created chain (`<base>/buffer/v2/<id>`), so without fsyncing those ancestors a power loss
+        // after first startup could drop the `<id>`, `v2`, or `buffer` directory -- taking an
+        // already-acknowledged data file inside it down with it -- even though each data file's own
+        // directory is fsynced when the file is created. To close that gap, fsync every directory we
+        // create, from the buffer directory's parent up to the first pre-existing ancestor, before
+        // any write into the buffer can be acknowledged.
+
+        // Record the deepest ancestor that already exists, so we only fsync the chain we create.
+        // Walk up only on `NotFound`; any other stat error (permissions, transient I/O) is treated
+        // as "exists" so we don't misclassify an existing buffer as freshly created and re-fsync
+        // its pre-existing ancestors.
+        let mut deepest_existing = config.data_dir.as_path();
+        while matches!(
+            fs::metadata(deepest_existing).await,
+            Err(ref e) if e.kind() == io::ErrorKind::NotFound
+        ) {
+            match deepest_existing.parent() {
+                Some(parent) => deepest_existing = parent,
+                None => break,
+            }
+        }
+        let buffer_dir_already_existed = deepest_existing == config.data_dir.as_path();
+
         fs::create_dir_all(&config.data_dir)
             .await
             .context(IoSnafu)?;
 
+        if !buffer_dir_already_existed {
+            let mut ancestor = config.data_dir.parent();
+            while let Some(dir) = ancestor {
+                // `parent()` yields an empty path for the final component of a relative `data_dir`;
+                // there is no directory to open and sync there, so skip it.
+                if !dir.as_os_str().is_empty() {
+                    config
+                        .filesystem
+                        .sync_directory(dir)
+                        .await
+                        .context(IoSnafu)?;
+                }
+                if dir == deepest_existing {
+                    break;
+                }
+                ancestor = dir.parent();
+            }
+        }
+
         // Acquire an exclusive lock on our lock file, which prevents another Vector process from
         // loading this buffer and clashing with us.  Specifically, though: this does _not_ prevent
         // another process from messing with our ledger files, or any of the data files, etc.

lib/vector-buffers/src/variants/disk_v2/tests/model/filesystem.rs

@@ -352,4 +352,10 @@ impl Filesystem for TestFilesystem {
             Err(io_err_not_found())
         }
     }
+
+    async fn sync_directory(&self, _path: &Path) -> io::Result<()> {
+        // The in-memory test filesystem does not model directory-entry durability separately, so
+        // there is nothing to synchronize here.
+        Ok(())
+    }
 }

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

@@ -1107,9 +1107,12 @@ where
                 .filesystem()
                 .open_file_writable_atomic(&data_file_path)
                 .await;
+            // The third tuple element records whether we just atomically created this file (and
+            // therefore a new directory entry), so we only fsync the directory when there is a new
+            // entry to persist -- not when we reopen an existing (possibly empty) file.
             let file = match maybe_data_file {
                 // We were able to create the file, so we're good to proceed.
-                Ok(data_file) => Some((data_file, 0)),
+                Ok(data_file) => Some((data_file, 0, true)),
                 // We got back an error trying to open the file: might be that it already exists,
                 // might be something else.
                 Err(e) => match e.kind() {
@@ -1135,7 +1138,7 @@ where
                             // or it's not empty but we're not skipping to the next file, which can
                             // only mean that we're still initializing, and so this would be the
                             // data file we left off writing to.
-                            Some((data_file, file_len))
+                            Some((data_file, file_len, false))
                         } else {
                             // The file isn't empty, and we're not in initialization anymore, which
                             // means this data file is one that the reader still hasn't finished
@@ -1149,7 +1152,7 @@ where
                 },
             };
 
-            if let Some((data_file, data_file_size)) = file {
+            if let Some((data_file, data_file_size, just_created)) = file {
                 // We successfully opened the file and it can be written to.
                 debug!(
                     data_file_path = data_file_path.to_string_lossy().as_ref(),
@@ -1160,6 +1163,17 @@ where
                 // Make sure the file is flushed to disk, especially if we just created it.
                 data_file.sync_all().await?;
 
+                // Syncing the file's contents above does not make its directory entry durable.
+                // Sync the data directory so a newly created file is present on disk before
+                // acknowledging anything written into it. Only needed when we actually created a
+                // new entry -- reopening an existing (even empty) file needs no directory sync.
+                if just_created {
+                    self.ledger
+                        .filesystem()
+                        .sync_directory(self.ledger.data_dir())
+                        .await?;
+                }
+
                 self.writer = Some(RecordWriter::new(
                     data_file,
                     data_file_size,