fix(disk buffer): respect when_full on capacity errors

changelog.d/disk_v2_runtime_capacity_backpressure.fix.md

@@ -0,0 +1,3 @@
+`disk_v2` buffers now apply their configured `when_full` policy when runtime filesystem space or quota is exhausted. Blocking buffers retry with backpressure, while drop-newest and overflow buffers promptly handle subsequent unwritten events according to policy. Records whose writes have already started remain owned by the disk buffer and complete exactly once after capacity recovers. Startup and non-capacity I/O failures remain fatal.
+
+authors: Jansen-w

lib/vector-buffers/Cargo.toml

@@ -20,6 +20,7 @@ crossbeam-utils.workspace = true
 derivative.workspace = true
 fslock = { version = "0.2.1", default-features = false, features = ["std"] }
 futures.workspace = true
+libc.workspace = true
 memmap2 = { version = "0.9.10", default-features = false }
 metrics.workspace = true
 num-traits = { version = "0.2.19", default-features = false }
@@ -51,9 +52,11 @@ rand.workspace = true
 serde_json.workspace = true
 serde_yaml.workspace = true
 temp-dir = "0.2.0"
+tokio = { workspace = true, features = ["test-util"] }
 tokio-test.workspace = true
 tracing-fluent-assertions = { version = "0.3" }
 tracing-subscriber = { workspace = true, features = ["env-filter", "fmt", "registry", "std", "ansi"] }
+vector-common = { path = "../vector-common", default-features = false, features = ["test"] }
 
 [[bench]]
 name = "sized_records"

lib/vector-buffers/src/internal_events.rs

@@ -1,4 +1,4 @@
-use std::time::Duration;
+use std::{path::Path, time::Duration};
 
 use metrics::Histogram;
 use vector_common::NamedInternalEvent;
@@ -8,6 +8,49 @@ use vector_common::{
     registered_event,
 };
 
+#[derive(Debug, NamedInternalEvent)]
+pub struct DiskBufferBackpressure<'a> {
+    pub operation: &'static str,
+    pub error: &'a std::io::Error,
+    pub retry_delay: Duration,
+    pub buffer_path: &'a Path,
+}
+
+impl InternalEvent for DiskBufferBackpressure<'_> {
+    fn emit(self) {
+        warn!(
+            message = "Disk buffer is waiting for filesystem capacity.",
+            operation = self.operation,
+            error = %self.error,
+            error_kind = ?self.error.kind(),
+            retry_delay_ms = self.retry_delay.as_millis(),
+            buffer_path = %self.buffer_path.display(),
+            internal_log_rate_limit = false,
+        );
+    }
+}
+
+#[derive(Debug, NamedInternalEvent)]
+pub struct DiskBufferBackpressureRecovered<'a> {
+    pub operation: &'static str,
+    pub retries: u64,
+    pub duration: Duration,
+    pub buffer_path: &'a Path,
+}
+
+impl InternalEvent for DiskBufferBackpressureRecovered<'_> {
+    fn emit(self) {
+        info!(
+            message = "Disk buffer filesystem capacity recovered.",
+            operation = self.operation,
+            retries = self.retries,
+            duration_ms = self.duration.as_millis(),
+            buffer_path = %self.buffer_path.display(),
+            internal_log_rate_limit = false,
+        );
+    }
+}
+
 #[derive(NamedInternalEvent)]
 pub struct BufferCreated {
     pub buffer_id: String,

lib/vector-buffers/src/lib.rs

@@ -46,7 +46,10 @@ use vector_common::{
     finalization::{AddBatchNotifier, Finalizable, GroupedFinalizable},
 };
 
-/// Event handling behavior when a buffer is full.
+/// Controls what happens when a buffer reaches its configured size limit.
+///
+/// If the disk fills, Vector follows this setting for new events. An event already being written
+/// finishes when space becomes available. Initialization and other I/O errors remain fatal.
 #[configurable_component]
 #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
 #[serde(rename_all = "snake_case")]