feat(upload): Retry objectstore requests

[jjbayer]

If necessary, retry requests to objectstore.

• For now, a simple loop with one second between attempts. Later on we might implement exponential backoff or similar.
• Only 429, 502, 503, 504 responses, connection errors, and timeout errors are retried.
• For streams, add a new wrapper that makes them retriable until they are polled the first time.

fixes: INGEST-826

[linear-code]

INGEST-826 Implement retries for objectstore requests

[Dav1dde]

I think in an ideal world you only need this:

• A wrapper which keeps track of whether a stream was polled or not. This is just another Stream combinator.
• The wrapper allows recovery if it hasn't been polled yet.
• You pass a &mut wrapper_stream to the client to transfer it.
• The request fails, you attempt to recover from the wrapper via wrapper_stream.reocver() -> Option<S>
• If you get None, the connection failed, if you get Some you can attempt to retry

There is just a small problem where the objectstore client requires BoxStream<'static, ..>, which I assume can't be changed because of reqwest requirements (?).

But you can can keep a similar API and functionality, you now need to employ interior mutability. Which leads to something similar what you have with the API surface. But instead of doing the Drop/oneshot dance I'd look into using a Mutex<Option<T>> (or fancier atomic based variant) and take()'ing the item on poll instead of trying to recover on Drop.

This avoids the oneshot and drop dance, which actually depends on the client dropping the stream where there is no guarantee it actually has to since you're giving full ownership.

[jjbayer]

But instead of doing the Drop/oneshot dance I'd look into using a Mutex<Option<T>> (or fancier atomic based variant) and take()'ing the item on poll instead of trying to recover on Drop.

My first iteration had an Arc<Mutex<Option<T>>>. I thought the channel implementation was more elegant, but you are right that the API is awkward. E.g. I would never wait on the channel Receiver, always call try_recv instead. I'll give this another go and will let you know how it goes.

[Dav1dde]

My first iteration had an Arc<Mutex<Option>>. [...]

It's a bit awkward to deal with directly, but I think with the right amount of wrappers (maybe some kind of TakeOnce(Mutex<Option<T>>)) it could turn out quite nice.

[jjbayer]

This is another case where the function ideally wouldn't even return anything. I didn't want to touch that function in this PR though.

[jjbayer]

This metric is now logged further down the call stack, so it disappears in a few places.

[jjbayer]

The reason why I added a new type, rather than implementing Stream for TakeOnce<S: Stream>, is that there is no point in having a mutex after the first poll -> the internal state of the stream transitions to a simple wrapper.

[Dav1dde]

Shouldn't the signature here not include the mut?:

Suggested change

	pub fn take(&mut self) -> Option<T> {
	pub fn take(&self) -> Option<T> {

[Dav1dde]

NonZero* type in the config and make that state impossible?

[Dav1dde]

We have a RetryBackoff type, maybe worth also using here.

[jjbayer]

👍 , will consider it in case the simple loop doesn't work.

[cursor]

Retries 503 but PR specifies only 429, 502, 504

Low Severity

is_retriable includes StatusCode::SERVICE_UNAVAILABLE (503) in the set of retriable status codes, but the PR description explicitly states "Only 429, 502, 504 responses and connection errors are retried." While 503 is arguably a reasonable retriable status, it wasn't part of the documented intent and could lead to unexpected retry behavior for service-unavailable responses.

 

^{Reviewed by Cursor Bugbot for commit 4317a26. Configure here.}

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

There are 2 total unresolved issues (including 1 from previous review).

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit 5259894. Configure here.}

[sentry]

Bug: In handle_trace_attachment, a session creation failure causes an early return, preventing the AttachmentUpload metric from being emitted and silently dropping the error.
_{Severity: MEDIUM}

Suggested Fix

Modify the error handling in handle_trace_attachment to ensure the AttachmentUpload metric is emitted when session creation fails. This can be done by explicitly handling the session creation error and emitting the metric before returning, similar to how it's handled in handle_envelope and handle_event_attachment.

Prompt for AI Agent

Review the code at the location below. A potential bug has been identified by an AI
agent. Verify if this is a real issue. If it is, propose a fix; if not, explain why it's
not valid.

Location: relay-server/src/services/objectstore.rs#L413-L414

Potential issue: When creating a session for a trace attachment fails within
`do_handle_trace_attachment`, the error is wrapped as `Error::UploadFailed` and the
function returns early. This prevents the `upload()` function from being called, which
is where the `AttachmentUpload` metric is supposed to be emitted. Consequently, session
creation failures for trace attachments are not reported, creating an observability gap.
This behavior is inconsistent with other handlers like `handle_envelope` and
`handle_event_attachment`, which correctly emit metrics in this scenario.

[jjbayer]

I will follow-up.