vector/config/
sink_validated.rs

1//! Validated sink infrastructure for the sinks-only lifecycle architecture.
2//!
3//! This module implements a split between the serde/typetag-facing `SinkConfig` role
4//! and the topology-facing runtime role. It introduces:
5//!
6//! - `ValidatedSink`: a generic, implementor-facing trait. A migrated sink implements
7//!   this and works entirely with its own concrete validated type `T` — it returns `T`
8//!   from `validate` and receives `&T` back in `build`. No `Any` in sight.
9//! - `DynValidatedSink`: the object-safe boundary the framework actually crosses.
10//!   A blanket impl derives it from any `ValidatedSink`, performing the `Box<dyn Any>`
11//!   erasure and downcast automatically.
12//!
13//! The erased validated state is stored on `SinkOuter` (see `SinkOuter::validated`) and
14//! consumed at build time by `SinkOuter::build`, which dispatches through
15//! `DynValidatedSink::build_dyn`.
16//!
17//! `ValidatedSink::validate` is the same phase as the RFC's
18//! `SinkConfig::validate_structure` (pure structural validation owned by config
19//! compilation), but it retains the validated state so `build` does not
20//! redo it.
21//!
22//! Design goals:
23//! - Validation returns retained values passed to build
24//! - Validation is pure: no filesystem/network/credentials/spawn/await
25//! - Build may do environment-dependent construction but must not redo pure validation
26//! - Preserve raw config serialization, reload diffing, and metadata access
27//! - Sink implementors never touch `Any`; the framework owns the erasure
28
29use std::any::Any;
30
31use async_trait::async_trait;
32use vector_lib::sink::VectorSink;
33
34use super::SinkContext;
35use crate::sinks::Healthcheck;
36
37/// Generic validated-sink trait, implemented by migrated sinks.
38///
39/// The implementor works entirely with their concrete validated type `Self::Validated`:
40///
41/// ```ignore
42/// #[async_trait]
43/// impl ValidatedSink for MySinkConfig {
44///     type Validated = ValidatedMySink;
45///
46///     fn validate(&self) -> crate::Result<Self::Validated> {
47///         ValidatedMySink::from_config(self)
48///     }
49///
50///     async fn build(
51///         &self,
52///         validated: &Self::Validated,
53///         cx: SinkContext,
54///     ) -> crate::Result<(VectorSink, Healthcheck)> {
55///         validated.build(cx).await
56///     }
57/// }
58/// ```
59///
60/// The framework automatically erases `Self::Validated` to `Box<dyn Any>` and restores
61/// it at build time via `DynValidatedSink`, so no `Any` appears in implementor code.
62#[async_trait]
63pub trait ValidatedSink {
64    /// The concrete validated state produced by `validate` and consumed by `build`.
65    type Validated: Send + Sync + 'static;
66
67    /// Performs pure structural validation, returning the validated state.
68    ///
69    /// This is the same phase as the RFC's `SinkConfig::validate_structure`, but it
70    /// retains the validated state so `build` does not redo it.
71    ///
72    /// # Purity Guarantees
73    ///
74    /// This method must be pure: no filesystem access, no network operations, no
75    /// credential resolution, no spawning, and no async/await. All such environment-
76    /// dependent operations belong in `build`.
77    fn validate(&self) -> crate::Result<Self::Validated>;
78
79    /// Builds the sink from the validated state, without redoing pure validation.
80    ///
81    /// May perform environment-dependent construction (HTTP clients, schema fetching, etc.).
82    async fn build(
83        &self,
84        validated: &Self::Validated,
85        cx: SinkContext,
86    ) -> crate::Result<(VectorSink, Healthcheck)>;
87}
88
89/// Object-safe `dyn` boundary used by the framework.
90///
91/// This is what actually crosses the `Box<dyn SinkConfig>` boundary. It is derived
92/// automatically from any `ValidatedSink` by the blanket impl below, which owns the
93/// `Box<dyn Any>` erasure and the downcast back to the concrete validated type.
94#[async_trait]
95pub trait DynValidatedSink {
96    /// Erases the validated state into a `Box<dyn Any>`.
97    fn validate_dyn(&self) -> crate::Result<Box<dyn Any + Send + Sync>>;
98
99    /// Restores the validated state from `&dyn Any` and builds the sink.
100    async fn build_dyn(
101        &self,
102        validated: &(dyn Any + Send + Sync),
103        cx: SinkContext,
104    ) -> crate::Result<(VectorSink, Healthcheck)>;
105}
106
107#[async_trait]
108impl<T> DynValidatedSink for T
109where
110    T: ValidatedSink + Send + Sync + 'static,
111{
112    fn validate_dyn(&self) -> crate::Result<Box<dyn Any + Send + Sync>> {
113        Ok(Box::new(self.validate()?))
114    }
115
116    async fn build_dyn(
117        &self,
118        validated: &(dyn Any + Send + Sync),
119        cx: SinkContext,
120    ) -> crate::Result<(VectorSink, Healthcheck)> {
121        let validated = validated
122            .downcast_ref::<T::Validated>()
123            .expect("validated state type mismatch");
124        self.build(validated, cx).await
125    }
126}
vector/config/sink_validated.rs

vector/config/
sink_validated.rs
