Retry streaming get requests (#15)#383
Conversation
|
|
||
| retry_error_body: bool, | ||
| /// Combines a [`RetryableRequest`] with a [`RetryContext`] | ||
| pub(crate) struct RetryableRequestBuilder { |
There was a problem hiding this comment.
This largely exists to keep the changeset down
| } | ||
| } | ||
|
|
||
| fn get_range_meta( |
There was a problem hiding this comment.
I opted to split up get_result as it was relatively hard to follow, being a single >100 line function
| let value = val | ||
| .to_str() | ||
| .map_err(|source| GetResultError::InvalidContentRange { source })?; | ||
| fn retry_stream( |
There was a problem hiding this comment.
This is the real meat of the PR - the rest is largely just the plumbing to make this possible
| }, | ||
| // Retry all response body errors | ||
| (Err(e), Some(etag)) if !ctx.retry_ctx.exhausted() => { | ||
| let sleep = ctx.retry_ctx.backoff(); |
There was a problem hiding this comment.
We use the same RetryContext for retrying body errors and response errors, I felt this was the most reasonable approach - if I set RetryConfig to have 10 max retries, I'd be suprised if it could make more than 10 requests if there are a mix of errors.
There was a problem hiding this comment.
This strategy will make many classes of transient error less severe which is great
However, it is still possible to get timeout errors when actively reading a streaming response body as the retry mechanism kicks in after timeout, even though progress was made
At some point, it may make sense to detect when progress was made and adjust the remaining retries / policy accordingly.
| fn is_end_stream(&self) -> bool { | ||
| self.0.is_end_stream() | ||
| } | ||
|
|
||
| fn size_hint(&self) -> SizeHint { | ||
| self.0.size_hint() | ||
| } |
There was a problem hiding this comment.
This is more correct, and without this the HttpServer won't populate the ContentLength header (as it uses the size hint to do this)
| let (parts, retry_body) = request.into_parts(); | ||
| let retry_etag = get_etag(&parts.headers).map_err(Self::err)?; | ||
|
|
||
| if etag != &retry_etag { |
| ) -> BoxStream<'static, Result<Bytes>> { | ||
| futures::stream::try_unfold( | ||
| (self, body, etag, range), | ||
| |(mut ctx, mut body, etag, mut range)| async move { |
There was a problem hiding this comment.
I think that code is rather elegant 👍
|
I tested this out and it works great. Details at #15 (comment) |
alamb
left a comment
There was a problem hiding this comment.
Thank you @tustvold and @crepererum -- this is pretty amazing
| }, | ||
| // Retry all response body errors | ||
| (Err(e), Some(etag)) if !ctx.retry_ctx.exhausted() => { | ||
| let sleep = ctx.retry_ctx.backoff(); |
There was a problem hiding this comment.
This strategy will make many classes of transient error less severe which is great
However, it is still possible to get timeout errors when actively reading a streaming response body as the retry mechanism kicks in after timeout, even though progress was made
At some point, it may make sense to detect when progress was made and adjust the remaining retries / policy accordingly.
|
The CI appears to fail after this PR: |
* Retry streaming get requests (apache#15) * Add tests * Fix size hint * Fix wasm32
Draft as needs testsWhich issue does this PR close?
Closes #15
Rationale for this change
With this PR we retry streaming get requests. Technically it is still possible to get response body errors on the other methods, however, I chose to just do get requests as these are the highest impact as they have the largest payloads, and we can always do the others later.
What changes are included in this PR?
Are there any user-facing changes?