fix(odm): declare source retry policy and time out a stalled inline read (#7111)

* fix(odm): declare the remote client retry policy per consumer

The SDK retry policy was an inherited default: one logical call could cost
three wire requests, so the migration breaker counted logical calls on top
of a threefold amplification against a source that was already failing.

Make it an explicit RemoteS3EndpointSpec field. Replication targets declare
today's standard three attempts and keep their behaviour; the on-demand
migration source and its admin probe declare a disabled policy, so one
counted failure is exactly one source request and pull.rs owns the only
retry budget.

* fix(odm): count a stalled inline source as a source timeout

The inline tee wraps its source body in the idle guard, but the tee turns a
stalled source into an ordinary body read error, so the write-back reported
it as a local write failure. Hand commit_inline the guard so the pull is
counted under source_timeout instead.

The background pump now enforces the idle budget through the same guard
rather than a second copy of the timeout loop.

* test(odm): cover a stalled source body end to end

The fake target can now deliver a GetObject body in slices with a pause
between them, so the inline abort can be driven by a stalled source instead
of a truncated one. Two fault cases drop the workarounds they carried for
the SDK's retries: the scripted fault count and the observed source request
count now have to agree.

The operations guide records the retry and idle-timeout guarantees.

crates/ecstore/src/api/mod.rs

@@ -162,8 +162,9 @@ pub mod bucket {
         };
         pub use crate::bucket::on_demand_migration::{
             EnqueueOutcome, LocalObject, MAX_MULTIPART_PARTS, OdmWriteBack, PULL_MAX_RETRIES, PULL_RETRY_BASE_DELAYS,
-            PullCompletion, PullQueue, PullReason, PullSource, QueuedPullOutcome, SourceBody, WriteBackBody, WriteBackError,
-            WriteBackOutcome, WriteBackPart, WriteBackRequest, commit_inline, commit_inline_with, idle_guarded_body,
+            PullCompletion, PullQueue, PullReason, PullSource, QueuedPullOutcome, SourceBody, SourceIdleGuard, WriteBackBody,
+            WriteBackError, WriteBackOutcome, WriteBackPart, WriteBackRequest, commit_inline, commit_inline_with,
+            idle_guarded_body,
         };
         pub mod backfill {
             pub use crate::bucket::on_demand_migration::backfill::{
@@ -241,7 +242,7 @@ pub mod bucket {
 
     pub mod remote_s3_client {
         pub use crate::bucket::remote_s3_client::{
-            PathStyle, RemoteCredentials, RemoteS3ClientError, RemoteS3EndpointSpec, build_remote_s3_client,
+            PathStyle, RemoteCredentials, RemoteS3ClientError, RemoteS3EndpointSpec, RemoteS3RetryPolicy, build_remote_s3_client,
             validate_remote_endpoint,
         };
     }

crates/ecstore/src/bucket/bucket_target_sys.rs

@@ -15,7 +15,9 @@
 use crate::bucket::metadata::BucketMetadata;
 use crate::bucket::metadata_sys::get_bucket_targets_config;
 use crate::bucket::metadata_sys::get_replication_config;
-use crate::bucket::remote_s3_client::{PathStyle, RemoteCredentials, RemoteS3EndpointSpec, build_remote_s3_client};
+use crate::bucket::remote_s3_client::{
+    PathStyle, REPLICATION_TARGET_RETRY_POLICY, RemoteCredentials, RemoteS3EndpointSpec, build_remote_s3_client,
+};
 use crate::bucket::replication::{ObjectLockIntegrity, object_lock_put_integrity};
 use crate::bucket::replication::{ReplicationStatusType, ReplicationTargetConfigBridge};
 use crate::bucket::target::ARN;
@@ -114,6 +116,10 @@ impl From<&BucketTarget> for RemoteS3EndpointSpec {
             ca_cert_pem: (!target.ca_cert_pem.trim().is_empty()).then(|| target.ca_cert_pem.clone()),
             connect_timeout: None,
             read_timeout: None,
+            // Replication has no retry budget of its own on the request path,
+            // so it keeps the SDK's standard three attempts; stating it here
+            // pins the behaviour to this line instead of an SDK default.
+            retry: REPLICATION_TARGET_RETRY_POLICY,
             user_agent_suffix: "",
         }
     }
@@ -2348,11 +2354,11 @@ impl Error for BucketTargetError {}
 mod tests {
     use super::*;
     use crate::bucket::remote_s3_client::{
-        EXPIRED_REMOTE_TARGET_CREDENTIALS, RemoteTargetCredentialsProvider, build_aws_s3_http_client_for_spec,
-        build_aws_s3_http_client_from_target_ca_pem, build_aws_s3_http_client_with_trust_store,
-        build_insecure_aws_s3_http_client, compose_replication_trust_store, ensure_rustls_crypto_provider,
-        load_tls_path_ca_bundles, remote_sdk_credentials, replication_request_checksum_calculation,
-        validate_remote_endpoint_inner, validate_target_ca_pem,
+        EXPIRED_REMOTE_TARGET_CREDENTIALS, RemoteS3RetryPolicy, RemoteTargetCredentialsProvider,
+        build_aws_s3_http_client_for_spec, build_aws_s3_http_client_from_target_ca_pem,
+        build_aws_s3_http_client_with_trust_store, build_insecure_aws_s3_http_client, compose_replication_trust_store,
+        ensure_rustls_crypto_provider, load_tls_path_ca_bundles, remote_sdk_credentials,
+        replication_request_checksum_calculation, validate_remote_endpoint_inner, validate_target_ca_pem,
     };
     use aws_credential_types::Credentials as SdkCredentials;
     use aws_sdk_s3::Config as S3Config;
@@ -3050,6 +3056,11 @@ mod tests {
         assert!(spec.ca_cert_pem.is_none(), "whitespace-only CA PEM means unset");
         assert!(spec.connect_timeout.is_none() && spec.read_timeout.is_none());
         assert_eq!(spec.user_agent_suffix, "");
+        assert_eq!(
+            spec.retry,
+            RemoteS3RetryPolicy::Standard { max_attempts: 3 },
+            "replication targets keep the SDK's historical three attempts"
+        );
         let credentials = spec.credentials.expect("credentials carry over");
         assert_eq!(credentials.account_id, "reset-1");
         assert!(credentials.session_token.is_none(), "blank session token is absent");

crates/ecstore/src/bucket/on_demand_migration/mod.rs

@@ -41,8 +41,8 @@ pub use config::{
 pub use negative_cache::{NEGATIVE_CACHE_MAX_ENTRIES, NegativeCache};
 pub use pull::{
     EnqueueOutcome, LocalObject, MAX_MULTIPART_PARTS, OdmWriteBack, PULL_MAX_RETRIES, PULL_RETRY_BASE_DELAYS, PullCompletion,
-    PullQueue, PullReason, PullSource, QueuedPullOutcome, SourceBody, WriteBackBody, WriteBackError, WriteBackOutcome,
-    WriteBackPart, WriteBackRequest, commit_inline, commit_inline_with, idle_guarded_body,
+    PullQueue, PullReason, PullSource, QueuedPullOutcome, SourceBody, SourceIdleGuard, WriteBackBody, WriteBackError,
+    WriteBackOutcome, WriteBackPart, WriteBackRequest, commit_inline, commit_inline_with, idle_guarded_body,
 };
 pub use stats::{
     GaugeGuard, LastSourceError, LatencyBucketSnapshot, OdmOp, OdmOutcome, OdmStats, OdmStatsSnapshot, PullFailureReason,

crates/ecstore/src/bucket/on_demand_migration/pull.rs

@@ -54,6 +54,7 @@ use std::fmt;
 use std::io;
 use std::pin::Pin;
 use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::task::{Context, Poll};
 use std::time::{Duration, Instant};
 use time::OffsetDateTime;
@@ -149,28 +150,58 @@ pub enum EnqueueOutcome {
 /// Body a source read produces; consumed inside the pump task only.
 pub type SourceBody = Pin<Box<dyn Stream<Item = io::Result<Bytes>> + Send + 'static>>;
 
+/// Says whether an [`idle_guarded_body`] stream ended because the source
+/// stalled. The tee flattens a stalled source into an ordinary body read
+/// error, which the write-back would otherwise report as a local write
+/// failure; the inline path consults this to count the pull under
+/// [`PullFailureReason::SourceTimeout`] instead.
+#[derive(Clone, Debug, Default)]
+pub struct SourceIdleGuard {
+    timed_out: Arc<AtomicBool>,
+}
+
+impl SourceIdleGuard {
+    /// True once the guarded stream ended on the idle budget.
+    pub fn timed_out(&self) -> bool {
+        self.timed_out.load(Ordering::Acquire)
+    }
+}
+
 /// Applies `source_timeout.idle_ms` to a source body chunk by chunk: a chunk
 /// that does not arrive within `idle_timeout` ends the stream with a timeout
-/// error. The background pump enforces the same budget itself; the inline tee
-/// path in the app layer has no pump and wraps its body with this instead, so
-/// a stalled source cannot hold an inline pull open past the configured
-/// budget.
-pub fn idle_guarded_body(body: SourceBody, idle_timeout: Duration) -> SourceBody {
-    Box::pin(futures::stream::unfold(Some(body), move |state| async move {
-        let mut body = state?;
-        match tokio::time::timeout(idle_timeout, body.next()).await {
-            Err(_elapsed) => Some((
-                Err(io::Error::new(
-                    io::ErrorKind::TimedOut,
-                    format!("source body stalled for more than {}ms", idle_timeout.as_millis()),
-                )),
-                None,
-            )),
-            Ok(None) => None,
-            Ok(Some(Err(err))) => Some((Err(err), None)),
-            Ok(Some(Ok(chunk))) => Some((Ok(chunk), Some(body))),
+/// error. Both pull paths share this one implementation — the background pump
+/// wraps the body it copies into the write-back channel, and the app layer
+/// wraps the body it tees between the client and [`commit_inline`].
+///
+/// The budget measures the source, not the consumer: the timeout is armed
+/// inside the stream's own poll, so a consumer that stops asking for bytes (a
+/// slow client, or a tee whose queue is full) leaves no timer running and is
+/// never mistaken for an idle source.
+pub fn idle_guarded_body(body: SourceBody, idle_timeout: Duration) -> (SourceBody, SourceIdleGuard) {
+    let guard = SourceIdleGuard::default();
+    let timed_out = Arc::clone(&guard.timed_out);
+    let stream = futures::stream::unfold(Some(body), move |state| {
+        let timed_out = Arc::clone(&timed_out);
+        async move {
+            let mut body = state?;
+            match tokio::time::timeout(idle_timeout, body.next()).await {
+                Err(_elapsed) => {
+                    timed_out.store(true, Ordering::Release);
+                    Some((
+                        Err(io::Error::new(
+                            io::ErrorKind::TimedOut,
+                            format!("source body stalled for more than {}ms", idle_timeout.as_millis()),
+                        )),
+                        None,
+                    ))
+                }
+                Ok(None) => None,
+                Ok(Some(Err(err))) => Some((Err(err), None)),
+                Ok(Some(Ok(chunk))) => Some((Ok(chunk), Some(body))),
+            }
         }
-    }))
+    });
+    (Box::pin(stream), guard)
 }
 
 /// Body handed to the write-back; `Sync` because the app-layer put path
@@ -362,13 +393,15 @@ impl PumpState {
 }
 
 /// Copies the source body into a bounded channel, enforcing `idle_timeout`
-/// per chunk, `cancel`, and the advertised `expected_size`.
+/// per chunk (through [`idle_guarded_body`]), `cancel`, and the advertised
+/// `expected_size`.
 fn spawn_pump(
-    mut body: SourceBody,
+    body: SourceBody,
     expected_size: u64,
     idle_timeout: Duration,
     cancel: CancellationToken,
 ) -> (mpsc::Receiver<io::Result<Bytes>>, Arc<PumpState>) {
+    let (mut body, _idle) = idle_guarded_body(body, idle_timeout);
     let (tx, rx) = mpsc::channel(PUMP_CHANNEL_CHUNKS);
     let state = Arc::new(PumpState::default());
     let pump_state = Arc::clone(&state);
@@ -377,13 +410,12 @@ fn spawn_pump(
         loop {
             let next = tokio::select! {
                 _ = cancel.cancelled() => Err(pump_state.fail(PumpFailure::Canceled)),
-                next = tokio::time::timeout(idle_timeout, body.next()) => match next {
-                    Err(_elapsed) => Err(pump_state.fail(PumpFailure::Source(SourceError::Timeout))),
-                    Ok(None) if delivered < expected_size => Err(pump_state.fail(PumpFailure::Source(SourceError::Connect(
+                next = body.next() => match next {
+                    None if delivered < expected_size => Err(pump_state.fail(PumpFailure::Source(SourceError::Connect(
                         format!("source body ended after {delivered} of {expected_size} bytes"),
                     )))),
-                    Ok(None) => return,
-                    Ok(Some(Err(err))) => {
+                    None => return,
+                    Some(Err(err)) => {
                         let failure = if err.kind() == io::ErrorKind::TimedOut {
                             SourceError::Timeout
                         } else {
@@ -391,7 +423,7 @@ fn spawn_pump(
                         };
                         Err(pump_state.fail(PumpFailure::Source(failure)))
                     }
-                    Ok(Some(Ok(chunk))) => {
+                    Some(Ok(chunk)) => {
                         let len = u64::try_from(chunk.len()).unwrap_or(u64::MAX);
                         delivered = delivered.saturating_add(len);
                         if delivered > expected_size {
@@ -738,19 +770,25 @@ fn record_completion(state: &BucketOdmState, key: &str, path: PullPath, result:
 /// Inline write-back of a body the GET handler is already streaming (the
 /// tee secondary). No retry: the bytes cannot be re-read. The caller owns
 /// the singleflight slot; this only writes and accounts.
+///
+/// `idle` is the guard of the [`idle_guarded_body`] the caller teed, so a
+/// write-back that failed because the source stalled is counted as the source
+/// timeout it is rather than as a local write failure. Callers that do not
+/// wrap the source body pass a default guard.
 pub async fn commit_inline(
     state: &Arc<BucketOdmState>,
     key: &str,
     head: SourceHead,
     tags: Option<HashMap<String, String>>,
     body: WriteBackBody,
+    idle: &SourceIdleGuard,
 ) -> Result<WriteBackOutcome, PullError> {
     let Some(write_back) = state.write_back() else {
         let error = PullError::new(PullFailureReason::LocalWrite, "on-demand migration write-back is not installed");
         record_completion(state, key, PullPath::Inline, &Err(error.clone()));
         return Err(error);
     };
-    commit_inline_with(state, write_back, key, head, tags, body).await
+    commit_inline_with(state, write_back, key, head, tags, body, idle).await
 }
 
 /// [`commit_inline`] with an explicit write-back (tests and embedders).
@@ -761,12 +799,16 @@ pub async fn commit_inline_with(
     head: SourceHead,
     tags: Option<HashMap<String, String>>,
     body: WriteBackBody,
+    idle: &SourceIdleGuard,
 ) -> Result<WriteBackOutcome, PullError> {
     let request = WriteBackRequest::new(state, key, head, tags);
-    let result = write_back
-        .put_object(&request, body)
-        .await
-        .map_err(|err| PullError::new(err.reason(), err.to_string()));
+    let result = write_back.put_object(&request, body).await.map_err(|err| {
+        if idle.timed_out() {
+            PullError::new(PullFailureReason::SourceTimeout, SourceError::Timeout.to_string())
+        } else {
+            PullError::new(err.reason(), err.to_string())
+        }
+    });
     let completion = result
         .as_ref()
         .map(|outcome| PullCompletion::Stored(outcome.clone()))
@@ -1516,10 +1558,11 @@ mod tests {
     async fn idle_guarded_body_ends_a_stalled_stream_and_passes_chunks_through() {
         let (tx, rx) = mpsc::channel::<io::Result<Bytes>>(4);
         let body: SourceBody = Box::pin(tokio_stream::wrappers::ReceiverStream::new(rx));
-        let mut guarded = idle_guarded_body(body, Duration::from_secs(2));
+        let (mut guarded, idle) = idle_guarded_body(body, Duration::from_secs(2));
 
         tx.send(Ok(Bytes::from_static(b"chunk"))).await.expect("send a chunk");
         assert_eq!(guarded.next().await.expect("a chunk").expect("chunk is ok"), Bytes::from_static(b"chunk"));
+        assert!(!idle.timed_out(), "a delivered chunk is not a stall");
 
         // The producer never sends again: the guard ends the stream itself.
         let started = tokio::time::Instant::now();
@@ -1530,10 +1573,117 @@ mod tests {
             .expect_err("a stalled body must time out");
         assert_eq!(err.kind(), io::ErrorKind::TimedOut);
         assert!(started.elapsed() >= Duration::from_secs(2));
+        assert!(idle.timed_out(), "the guard reports why the stream ended");
         assert!(guarded.next().await.is_none(), "the stream ends after the timeout");
         drop(tx);
     }
 
+    /// A source that keeps producing, only slowly, must survive: the budget
+    /// is per chunk, not for the whole body.
+    #[tokio::test(start_paused = true)]
+    async fn idle_guarded_body_accepts_a_slow_but_advancing_source() {
+        let (tx, rx) = mpsc::channel::<io::Result<Bytes>>(1);
+        let body: SourceBody = Box::pin(tokio_stream::wrappers::ReceiverStream::new(rx));
+        let (mut guarded, idle) = idle_guarded_body(body, Duration::from_secs(2));
+
+        let producer = tokio::spawn(async move {
+            for _ in 0..5 {
+                tokio::time::sleep(Duration::from_millis(1_500)).await;
+                if tx.send(Ok(Bytes::from_static(b"chunk"))).await.is_err() {
+                    return;
+                }
+            }
+        });
+
+        let mut chunks = 0;
+        while let Some(chunk) = guarded.next().await {
+            chunk.expect("a source that keeps advancing must not time out");
+            chunks += 1;
+        }
+        producer.await.expect("producer task");
+        assert_eq!(chunks, 5);
+        assert!(!idle.timed_out());
+    }
+
+    /// The budget measures the source, not the consumer: a reader that stops
+    /// asking for bytes for far longer than the budget still gets the rest of
+    /// the body once it resumes.
+    #[tokio::test(start_paused = true)]
+    async fn idle_guarded_body_does_not_time_out_a_slow_consumer() {
+        let (tx, rx) = mpsc::channel::<io::Result<Bytes>>(4);
+        let body: SourceBody = Box::pin(tokio_stream::wrappers::ReceiverStream::new(rx));
+        let (mut guarded, idle) = idle_guarded_body(body, Duration::from_secs(2));
+
+        for _ in 0..3 {
+            tx.send(Ok(Bytes::from_static(b"chunk"))).await.expect("send a chunk");
+        }
+        drop(tx);
+
+        let mut chunks = 0;
+        while let Some(chunk) = guarded.next().await {
+            chunk.expect("a stalled consumer must not fail the source");
+            chunks += 1;
+            // Ten times the budget passes between reads.
+            tokio::time::sleep(Duration::from_secs(20)).await;
+        }
+        assert_eq!(chunks, 3);
+        assert!(!idle.timed_out(), "the consumer's own pace is not source idleness");
+    }
+
+    /// The inline path has no pump: the guard is what turns a stalled source
+    /// into a `source_timeout` failure instead of a local write failure.
+    #[tokio::test(start_paused = true)]
+    async fn commit_inline_reports_a_stalled_source_as_a_source_timeout() {
+        let sys = OnDemandMigrationSys::new();
+        let mock = Arc::new(MockWriteBack::default());
+        let mock_dyn: Arc<dyn OdmWriteBack> = mock.clone();
+        sys.set_write_back(mock_dyn);
+        let state = enabled_state(&sys, &config()).await;
+
+        let (tx, rx) = mpsc::channel::<io::Result<Bytes>>(4);
+        let body: SourceBody = Box::pin(tokio_stream::wrappers::ReceiverStream::new(rx));
+        let (mut guarded, idle) = idle_guarded_body(body, Duration::from_secs(1));
+        tx.send(Ok(Bytes::from(body_bytes(100)))).await.expect("send the first chunk");
+        // The source announced 300 bytes and then stops sending; holding the
+        // sender keeps the stream open the way a stalled connection would.
+        let keep_open = tx;
+
+        // Stands in for the tee, which forwards the guarded source to the
+        // write-back and is the reason the write-back only ever sees a plain
+        // body read error.
+        let (relay_tx, relay_rx) = mpsc::channel::<io::Result<Bytes>>(4);
+        tokio::spawn(async move {
+            while let Some(chunk) = guarded.next().await {
+                let failed = chunk.is_err();
+                if relay_tx.send(chunk).await.is_err() || failed {
+                    return;
+                }
+            }
+        });
+        let write_body: WriteBackBody = Box::pin(tokio_stream::wrappers::ReceiverStream::new(relay_rx));
+
+        // The inline path holds the singleflight slot across the commit, the
+        // way `odm_get_inline` does.
+        let PullSlot::Leader(leader) = state.acquire_pull_slot("stalled").await.expect("the first caller leads") else {
+            panic!("the first caller must be the leader");
+        };
+        assert_eq!(state.stats().inflight_pulls(), 1);
+
+        let err = commit_inline(&state, "stalled", head(300), None, write_body, &idle)
+            .await
+            .expect_err("a stalled source must not commit");
+        assert_eq!(err.reason, PullFailureReason::SourceTimeout, "{err}");
+        assert!(idle.timed_out());
+        assert_eq!(failures(&state).get("source_timeout"), Some(&1));
+        assert_eq!(failures(&state).get("local_write"), None, "a stalled source is not a local write failure");
+        assert!(mock.local.lock().get("stalled").is_none(), "nothing is stored");
+
+        leader.complete(Err(err));
+        assert_eq!(state.stats().inflight_pulls(), 0, "the aborted pull releases its slot");
+        assert_eq!(state.inflight_keys(), 0);
+        drop(keep_open);
+    }
+
     #[tokio::test(start_paused = true)]
     async fn stalled_source_body_hits_the_idle_timeout() {
         let sys = OnDemandMigrationSys::new();
@@ -1695,7 +1845,7 @@ mod tests {
 
         let body = body_bytes(300);
         let stream: WriteBackBody = Box::pin(futures::stream::iter(vec![Ok(Bytes::from(body.clone()))]));
-        let outcome = commit_inline(&state, "inline", head(300), None, stream)
+        let outcome = commit_inline(&state, "inline", head(300), None, stream, &SourceIdleGuard::default())
             .await
             .expect("inline commit succeeds");
         assert_eq!(outcome.size, 300);
@@ -1706,7 +1856,7 @@ mod tests {
 
         *mock.forced_put_error.lock() = Some(WriteBackError::Integrity);
         let stream: WriteBackBody = Box::pin(futures::stream::iter(vec![Ok(Bytes::from(body.clone()))]));
-        let err = commit_inline_with(&state, &mock_dyn, "inline", head(300), None, stream)
+        let err = commit_inline_with(&state, &mock_dyn, "inline", head(300), None, stream, &SourceIdleGuard::default())
             .await
             .expect_err("integrity failure surfaces");
         assert_eq!(err.reason, PullFailureReason::EtagMismatch);
@@ -1717,7 +1867,7 @@ mod tests {
             Ok(Bytes::from(body_bytes(100))),
             Err(io::Error::new(io::ErrorKind::BrokenPipe, "tee primary dropped")),
         ]));
-        let err = commit_inline(&state, "torn", head(300), None, stream)
+        let err = commit_inline(&state, "torn", head(300), None, stream, &SourceIdleGuard::default())
             .await
             .expect_err("a broken secondary must fail");
         assert_eq!(err.reason, PullFailureReason::LocalWrite);
@@ -1728,7 +1878,7 @@ mod tests {
         let bare_state = enabled_state(&bare, &config()).await;
         assert!(bare_state.write_back().is_none());
         let stream: WriteBackBody = Box::pin(futures::stream::empty());
-        let err = commit_inline(&bare_state, "x", head(0), None, stream)
+        let err = commit_inline(&bare_state, "x", head(0), None, stream, &SourceIdleGuard::default())
             .await
             .expect_err("no write-back");
         assert_eq!(err.reason, PullFailureReason::LocalWrite);

crates/ecstore/src/bucket/on_demand_migration/source_client.rs

@@ -26,11 +26,10 @@
 //! forwarded: v1 rejects SSE-C source objects outright.
 
 use crate::bucket::remote_s3_client::{
-    PathStyle, RemoteCredentials, RemoteS3ClientError, RemoteS3EndpointSpec, build_remote_s3_config,
+    PathStyle, RemoteCredentials, RemoteS3ClientError, RemoteS3EndpointSpec, RemoteS3RetryPolicy, build_remote_s3_config,
 };
 use crate::storage_api_contracts::range::HTTPRangeSpec;
 use aws_sdk_s3::Client as S3Client;
-use aws_sdk_s3::config::retry::RetryConfig;
 use aws_sdk_s3::error::{ProvideErrorMetadata, SdkError};
 use aws_sdk_s3::operation::get_object::GetObjectOutput;
 use aws_sdk_s3::operation::head_object::HeadObjectOutput;
@@ -150,6 +149,12 @@ pub struct SourceClientSpec {
     pub skip_tls_verify: bool,
     pub ca_cert_pem: Option<String>,
     pub timeouts: SourceTimeouts,
+    /// Wire requests one logical source call may cost. The pull pipeline and
+    /// the backfill job own the retry budget (`pull.rs` `PULL_MAX_RETRIES`,
+    /// `backfill.rs` `LIST_MAX_RETRIES`) and the breaker counts logical calls,
+    /// so ODM declares [`RemoteS3RetryPolicy::Disabled`] and keeps one counted
+    /// failure equal to one request against a struggling source.
+    pub retry: RemoteS3RetryPolicy,
     /// Bytes per second the pull pipeline may consume from this source;
     /// `None` means unlimited. Enforced by the consumer, not by this client.
     pub bandwidth_limit: Option<NonZeroU64>,
@@ -193,6 +198,7 @@ impl SourceClientSpec {
             ca_cert_pem: self.ca_cert_pem.clone(),
             connect_timeout: Some(self.timeouts.connect),
             read_timeout: Some(self.timeouts.read),
+            retry: self.retry,
             user_agent_suffix: USER_AGENT_SUFFIX,
         })
     }
@@ -579,19 +585,11 @@ impl SourceClient {
         Ok(Self::from_config_builder(config, endpoint.endpoint_url(), spec))
     }
 
+    /// `config` must come from [`SourceClientSpec::endpoint_spec`], which is
+    /// where the retry policy that keeps one logical call equal to one wire
+    /// request is declared.
     fn from_config_builder(config: aws_sdk_s3::config::Builder, endpoint: String, spec: &SourceClientSpec) -> Self {
-        // The pull pipeline and the backfill job own the retry budget for a
-        // source call (`pull.rs` PULL_MAX_RETRIES, `backfill.rs`
-        // LIST_MAX_RETRIES), and the breaker counts logical calls. Leaving the
-        // smithy standard policy on top would turn one counted failure into
-        // three wire requests against a source that is already struggling, so
-        // keep one logical call equal to one wire request.
-        let client = S3Client::from_conf(
-            config
-                .retry_config(RetryConfig::disabled())
-                .interceptor(SourceProxyMarkerInterceptor::new())
-                .build(),
-        );
+        let client = S3Client::from_conf(config.interceptor(SourceProxyMarkerInterceptor::new()).build());
         Self {
             client,
             endpoint,
@@ -862,6 +860,7 @@ mod tests {
             }),
             skip_tls_verify: false,
             ca_cert_pem: None,
+            retry: RemoteS3RetryPolicy::Disabled,
             timeouts: SourceTimeouts::default(),
             bandwidth_limit: NonZeroU64::new(1_000_000),
         }

crates/ecstore/src/bucket/on_demand_migration/sys.rs

@@ -48,7 +48,9 @@ use super::negative_cache::NegativeCache;
 use super::pull::{OdmWriteBack, PullQueue};
 use super::source_client::{SourceClient, SourceClientSpec, SourceError, SourceProvider, SourceTimeouts};
 use super::stats::{GaugeGuard, OdmStats, OdmStatsSnapshot, PullFailureReason};
-use crate::bucket::remote_s3_client::{PathStyle as ClientPathStyle, RemoteCredentials, RemoteS3ClientError};
+use crate::bucket::remote_s3_client::{
+    PathStyle as ClientPathStyle, RemoteCredentials, RemoteS3ClientError, RemoteS3RetryPolicy,
+};
 use parking_lot::{Mutex, RwLock};
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
@@ -602,6 +604,10 @@ pub fn source_client_spec(config: &OnDemandMigrationConfig) -> SourceClientSpec
             connect: Duration::from_millis(policy.source_timeout.connect_ms),
             read: Duration::from_millis(policy.source_timeout.first_byte_ms),
         },
+        // The pull pipeline and the backfill job already retry, and the
+        // breaker counts logical calls: an SDK retry on top would triple the
+        // load on a source that is already failing.
+        retry: RemoteS3RetryPolicy::Disabled,
         bandwidth_limit: policy.bandwidth_limit_bytes_per_sec.and_then(NonZeroU64::new),
     }
 }
@@ -1192,6 +1198,11 @@ mod tests {
         assert_eq!(spec.timeouts.connect, Duration::from_millis(1500));
         assert_eq!(spec.timeouts.read, Duration::from_millis(2500));
         assert_eq!(spec.bandwidth_limit, NonZeroU64::new(1 << 20));
+        assert_eq!(
+            spec.retry,
+            RemoteS3RetryPolicy::Disabled,
+            "the pull pipeline owns the retry budget, so one source call is one wire request"
+        );
 
         let mut aws = config(None);
         aws.source.provider = Provider::Aws;

crates/ecstore/src/bucket/remote_s3_client.rs

@@ -17,8 +17,11 @@
 //! Replication targets (`bucket_target_sys`) and the on-demand migration
 //! source client build their remote clients from one neutral
 //! [`RemoteS3EndpointSpec`]: endpoint assembly, credential handling, path-style
-//! selection, custom CA / skip-TLS transports and the outbound SSRF gate all
-//! live here so both callers share exactly one policy. The gate keeps the
+//! selection, custom CA / skip-TLS transports, the SDK retry policy and the
+//! outbound SSRF gate all live here so both callers share exactly one policy.
+//! The retry policy is the one knob the two consumers deliberately disagree
+//! on, so [`RemoteS3EndpointSpec::retry`] is a required field rather than an
+//! inherited SDK default. The gate keeps the
 //! relaxed replication semantics documented in
 //! `docs/operations/outbound-connection-policy.md`: private addresses are
 //! always allowed, loopback only behind `RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET`.
@@ -29,6 +32,7 @@ use aws_sdk_s3::config::Region as SdkRegion;
 use aws_sdk_s3::config::RequestChecksumCalculation;
 use aws_sdk_s3::config::SharedCredentialsProvider;
 use aws_sdk_s3::config::SharedHttpClient;
+use aws_sdk_s3::config::retry::RetryConfig;
 use aws_sdk_s3::{Client as S3Client, Config as S3Config};
 use aws_smithy_http_client::{Builder as SmithyHttpClientBuilder, tls as smithy_tls};
 use aws_smithy_runtime_api::box_error::BoxError;
@@ -80,6 +84,34 @@ impl PathStyle {
     }
 }
 
+/// SDK-level retry policy for a remote client. Retries are invisible to the
+/// caller — one logical call becomes several wire requests — so every consumer
+/// states its own instead of inheriting the SDK default: a caller that already
+/// owns a retry budget would otherwise multiply it against an endpoint that is
+/// by definition already failing.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum RemoteS3RetryPolicy {
+    /// One logical call is exactly one wire request; the caller owns the
+    /// retry budget.
+    Disabled,
+    /// Smithy's standard strategy, capped at `max_attempts` attempts in total
+    /// (the initial request included). Values below 1 are clamped to 1.
+    Standard { max_attempts: u32 },
+}
+
+/// The SDK default replication targets have always run with, written out so a
+/// change to it is a change to this line rather than to a dependency default.
+pub const REPLICATION_TARGET_RETRY_POLICY: RemoteS3RetryPolicy = RemoteS3RetryPolicy::Standard { max_attempts: 3 };
+
+impl RemoteS3RetryPolicy {
+    fn retry_config(self) -> RetryConfig {
+        match self {
+            RemoteS3RetryPolicy::Disabled => RetryConfig::disabled(),
+            RemoteS3RetryPolicy::Standard { max_attempts } => RetryConfig::standard().with_max_attempts(max_attempts.max(1)),
+        }
+    }
+}
+
 /// Static or temporary credentials for a remote endpoint. `expiration` without
 /// a `session_token` is rejected at build time: only STS-style temporary
 /// credentials expire, so that combination is a corrupted configuration
@@ -123,6 +155,9 @@ pub struct RemoteS3EndpointSpec {
     pub ca_cert_pem: Option<String>,
     pub connect_timeout: Option<Duration>,
     pub read_timeout: Option<Duration>,
+    /// How many wire requests one logical call may cost. Every consumer
+    /// declares it; see [`RemoteS3RetryPolicy`].
+    pub retry: RemoteS3RetryPolicy,
     /// Appended to the SDK `User-Agent` (space separated) so the remote side
     /// can identify the caller; empty means no suffix.
     pub user_agent_suffix: &'static str,
@@ -263,7 +298,8 @@ pub(crate) async fn build_remote_s3_config(
         .credentials_provider(SharedCredentialsProvider::new(RemoteTargetCredentialsProvider { credentials: creds }))
         .region(SdkRegion::new(spec.region.clone()))
         .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest())
-        .request_checksum_calculation(replication_request_checksum_calculation());
+        .request_checksum_calculation(replication_request_checksum_calculation())
+        .retry_config(spec.retry.retry_config());
 
     if spec.path_style.force_path_style() {
         config_builder = config_builder.force_path_style(true);
@@ -618,6 +654,7 @@ mod tests {
     use super::*;
     use aws_smithy_runtime_api::http::StatusCode as SmithyStatusCode;
     use std::sync::Mutex;
+    use std::sync::atomic::{AtomicUsize, Ordering};
 
     fn spec(endpoint: &str, secure: bool) -> RemoteS3EndpointSpec {
         RemoteS3EndpointSpec {
@@ -636,6 +673,7 @@ mod tests {
             ca_cert_pem: None,
             connect_timeout: None,
             read_timeout: None,
+            retry: RemoteS3RetryPolicy::Disabled,
             user_agent_suffix: "",
         }
     }
@@ -726,6 +764,66 @@ mod tests {
         assert!(err.to_string().contains("invalid target CA PEM"));
     }
 
+    /// Answers every request with a retryable 503 and counts the wire
+    /// requests one logical call produced.
+    #[derive(Clone, Debug)]
+    struct CountingUnavailableConnector {
+        wire_requests: Arc<AtomicUsize>,
+    }
+
+    impl SmithyHttpConnector for CountingUnavailableConnector {
+        fn call(&self, _request: HttpRequest) -> HttpConnectorFuture {
+            self.wire_requests.fetch_add(1, Ordering::SeqCst);
+            HttpConnectorFuture::ready(Ok(HttpResponse::new(
+                SmithyStatusCode::try_from(503_u16).expect("503 should be a valid response status"),
+                SdkBody::empty(),
+            )))
+        }
+    }
+
+    async fn wire_requests_for_one_failed_call(retry: RemoteS3RetryPolicy) -> usize {
+        let wire_requests = Arc::new(AtomicUsize::new(0));
+        let connector = SharedHttpConnector::new(CountingUnavailableConnector {
+            wire_requests: Arc::clone(&wire_requests),
+        });
+        let http_client = http_client_fn(move |_settings, _components| connector.clone());
+
+        let mut spec = spec("s3.example.com", true);
+        spec.retry = retry;
+        let config = build_remote_s3_config(&spec)
+            .await
+            .expect("spec should build")
+            .http_client(http_client)
+            .build();
+        S3Client::from_conf(config)
+            .head_bucket()
+            .bucket("bucket")
+            .send()
+            .await
+            .expect_err("a 503 must fail the call");
+
+        wire_requests.load(Ordering::SeqCst)
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn retry_policy_decides_how_many_wire_requests_one_call_costs() {
+        assert_eq!(
+            wire_requests_for_one_failed_call(RemoteS3RetryPolicy::Disabled).await,
+            1,
+            "a disabled policy must not amplify one logical call"
+        );
+        assert_eq!(
+            wire_requests_for_one_failed_call(REPLICATION_TARGET_RETRY_POLICY).await,
+            3,
+            "replication targets keep the three-attempt SDK default"
+        );
+        assert_eq!(
+            wire_requests_for_one_failed_call(RemoteS3RetryPolicy::Standard { max_attempts: 0 }).await,
+            1,
+            "a zero attempt budget is clamped to the initial request"
+        );
+    }
+
     #[test]
     fn path_style_auto_and_path_force_path_style() {
         assert!(PathStyle::Auto.force_path_style());