From 2a106726c089865a5892df75d36d34c13ef2d01f Mon Sep 17 00:00:00 2001
From: sundaresanr <sundar@exaforce.com>
Date: Sat, 19 Sep 2026 13:13:40 -0700
Subject: [PATCH] fix(aws_s3 source): delete SQS messages that cannot be parsed

A message that fails to parse as an S3 notification fails identically on
every redelivery, but was never deleted: the parse error returned before
the delete, so SQS kept making the message visible until the retention
period expired, growing the queue and competing with valid messages
sharing it.

Queue these messages for deletion after emitting the same parse error as
before, so an unrecognized message shape does not silently disappear.
---
 src/sources/aws_s3/sqs.rs | 35 +++++++++++++++++++++++++++++++++++
 1 file changed, 35 insertions(+)

diff --git a/src/sources/aws_s3/sqs.rs b/src/sources/aws_s3/sqs.rs
index c8a922f7b69e0..cf05b13e3309b 100644
--- a/src/sources/aws_s3/sqs.rs
+++ b/src/sources/aws_s3/sqs.rs
@@ -530,6 +530,30 @@ impl IngestorProcess {
                                 );
                             }
                         }
+                        // A body that fails to parse fails identically on every
+                        // redelivery, so leaving it queued grows the backlog until
+                        // retention expires and starves the messages sharing the queue.
+                        ProcessingError::InvalidSqsMessage { .. }
+                            if self.state.delete_message
+                                && self.state.delete_failed_message =>
+                        {
+                            emit!(SqsMessageProcessingError {
+                                message_id: &message_id,
+                                error: &err,
+                            });
+                            trace!(
+                                message = "Queued unparseable SQS message for deletion.",
+                                id = message_id,
+                                receipt_handle = receipt_handle,
+                            );
+                            delete_entries.push(
+                                DeleteMessageBatchRequestEntry::builder()
+                                    .id(message_id)
+                                    .receipt_handle(receipt_handle)
+                                    .build()
+                                    .expect("all required builder params specified"),
+                            );
+                        }
                         _ => {
                             emit!(SqsMessageProcessingError {
                                 message_id: &message_id,
@@ -1311,3 +1335,14 @@ fn parse_sqs_config() {
     );
     assert!(test.is_err());
 }
+
+// A real body that no variant matches, kept as a regression case for the parse
+// failure the delete gate acts on. This one is CloudTrail's own log-file-delivery
+// notification, which a trail publishes when its SNS topic is shared with the
+// bucket's S3 event topic.
+#[test]
+fn test_unrecognized_notification_fails_to_parse() {
+    let body = r#"{"s3Bucket":"eu-prod-cloudtraillogs","s3ObjectKey":["cloudtrail/AWSLogs/665168952601/CloudTrail/us-east-1/2026/09/19/665168952601_CloudTrail_us-east-1_20260919T1305Z_g01urzqsneIKLYfW.json.gz"]}"#;
+
+    assert!(serde_json::from_str::<SqsEvent>(body).is_err());
+}