AsyncRead/AsyncWrite Poisoning Behaviour

[tustvold]

Is your feature request related to a problem or challenge? Please describe what you are trying to do.

Currently where ObjectStore exposes APIs in terms of tokio's AsyncWrite and AsyncRead, any error poisons the entire operation. Subsequent attempts to read/write will likely result in a panic. This is not well documented, and may not be ideal.

Describe the solution you'd like

At the very least we should document the current behaviour, but it is unclear, at least to me, what the "correct" behaviour here even is:

AsyncWrite::poll_write returns when the bytes have been "written" to the writer, including potentially to an in-flight buffer, see here. In the case of WriteMultiPart this means AsyncWrite::poll_write returns Ok before any network to actually write the data to object storage.

Any errors will therefore be surfaced in AsyncWrite::poll_flush or AsyncWrite::poll_shutdown, which presents a few problems:

• The PutPart implementation retries intermittent errors based on the RetryConfig, and so we must surface any errors to the user
• It is unclear how the caller can determine from the error what byte range needs to be retried, as part uploads are chunked and parallel
• It is unclear how the caller could retry this byte range even if it could be ascertained

This all makes me think that the current behaviour is probably the best we can do, short of not using the tokio IO traits, but I wonder if others have any thoughts on this

Describe alternatives you've considered

Additional context

[tustvold]

One option might be to return the error, but also re-enqueue the operation to run again. That way if polled again it will effectively just try the operation again 🤔

This would be similar to how std::io::BufWriter handles this particular scenario.

It would then be conceivable for code to retry on write error, although it is rather convoluted:

async fn write_all_with_retry<'a, W: AsyncWrite + Unpin>(
    writer: &'a mut W,
    mut buf: &'a [u8],
) -> impl Stream<Item = std::io::Result<usize>> + 'a {
    futures::stream::poll_fn(move |cx| {
        if buf.is_empty() {
            return Poll::Ready(None);
        }
        return Poll::Ready(Some(
            match futures::ready!(Pin::new(&mut *writer).poll_write(cx, buf)) {
                Ok(x) => {
                    buf.consume(x);
                    Ok(x)
                }
                Err(e) => Err(e),
            },
        ));
    })
}