fix(odm): close three on-demand migration follow-ups from the e2e matrix (#7101)

crates/ecstore/src/api/mod.rs

@@ -163,7 +163,7 @@ 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,
+            WriteBackOutcome, WriteBackPart, WriteBackRequest, commit_inline, commit_inline_with, idle_guarded_body,
         };
         pub mod backfill {
             pub use crate::bucket::on_demand_migration::backfill::{

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

@@ -42,7 +42,7 @@ 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,
+    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

@@ -149,6 +149,30 @@ 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>>;
 
+/// 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))),
+        }
+    }))
+}
+
 /// Body handed to the write-back; `Sync` because the app-layer put path
 /// wraps it into an S3 streaming blob.
 pub type WriteBackBody = Pin<Box<dyn Stream<Item = io::Result<Bytes>> + Send + Sync + 'static>>;
@@ -262,7 +286,8 @@ impl WriteBackError {
         match self {
             WriteBackError::Integrity => PullFailureReason::EtagMismatch,
             WriteBackError::Unsupported(_) => PullFailureReason::SourceUnsupported,
-            WriteBackError::Quota(_) | WriteBackError::Local(_) => PullFailureReason::LocalWrite,
+            WriteBackError::Quota(_) => PullFailureReason::Quota,
+            WriteBackError::Local(_) => PullFailureReason::LocalWrite,
         }
     }
 }
@@ -1487,6 +1512,28 @@ mod tests {
         assert_eq!(failures(&state).get("source_connect"), Some(&1));
     }
 
+    #[tokio::test(start_paused = true)]
+    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));
+
+        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"));
+
+        // The producer never sends again: the guard ends the stream itself.
+        let started = tokio::time::Instant::now();
+        let err = guarded
+            .next()
+            .await
+            .expect("the guard yields the timeout")
+            .expect_err("a stalled body must time out");
+        assert_eq!(err.kind(), io::ErrorKind::TimedOut);
+        assert!(started.elapsed() >= Duration::from_secs(2));
+        assert!(guarded.next().await.is_none(), "the stream ends after the timeout");
+        drop(tx);
+    }
+
     #[tokio::test(start_paused = true)]
     async fn stalled_source_body_hits_the_idle_timeout() {
         let sys = OnDemandMigrationSys::new();
@@ -1744,7 +1791,7 @@ mod tests {
         assert_eq!(PullReason::Backfill.path(), PullPath::Backfill);
         assert_eq!(PullReason::RangeGet.as_str(), "range_get");
         assert_eq!(WriteBackError::Integrity.reason(), PullFailureReason::EtagMismatch);
-        assert_eq!(WriteBackError::Quota("full".into()).reason(), PullFailureReason::LocalWrite);
+        assert_eq!(WriteBackError::Quota("full".into()).reason(), PullFailureReason::Quota);
         assert_eq!(WriteBackError::Local("x".into()).reason(), PullFailureReason::LocalWrite);
         assert_eq!(WriteBackError::Unsupported("x".into()).reason(), PullFailureReason::SourceUnsupported);
         for (retry, base) in PULL_RETRY_BASE_DELAYS.iter().enumerate() {

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

@@ -30,6 +30,7 @@ use crate::bucket::remote_s3_client::{
 };
 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;
@@ -579,7 +580,18 @@ impl SourceClient {
     }
 
     fn from_config_builder(config: aws_sdk_s3::config::Builder, endpoint: String, spec: &SourceClientSpec) -> Self {
-        let client = S3Client::from_conf(config.interceptor(SourceProxyMarkerInterceptor::new()).build());
+        // 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(),
+        );
         Self {
             client,
             endpoint,
@@ -736,7 +748,6 @@ impl SourceClient {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use aws_sdk_s3::config::retry::RetryConfig;
     use aws_smithy_runtime_api::client::http::{HttpConnector, HttpConnectorFuture, SharedHttpConnector, http_client_fn};
     use aws_smithy_runtime_api::client::orchestrator::HttpRequest;
     use aws_smithy_runtime_api::client::result::ConnectorError;
@@ -867,8 +878,7 @@ mod tests {
         let config = build_remote_s3_config(&endpoint)
             .await
             .expect("test spec should build")
-            .http_client(http_client)
-            .retry_config(RetryConfig::disabled());
+            .http_client(http_client);
         (SourceClient::from_config_builder(config, endpoint.endpoint_url(), spec), requests)
     }
 

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

@@ -123,6 +123,8 @@ pub enum PullFailureReason {
     EtagMismatch,
     /// The local write (internal PUT) failed.
     LocalWrite,
+    /// The bucket quota rejected the write-back.
+    Quota,
     /// The bucket state was removed or the process is shutting down.
     Canceled,
     /// The background pull queue was full.
@@ -130,7 +132,7 @@ pub enum PullFailureReason {
 }
 
 impl PullFailureReason {
-    pub const ALL: [PullFailureReason; 12] = [
+    pub const ALL: [PullFailureReason; 13] = [
         PullFailureReason::SourceNotFound,
         PullFailureReason::SourceAccessDenied,
         PullFailureReason::SourceThrottled,
@@ -141,6 +143,7 @@ impl PullFailureReason {
         PullFailureReason::SourceOther,
         PullFailureReason::EtagMismatch,
         PullFailureReason::LocalWrite,
+        PullFailureReason::Quota,
         PullFailureReason::Canceled,
         PullFailureReason::QueueFull,
     ];
@@ -157,6 +160,7 @@ impl PullFailureReason {
             PullFailureReason::SourceOther => "source_other",
             PullFailureReason::EtagMismatch => "etag_mismatch",
             PullFailureReason::LocalWrite => "local_write",
+            PullFailureReason::Quota => "quota",
             PullFailureReason::Canceled => "canceled",
             PullFailureReason::QueueFull => "queue_full",
         }
@@ -457,7 +461,7 @@ mod tests {
             "pulled_bytes_total": 4096,
             "pulled_objects_total": { "backfill": 0, "background": 0, "inline": 1 },
             "pull_failures_total": {
-                "canceled": 0, "etag_mismatch": 0, "local_write": 0, "queue_full": 0,
+                "canceled": 0, "etag_mismatch": 0, "local_write": 0, "queue_full": 0, "quota": 0,
                 "source_access_denied": 0, "source_connect": 0, "source_not_found": 0, "source_other": 0,
                 "source_server_error": 0, "source_throttled": 0, "source_timeout": 1, "source_unsupported": 0
             },