alamb commented on code in PR #849: URL: https://github.com/apache/arrow-rs-object-store/pull/849#discussion_r3968794275
########## src/retry.rs: ########## @@ -0,0 +1,257 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Retry policies for HTTP-backed multipart uploads +//! +//! Each multipart part first uses the backend's existing bounded request retry +//! loop. If those retries are exhausted, a [`RetryPolicy`] can restart the +//! entire part operation. Restarting at this boundary rebuilds the request, +//! including fetching credentials and generating a new signature. +//! +//! Configure a policy with [`PutMultipartOptions::with_retry_policy`]. The +//! policy applies independently to each part and does not change the retry +//! behavior of multipart initiation, completion, or abort requests. + +use crate::client::retry::{RequestError, RetryError}; +use crate::client::{HttpError, HttpErrorKind}; +use crate::{Error, PutMultipartOptions}; +use async_trait::async_trait; +use http::StatusCode; +use std::error::Error as StdError; +use std::fmt::Debug; +use std::sync::Arc; +use std::time::Duration; +#[cfg(not(all(target_arch = "wasm32", target_os = "unknown")))] +use std::time::Instant; +#[cfg(all(target_arch = "wasm32", target_os = "unknown"))] +use web_time::Instant; + +/// A normalized HTTP failure passed to a [`RetryPolicy`] +/// +/// Backends wrap request failures in provider-specific [`Error`] variants. +/// Before invoking a policy, `object_store` reduces a supported failure to an +/// HTTP response status or transport error kind. Failures that cannot be +/// represented by this enum are returned without invoking the policy. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[non_exhaustive] +pub enum RetryFailure { + /// The HTTP status of a response that the backend classified as failed + /// + /// This is usually a non-success status. Some services, including S3, can + /// report an error in the body of a successful response. In that case this + /// contains the actual successful status, such as `200 OK`. + Status(StatusCode), + + /// The request failed before producing a usable HTTP response + /// + /// The kind distinguishes failures such as connection errors, timeouts, + /// interrupted requests, and response decoding errors. + Transport(HttpErrorKind), +} + +/// The failure and retry state for one operation attempt +/// +/// For multipart uploads, an attempt is one complete call to upload a part, +/// including its bounded request retries. Each part has an independent attempt +/// count and elapsed time. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[non_exhaustive] +pub struct RetryContext { + /// The normalized HTTP failure + pub failure: RetryFailure, + + /// The failed operation attempt number, starting at one + pub attempt: usize, + + /// Time since the first operation attempt started + /// + /// This includes time spent in requests and in earlier policy calls. + pub elapsed: Duration, +} + +/// Controls whether and when a failed operation should be attempted again +/// +/// The policy is shared by concurrently uploaded parts. Implementations must +/// manage their own retry limit and backoff. `object_store` does not add a delay +/// or impose an outer retry limit after this method returns `true`. +#[async_trait] +pub trait RetryPolicy: Debug + Send + Sync + 'static { + /// Decide whether to attempt the failed operation again + /// + /// Because this method is asynchronous, implementations can wait using + /// their own backoff strategy, clock, or sleeper before returning `true`. + /// Returning `false` returns the current operation error to the caller. + async fn retry(&self, context: RetryContext) -> bool; +} + +#[derive(Clone)] +struct MultipartRetryPolicy(Arc<dyn RetryPolicy>); + +impl PutMultipartOptions { + /// Retry failed multipart part uploads according to `policy` + /// + /// The policy runs after the existing bounded request retries are + /// exhausted. A retry repeats the same logical part from the provider + /// operation boundary, allowing credentials and request signatures to be + /// refreshed. Initiating, completing, and aborting the multipart upload + /// retain their existing retry behavior. + #[must_use] + pub fn with_retry_policy(mut self, policy: Arc<dyn RetryPolicy>) -> Self { + self.extensions.insert(MultipartRetryPolicy(policy)); + self + } + + pub(crate) fn retry_policy(&self) -> Option<Arc<dyn RetryPolicy>> { Review Comment: Another thing that Claude flagged is that since `MultipartRetryPolicy` extension wrapper is private and `retry_policy()` is pub(crate), so only in-crate backends can ever read the policy; As a follow on it may make sense to make this getter public (and perhaps document via doc examples) so others implementations can use it ########## src/retry.rs: ########## @@ -0,0 +1,257 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Retry policies for HTTP-backed multipart uploads +//! +//! Each multipart part first uses the backend's existing bounded request retry +//! loop. If those retries are exhausted, a [`RetryPolicy`] can restart the +//! entire part operation. Restarting at this boundary rebuilds the request, +//! including fetching credentials and generating a new signature. +//! +//! Configure a policy with [`PutMultipartOptions::with_retry_policy`]. The +//! policy applies independently to each part and does not change the retry +//! behavior of multipart initiation, completion, or abort requests. + +use crate::client::retry::{RequestError, RetryError}; +use crate::client::{HttpError, HttpErrorKind}; +use crate::{Error, PutMultipartOptions}; +use async_trait::async_trait; +use http::StatusCode; +use std::error::Error as StdError; +use std::fmt::Debug; +use std::sync::Arc; +use std::time::Duration; +#[cfg(not(all(target_arch = "wasm32", target_os = "unknown")))] +use std::time::Instant; +#[cfg(all(target_arch = "wasm32", target_os = "unknown"))] +use web_time::Instant; + +/// A normalized HTTP failure passed to a [`RetryPolicy`] +/// +/// Backends wrap request failures in provider-specific [`Error`] variants. +/// Before invoking a policy, `object_store` reduces a supported failure to an +/// HTTP response status or transport error kind. Failures that cannot be +/// represented by this enum are returned without invoking the policy. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] Review Comment: deriving Copy here I think makes it impossible to add the actual error (which may not derive copy). So I am going to remove that derivation to give us more options in the future ########## src/azure/mod.rs: ########## @@ -287,10 +291,23 @@ impl MultipartUpload for AzureMultiPartUpload { let idx = self.part_idx; self.part_idx += 1; let state = Arc::clone(&self.state); + let content_id = AzureClient::new_block_id(); Box::pin(async move { - let part = state.client.put_block(&state.location, idx, data).await?; - state.parts.put(idx, part); - Ok(()) + let mut retry = MultipartRetry::new(state.retry_policy.clone()); Review Comment: these are very elegant loops (in that they express the "retry on failure" very clearly) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
