vector_buffers/

lib.rs

//! The Vector Core buffer
//!
//! This library implements a channel like functionality, one variant which is
//! solely in-memory and the other that is on-disk. Both variants are bounded.

#![deny(warnings)]
#![deny(clippy::all)]
#![deny(clippy::pedantic)]
#![allow(clippy::module_name_repetitions)]
#![allow(clippy::type_complexity)] // long-types happen, especially in async code
#![allow(clippy::must_use_candidate)]
#![allow(async_fn_in_trait)]

#[macro_use]
extern crate tracing;

mod buffer_usage_data;

pub mod config;
pub use config::{BufferConfig, BufferType, MemoryBufferSize};
use encoding::Encodable;
pub(crate) use vector_common::Result;
use vector_config::configurable_component;

pub mod encoding;

mod internal_events;

#[cfg(test)]
pub mod test;
pub mod topology;

pub(crate) mod variants;

/// `disk_v2`'s write-buffer size, re-exported under `test` so the harness can
/// size payloads against the real value instead of hardcoding it.
#[cfg(feature = "test")]
pub use variants::disk_v2::common::DEFAULT_WRITE_BUFFER_SIZE as WRITE_BUFFER_SIZE_V2;

use std::fmt::Debug;

#[cfg(test)]
use quickcheck::{Arbitrary, Gen};
use vector_common::{
    byte_size_of::ByteSizeOf,
    finalization::{AddBatchNotifier, Finalizable, GroupedFinalizable},
};

/// Event handling behavior when a buffer is full.
#[configurable_component]
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum WhenFull {
    /// Wait for free space in the buffer.
    ///
    /// This applies backpressure up the topology, signalling that sources should slow down
    /// the acceptance/consumption of events. This means that while no data is lost, data will pile
    /// up at the edge.
    #[default]
    Block,

    /// Drops the event instead of waiting for free space in buffer.
    ///
    /// The event will be intentionally dropped. This mode is typically used when performance is the
    /// highest priority, and it is preferable to temporarily lose events rather than cause a
    /// slowdown in the acceptance/consumption of events.
    DropNewest,

    /// Overflows to the next stage in the buffer topology.
    ///
    /// If the current buffer stage is full, attempt to send this event to the next buffer stage.
    /// That stage may also be configured overflow, and so on, but ultimately the last stage in a
    /// buffer topology must use one of the other handling behaviors. This means that next stage may
    /// potentially be able to buffer the event, but it may also block or drop the event.
    ///
    /// This mode can only be used when two or more buffer stages are configured.
    #[configurable(metadata(docs::hidden))]
    Overflow,
}

#[cfg(test)]
impl Arbitrary for WhenFull {
    fn arbitrary(g: &mut Gen) -> Self {
        // TODO: We explicitly avoid generating "overflow" as a possible value because nothing yet
        // supports handling it, and will be defaulted to using "block" if they encounter
        // "overflow".  Thus, there's no reason to emit it here... yet.
        if bool::arbitrary(g) {
            WhenFull::Block
        } else {
            WhenFull::DropNewest
        }
    }
}

/// An item that can be buffered in memory.
///
/// This supertrait serves as the base trait for any item that can be pushed into a memory buffer.
/// It is a relaxed version of `Bufferable` that allows for items that are not `Encodable` (e.g., `Instant`),
/// which is an unnecessary constraint for memory buffers.
pub trait InMemoryBufferable:
    AddBatchNotifier
    + Finalizable
    + ByteSizeOf
    + EventCount
    + Debug
    + Send
    + Sync
    + Unpin
    + Sized
    + 'static
{
}

// Blanket implementation for anything that is already in-memory bufferable.
impl<T> InMemoryBufferable for T where
    T: AddBatchNotifier
        + Finalizable
        + ByteSizeOf
        + EventCount
        + Debug
        + Send
        + Sync
        + Unpin
        + Sized
        + 'static
{
}

/// An item that can be buffered.
///
/// This supertrait serves as the base trait for any item that can be pushed into a buffer.
pub trait Bufferable: InMemoryBufferable + Encodable + GroupedFinalizable {
    /// Drops any sub-items that cannot be persisted by the calling backend (e.g. due to
    /// format-imposed nesting depth limits), reporting them as dropped via the appropriate
    /// telemetry. Returns `None` if nothing remains worth writing.
    ///
    /// # Who calls this
    ///
    /// Only persistent backends with wire-format constraints invoke this — today that's
    /// the disk-v2 sender (`SenderAdapter::send`/`try_send`). In-memory channels skip it
    /// entirely because they hold the in-memory representation and have no nesting-limit
    /// risk. A new backend with similar constraints should call this in the same place
    /// and surface the resulting `FilterDrops` to `BufferSender` so that buffer-usage
    /// instrumentation stays consistent with what actually lands in the buffer.
    ///
    /// # Default behaviour
    ///
    /// The default returns `Some(self)` if the item carries any events, and `None` if
    /// it is already empty. This means an item that arrives empty (`event_count() == 0`)
    /// is silently *not* persisted — preserving the pre-existing
    /// "don't write empty records to disk" behaviour the call site used to enforce.
    /// Types whose owners want empty items to be persisted must override this.
    ///
    /// # Skipping this call
    ///
    /// If a persistent backend writes an item without first calling `filter_unencodable`,
    /// any sub-item that exceeds the format's limits will surface as a hard
    /// [`Encodable::encode`] error and the *entire* item is rejected — including any
    /// sibling sub-items that would otherwise have encoded fine. The filter is the only
    /// path that produces graceful per-item drop with telemetry and a `Rejected` event
    /// status; the encode-level check exists purely as defense-in-depth to ensure a
    /// corrupt record cannot reach disk if a future caller forgets to filter.
    fn filter_unencodable(self) -> Option<Self> {
        if self.event_count() > 0 {
            Some(self)
        } else {
            None
        }
    }

    /// Returns whether every sub-item can be persisted by a backend with wire-format
    /// constraints, without consuming or modifying the item.
    ///
    /// This is the non-destructive counterpart to [`Bufferable::filter_unencodable`], and
    /// exists so routing policy can be decided *before* any filtering happens. In
    /// particular `WhenFull::Overflow` needs to know that an item can never reach disk, so
    /// it can hand the item to the overflow stage intact rather than pruning sub-items for
    /// a write that would not have succeeded at any buffer occupancy.
    ///
    /// The default returns `true`, which is correct for any type without format limits.
    /// Implementors overriding [`Bufferable::filter_unencodable`] must override this too,
    /// and the two must agree: this returns `false` exactly when `filter_unencodable` would
    /// drop at least one sub-item.
    fn is_fully_encodable(&self) -> bool {
        true
    }
}

/// Hook for observing items as they are sent into a `BufferSender`.
pub trait BufferInstrumentation<T: Bufferable>: Send + Sync + 'static {
    /// Called immediately before the item is emitted to the underlying buffer.
    /// The underlying type is stored in an `Arc`, so we cannot have `&mut self`.
    fn on_send(&self, item: &mut T);
}

pub trait EventCount {
    fn event_count(&self) -> usize;
}

impl<T> EventCount for Vec<T>
where
    T: EventCount,
{
    fn event_count(&self) -> usize {
        self.iter().map(EventCount::event_count).sum()
    }
}

impl<T> EventCount for &T
where
    T: EventCount,
{
    fn event_count(&self) -> usize {
        (*self).event_count()
    }
}

#[track_caller]
pub(crate) fn spawn_named<T>(
    task: impl std::future::Future<Output = T> + Send + 'static,
    _name: &str,
) -> tokio::task::JoinHandle<T>
where
    T: Send + 'static,
{
    #[cfg(tokio_unstable)]
    return tokio::task::Builder::new()
        .name(_name)
        .spawn(task)
        .expect("tokio task should spawn");

    #[cfg(not(tokio_unstable))]
    tokio::spawn(task)
}