fix(topology): Fix for issue causing stalling on shutdown for sinks configured w/ disk buffers (#24949)

* fix(buffers): prevent reload stall when disk buffer config changes

- Changing a disk buffer's configuration (e.g. `max_size`) during a
live config reload would stall indefinitely or fail with a `buffer.lock` error.
This was caused by two issues:
    1. The sink's detach trigger was only cancelled for buffer-reuse
cases, so sinks with changed disk buffer configs never had their input
stream terminated — disk buffer readers do not return `None` when the
writer disconnects, causing the old sink task to hang forever.
    2. The source output pump only processed fanout control messages
(Remove/Pause) during active sends, so idle sources would never drop their
`BufferSender` clone, keeping the `Arc<Ledger>` file lock alive even after the
sink task completed.
  - Cancel the detach trigger for changed disk buffer sinks so the old
sink task can complete.
  - Make the source output pump `select!` on both events and fanout
control messages, so Remove/Pause is processed promptly even when the source
is idle.
  - Add a retry loop (30s timeout) when acquiring the disk buffer lock
to handle the small race window between the sink task completing and the
fanout releasing the writer.
  - Add `BufferConfig::has_disk_stage()` helper for identifying disk
buffer configurations.

* Add unit tests for reload when disk buffer config modified

* Add changelog fragment

* Integration test exercising full disk buffer path w/ s3 sink

- This test sets up a pipeline that sends using the s3 sinks with disk
buffers enabled.

- Data is sent through the disk buffer before and after configuration
reload.

* Clean-up changelog fragment

* Fix spelling issues

* Fix merge error and cleanup use of config()

- Make consistent use of the config() method in all tests within this
file

* Simplify set of control_channel_open status

* Remove unnecessary mutable modifiers

* Prefer tempDir to just writing to /tmp

---------

Co-authored-by: Pavlos Rontidis <pavlos.rontidis@gmail.com>

Author: graphcareful

changelog.d/24929_fix_stall_on_disk_shutdown.fix.md

@@ -0,0 +1,6 @@
+Fixed issue during in place reload of a sink with a disk buffer configured, where
+the component would stall for batch.timeout_sec before gracefully reloading.
+This fix also resolves issues Vector had where it would ignore SIGINT during
+cases where the pipeline stall had occurred.
+
+authors: graphcareful

lib/vector-buffers/src/config.rs

@@ -371,6 +371,13 @@ impl Default for BufferConfig {
 }
 
 impl BufferConfig {
+    /// Returns true if any stage in this buffer configuration uses disk-based storage.
+    pub fn has_disk_stage(&self) -> bool {
+        self.stages()
+            .iter()
+            .any(|stage| matches!(stage, BufferType::DiskV2 { .. }))
+    }
+
     /// Gets all of the configured stages for this buffer.
     pub fn stages(&self) -> &[BufferType] {
         match self {

lib/vector-core/src/fanout.rs

@@ -108,6 +108,20 @@ impl Fanout {
         }
     }
 
+    /// Waits for the next control message and applies it.
+    ///
+    /// Returns `true` if a message was processed, `false` if the control
+    /// channel was closed.
+    pub async fn recv_control_message(&mut self) -> bool {
+        match self.control_channel.recv().await {
+            Some(msg) => {
+                self.apply_control_message(msg);
+                true
+            }
+            None => false,
+        }
+    }
+
     /// Apply a control message directly against this instance.
     ///
     /// This method should not be used if there is an active `SendGroup` being processed.