fix(sources): cap compressed and decompressed payload size to prevent OOM

Cargo.toml

@@ -494,7 +494,7 @@ vrl.workspace = true
 rustls = "0.23"
 tokio-rustls = "0.26"
 wiremock = "0.6.2"
-zstd = { version = "0.13.0", default-features = false }
+zstd.workspace = true
 
 [patch.crates-io]
 mlua = { git = "https://github.com/janmejay-s1/mlua.git", branch = "mlua_0.10.2_list_iter_safety" }
@@ -656,10 +656,10 @@ sources-metrics = [
 sources-amqp = ["lapin"]
 sources-apache_metrics = ["sources-utils-http-client"]
 sources-aws_ecs_metrics = ["sources-utils-http-client"]
-sources-aws_kinesis_firehose = ["dep:base64"]
+sources-aws_kinesis_firehose = ["dep:base64", "sources-utils-http-encoding"]
 sources-aws_s3 = ["aws-core", "dep:aws-sdk-sqs", "dep:aws-sdk-s3", "dep:semver", "dep:async-compression", "sources-aws_sqs", "tokio-util/io"]
 sources-aws_sqs = ["aws-core", "dep:aws-sdk-sqs"]
-sources-datadog_agent = ["sources-utils-http-error", "protobuf-build", "dep:prost"]
+sources-datadog_agent = ["sources-utils-http-encoding", "sources-utils-http-error", "protobuf-build", "dep:prost"]
 sources-demo_logs = ["dep:fakedata"]
 sources-dnstap = ["sources-utils-net-tcp", "dep:base64", "dep:hickory-proto", "dep:dnsmsg-parser", "dep:dnstap-parser", "protobuf-build", "dep:prost"]
 sources-docker_logs = ["docker"]
@@ -695,7 +695,7 @@ sources-prometheus-pushgateway = ["sinks-prometheus", "sources-utils-http", "vec
 sources-pulsar = ["dep:apache-avro", "dep:pulsar"]
 sources-redis = ["dep:redis"]
 sources-socket = ["sources-utils-net", "tokio-util/net"]
-sources-splunk_hec = ["dep:roaring"]
+sources-splunk_hec = ["dep:roaring", "sources-utils-http-encoding"]
 sources-statsd = ["sources-utils-net", "tokio-util/net"]
 sources-stdin = ["tokio-util/io"]
 sources-syslog = ["codecs-syslog", "sources-utils-net", "tokio-util/net"]
@@ -891,6 +891,11 @@ sinks-webhdfs = ["dep:opendal"]
 # Identifies that the build is a nightly build
 nightly = []
 
+# Timing-sensitive tests. Kept out of the default suite so CI scheduling noise cannot fail a run;
+# enable explicitly (dataplane-build does) to exercise them, and eventually in a dedicated fork of
+# the test workflow.
+performance-tests = []
+
 # Integration testing-related features
 all-integration-tests = [
   "amqp-integration-tests",

clippy.toml

@@ -4,11 +4,23 @@ cognitive-complexity-threshold = 75
 # https://rust-lang.github.io/rust-clippy/master/index.html#disallowed_method
 disallowed-methods = [
   { path = "std::io::Write::write", reason = "This doesn't handle short writes, use `write_all` instead." },
+  { path = "zstd::stream::copy_decode", reason = "Use `vector_common::decompression::CappedDecoder::zstd` (or `CappedDecoder::zstd_http` in HTTP contexts) to enforce the decompression size cap." },
+  { path = "zstd::stream::decode_all", reason = "Use `vector_common::decompression::CappedDecoder::zstd` (or `CappedDecoder::zstd_http` in HTTP contexts) to enforce the decompression size cap." },
+  { path = "zstd::bulk::decompress", reason = "Use `vector_common::decompression::CappedDecoder::zstd` (or `CappedDecoder::zstd_http` in HTTP contexts) to enforce the decompression size cap." },
+  { path = "warp::body::bytes", reason = "Reads the whole request body into memory unbounded. Use `crate::sources::util::http::capped_body()` to cap the compressed body size at the global decompressed-size limit." },
 ]
 
+# Decompression limits are per-component configuration, not process state: take them from the
+# component's context (`SourceContext` / `SinkContext` / `TransformContext` -> `GlobalOptions`)
+# and pass them into the capped wrapper, rather than reading a global.
 disallowed-types = [
   { path = "once_cell::sync::OnceCell", reason = "Use `std::sync::OnceLock` instead." },
   { path = "once_cell::unsync::OnceCell", reason = "Use `std::cell::OnceCell` instead." },
   { path = "once_cell::sync::Lazy", reason = "Use `std::sync::LazyLock` instead." },
   { path = "once_cell::unsync::Lazy", reason = "Use `std::sync::LazyCell` instead." },
+  { path = "flate2::read::GzDecoder", reason = "Use `vector_common::decompression::CappedDecoder::gzip` to enforce the decompression size cap." },
+  { path = "flate2::read::MultiGzDecoder", reason = "Use `vector_common::decompression::CappedDecoder::gzip` to enforce the decompression size cap." },
+  { path = "flate2::read::ZlibDecoder", reason = "Use `vector_common::decompression::CappedDecoder::zlib` to enforce the decompression size cap." },
+  { path = "flate2::read::DeflateDecoder", reason = "Use `vector_common::decompression::CappedDecoder` for decompression with the size cap." },
+  { path = "zstd::stream::read::Decoder", reason = "Use `vector_common::decompression::CappedDecoder::zstd` (or `CappedDecoder::zstd_http` in HTTP contexts) to enforce the decompression size cap." },
 ]

lib/codecs/src/actions/decoding/config.rs

@@ -1,5 +1,6 @@
 use crate::decoding::{DeserializerConfig, FramingConfig};
 use serde::{Deserialize, Serialize};
+use vector_common::decompression::CompressionLimits;
 use vector_core::config::LogNamespace;
 
 use super::Decoder;
@@ -13,6 +14,12 @@ pub struct DecodingConfig {
     decoding: DeserializerConfig,
     /// The namespace used when decoding.
     log_namespace: LogNamespace,
+    /// Limits applied by framers that decompress.
+    ///
+    /// Defaults to the documented cap; a component with access to its context should override this
+    /// with `GlobalOptions`' value via [`Self::with_compression_limits`].
+    #[serde(default, skip)]
+    compression_limits: CompressionLimits,
 }
 
 impl DecodingConfig {
@@ -27,9 +34,22 @@ impl DecodingConfig {
             framing,
             decoding,
             log_namespace,
+            compression_limits: CompressionLimits::with_max_decompressed_size_bytes(
+                vector_common::decompression::DEFAULT_MAX_DECOMPRESSED_SIZE_BYTES,
+            ),
         }
     }
 
+    /// Sets the compression limits framers should decompress under.
+    ///
+    /// Take these from the component's context (`cx.globals.limits.compression`) so the deployment
+    /// controls the cap rather than a process-wide default.
+    #[must_use]
+    pub const fn with_compression_limits(mut self, limits: CompressionLimits) -> Self {
+        self.compression_limits = limits;
+        self
+    }
+
     /// Get the decoding configuration.
     pub const fn config(&self) -> &DeserializerConfig {
         &self.decoding
@@ -43,7 +63,7 @@ impl DecodingConfig {
     /// Builds a `Decoder` from the provided configuration.
     pub fn build(&self) -> vector_common::Result<Decoder> {
         // Build the framer.
-        let framer = self.framing.build();
+        let framer = self.framing.build(self.compression_limits);
 
         // Build the deserializer.
         let deserializer = self.decoding.build()?;