fix(sources): map capped-body rejections to client errors in firehose and splunk_hec sources (#25826)

Author: thomasqueirozb

src/sources/aws_kinesis_firehose/filters.rs

@@ -18,6 +18,7 @@ use super::{
 };
 use crate::{
     SourceSender, codecs,
+    common::http::ErrorMessage,
     internal_events::{AwsKinesisFirehoseRequestError, AwsKinesisFirehoseRequestReceived},
     sources::http_server::HttpConfigParamKind,
     sources::util::{decompression::CappedDecoder, http::capped_body},
@@ -208,6 +209,10 @@ async fn handle_firehose_rejection(err: warp::Rejection) -> Result<impl warp::Re
         message = e.to_string();
         code = e.status();
         request_id = e.request_id();
+    } else if let Some(e) = err.find::<ErrorMessage>() {
+        code = e.status_code();
+        message = e.message().to_string();
+        request_id = None;
     } else if let Some(e) = err.find::<warp::reject::MissingHeader>() {
         code = StatusCode::BAD_REQUEST;
         message = format!("Required header missing: {}", e.name());
@@ -282,4 +287,23 @@ mod tests {
 
         assert_eq!(parsed.access_key, Some("secret123".to_owned()));
     }
+
+    #[tokio::test]
+    async fn capped_body_rejection_maps_to_client_error() {
+        use warp::Reply;
+
+        // `capped_body()` rejects oversized payloads with an `ErrorMessage` (e.g. 413). The
+        // recovery must surface that status instead of falling through to a 500.
+        let rejection = warp::reject::custom(ErrorMessage::new(
+            StatusCode::PAYLOAD_TOO_LARGE,
+            "Request body exceeds limit of 1024 bytes.".to_owned(),
+        ));
+
+        let response = handle_firehose_rejection(rejection)
+            .await
+            .unwrap()
+            .into_response();
+
+        assert_eq!(response.status(), StatusCode::PAYLOAD_TOO_LARGE);
+    }
 }

src/sources/splunk_hec/mod.rs

@@ -65,6 +65,7 @@ use self::{
 use crate::{
     SourceSender,
     codecs::DecodingConfig,
+    common::http::ErrorMessage,
     config::{DataType, Resource, SourceConfig, SourceContext, SourceOutput, log_schema},
     event::{Event, LogEvent, Value},
     http::{KeepaliveConfig, MaxConnectionAgeLayer, build_http_trace_layer},
@@ -2014,6 +2015,8 @@ async fn finish_err(rejection: Rejection) -> Result<(Response,), Rejection> {
                 response_json(StatusCode::BAD_REQUEST, splunk_response::ACK_IS_DISABLED)
             }
         },))
+    } else if let Some(error) = rejection.find::<ErrorMessage>() {
+        Ok((response_json(error.status_code(), error),))
     } else {
         Err(rejection)
     }
@@ -2084,6 +2087,22 @@ mod tests {
         crate::test_util::test_generate_config::<SplunkConfig>();
     }
 
+    #[tokio::test]
+    async fn finish_err_maps_capped_body_to_client_error() {
+        // `capped_body()` rejects oversized payloads with an `ErrorMessage` (e.g. 413). The
+        // recovery must surface that status instead of letting it fall through to a 500.
+        let rejection = warp::reject::custom(ErrorMessage::new(
+            StatusCode::PAYLOAD_TOO_LARGE,
+            "Request body exceeds limit of 1024 bytes.".to_owned(),
+        ));
+
+        let (response,) = finish_err(rejection)
+            .await
+            .expect("capped-body rejection should be recovered");
+
+        assert_eq!(response.status(), StatusCode::PAYLOAD_TOO_LARGE);
+    }
+
     /// Splunk token
     const TOKEN: &str = "token";
     const VALID_TOKENS: &[&str; 2] = &[TOKEN, "secondary-token"];