Skip to content

fix(aws_s3 source): add retry delay in sqs::Ingestor #22999

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

fix(aws_s3 source): add retry delay in sqs::Ingestor #22999

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 23 additions & 11 deletions src/sources/aws_s3/sqs.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
use std::collections::HashMap;
use std::time::Duration;
use std::{future::ready, num::NonZeroUsize, panic, sync::Arc, sync::LazyLock};

use aws_sdk_s3::operation::get_object::GetObjectError;
Expand Down Expand Up @@ -401,27 +402,37 @@ impl IngestorProcess {
async fn run(mut self) {
let shutdown = self.shutdown.clone().fuse();
pin!(shutdown);
let delay = Duration::from_millis(500);
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit: This could be a constant.


loop {
select! {
_ = &mut shutdown => break,
_ = self.run_once() => {},
result = self.run_once() => {
match result {
Ok(()) => {}
Err(_) => {
trace!("run_once failed, will retry after delay");
tokio::time::sleep(delay).await;
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I wonder how this constant was chosen. Would an exponential backoff strategy work better here or it's an overkill?

Copy link
Author

@medzin medzin Jun 13, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I was treating this code as a hotfix rather than a long-term solution, so I chose a simple constant delay to quickly address the issue. The current 0.5-second delay doesn't introduce significant latency but greatly reduces the number of retries triggered by the source, which helps stabilize things short-term. That said, I agree a constant delay may not be ideal across all use cases - different values might work better for different users. An exponential backoff could offer a more flexible and resilient solution without needing to make the delay configurable. Happy to explore that in a follow-up PR.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I can see how this improves things in your situation but we generally don't want to merge hotfixes. If you have urgent need for this commit, you can always maintain a fork with this commit and use a custom Vector build.

Going back to this PR, what do you think about introducing a new opt-in config parameter here? We can probably reuse ExponentialBackoff from src/sinks/util/retries.rs.

Example:

  • 1st retry: 1 second
  • 2nd retry: 2 seconds
  • 3rd retry: 4 seconds
  • 4th retry: 8 seconds
  • 5th retry: 16 seconds
  • 6th retry: 30 seconds (capped)

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sure. I will check it.

}
}
},
}
}
}

async fn run_once(&mut self) {
let messages = self.receive_messages().await;
let messages = messages
.inspect(|messages| {
async fn run_once(&mut self) -> Result<(), ()> {
let messages = match self.receive_messages().await {
Ok(messages) => {
emit!(SqsMessageReceiveSucceeded {
count: messages.len(),
});
})
.inspect_err(|err| {
emit!(SqsMessageReceiveError { error: err });
})
.unwrap_or_default();
messages
}
Err(err) => {
emit!(SqsMessageReceiveError { error: &err });
return Err(());
}
};

let mut delete_entries = Vec::new();
let mut deferred_entries = Vec::new();
Expand Down Expand Up @@ -517,7 +528,7 @@ impl IngestorProcess {
message = "Deferred queue not configured, but received deferred entries.",
internal_log_rate_limit = true
);
return;
return Ok(());
};
let cloned_entries = deferred_entries.clone();
match self
Expand Down Expand Up @@ -572,6 +583,7 @@ impl IngestorProcess {
}
}
}
Ok(())
}

async fn handle_sqs_message(&mut self, message: Message) -> Result<(), ProcessingError> {
Expand Down