From 212946fd02606971f16e701877602143c0ab7d70 Mon Sep 17 00:00:00 2001 From: "taylor.chandler.young" Date: Thu, 16 Jul 2026 15:51:27 +0200 Subject: [PATCH] fix(aws_s3 source): bound S3 GetObject with default client timeouts The aws_s3 source created its S3 client with no read or operation timeout (only the SDK-default ~3.1s connect timeout). A half-open or silently dropped connection -- for example one reaped by a NAT gateway idle timeout -- could therefore hang a polling task indefinitely on the GetObject response-body read, stopping SQS polling entirely and stalling S3-based ingestion until the process was restarted. Retries are disabled on this client, so there was no recovery. Apply default S3 client timeouts (connect_timeout_seconds=5, read_timeout_seconds=30) and expose them as source configuration. The read timeout is applied per-read, so a slow but steadily progressing transfer of a large object is not cut off; no operation timeout is set by default to avoid capping legitimately long transfers. This mirrors the existing SQS client timeout support on the same source. Co-Authored-By: Claude Opus 4.8 (1M context) --- changelog.d/aws_s3_source_s3_timeouts.fix.md | 3 ++ src/aws/timeout.rs | 15 ++++++ src/sources/aws_s3/mod.rs | 51 +++++++++++++++++++- 3 files changed, 67 insertions(+), 2 deletions(-) create mode 100644 changelog.d/aws_s3_source_s3_timeouts.fix.md diff --git a/changelog.d/aws_s3_source_s3_timeouts.fix.md b/changelog.d/aws_s3_source_s3_timeouts.fix.md new file mode 100644 index 0000000000000..5b6272833a9a1 --- /dev/null +++ b/changelog.d/aws_s3_source_s3_timeouts.fix.md @@ -0,0 +1,3 @@ +The `aws_s3` source now bounds its S3 `GetObject` requests with default client timeouts (`connect_timeout_seconds = 5`, `read_timeout_seconds = 30`) and exposes them as configuration options. Previously the S3 client was created without a read or operation timeout, so a half-open or silently dropped connection — for example one reaped by a NAT gateway idle timeout — could hang a polling task indefinitely, stalling SQS-based S3 ingestion until the process was restarted. The `read_timeout` applies per-read, so slow but steadily progressing transfers of large objects are not cut off. Any dimension can be overridden (or an operation timeout added) via `connect_timeout_seconds`, `read_timeout_seconds`, and `operation_timeout_seconds` on the source. + +authors: taylorchandleryoung diff --git a/src/aws/timeout.rs b/src/aws/timeout.rs index 93bee13cc8e0e..6c8a239ee0efd 100644 --- a/src/aws/timeout.rs +++ b/src/aws/timeout.rs @@ -47,6 +47,21 @@ pub struct AwsTimeout { } impl AwsTimeout { + /// Creates a new timeout configuration from the given connect, operation, and read + /// timeouts, each expressed in seconds. A `None` leaves that dimension unbounded (subject + /// to the AWS SDK defaults). + pub const fn new( + connect_timeout: Option, + operation_timeout: Option, + read_timeout: Option, + ) -> Self { + Self { + connect_timeout, + operation_timeout, + read_timeout, + } + } + /// returns the connection timeout pub const fn connect_timeout(&self) -> Option { self.connect_timeout diff --git a/src/sources/aws_s3/mod.rs b/src/sources/aws_s3/mod.rs index 5fd849be0d2f7..459569b8841c3 100644 --- a/src/sources/aws_s3/mod.rs +++ b/src/sources/aws_s3/mod.rs @@ -19,7 +19,10 @@ use vrl::value::{Kind, kind::Collection}; use super::util::MultilineConfig; use crate::{ - aws::{RegionOrEndpoint, auth::AwsAuthentication, create_client, create_client_and_region}, + aws::{ + AwsTimeout, RegionOrEndpoint, auth::AwsAuthentication, create_client, + create_client_and_region, + }, codecs::DecodingConfig, common::{s3::S3ClientBuilder, sqs::SqsClientBuilder}, config::{ @@ -117,6 +120,19 @@ pub struct AwsS3Config { #[configurable(derived)] tls_options: Option, + /// Client timeout configuration for S3 operations. + /// + /// These settings bound the S3 `GetObject` request and the read of its response body. Any + /// dimension left unset falls back to a safe default (`connect_timeout_seconds = 5`, + /// `read_timeout_seconds = 30`) so that a half-open or silently dropped connection — for + /// example one reaped by a NAT gateway idle timeout — cannot hang a polling task + /// indefinitely and stall S3 ingestion. `read_timeout_seconds` applies per-read rather than + /// to the whole request, so a slow but steadily progressing transfer of a large object is + /// not cut off. + #[configurable(derived)] + #[serde(default)] + timeout: Option, + /// The namespace to use for logs. This overrides the global setting. #[configurable(metadata(docs::hidden))] #[serde(default)] @@ -151,6 +167,33 @@ const fn default_true() -> bool { true } +/// Default connect timeout for the S3 client, in seconds. +const DEFAULT_S3_CONNECT_TIMEOUT_SECONDS: u64 = 5; + +/// Default read (per-read inactivity) timeout for the S3 client, in seconds. +/// +/// This bounds the response-body read of `GetObject` so a stalled or silently dropped +/// connection cannot hang a polling task indefinitely. It is applied per-read, so a slow but +/// steadily progressing transfer of a large object is not cut off. No operation timeout is set +/// by default, to avoid capping legitimately long large-object transfers. +const DEFAULT_S3_READ_TIMEOUT_SECONDS: u64 = 30; + +/// Resolves the effective S3 client timeout, applying default bounds to any dimension the +/// configuration leaves unset. Connect and read timeouts default to bounded values; the +/// operation timeout is left unset so large-object transfers are not capped. +fn resolve_s3_timeout(configured: Option) -> AwsTimeout { + let configured = configured.unwrap_or_default(); + AwsTimeout::new( + configured + .connect_timeout() + .or(Some(DEFAULT_S3_CONNECT_TIMEOUT_SECONDS)), + configured.operation_timeout(), + configured + .read_timeout() + .or(Some(DEFAULT_S3_READ_TIMEOUT_SECONDS)), + ) +} + impl_generate_config_from_default!(AwsS3Config); #[async_trait::async_trait] @@ -243,6 +286,10 @@ impl AwsS3Config { let region = self.region.region(); let endpoint = self.region.endpoint(); + // Bound the S3 client's requests. Historically this was `None`, which left the + // response-body read of `GetObject` unbounded, so a half-open or silently dropped + // connection could hang a polling task indefinitely and stall SQS-based ingestion. + let s3_timeout = resolve_s3_timeout(self.timeout); let s3_client = create_client::( &S3ClientBuilder { force_path_style: Some(self.force_path_style), @@ -252,7 +299,7 @@ impl AwsS3Config { endpoint.clone(), proxy, self.tls_options.as_ref(), - None, + Some(&s3_timeout), ) .await?;