From 3005efe845d2fee9dc84c1c9d09211bd43f5647a Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Fri, 4 Sep 2026 02:24:53 +0800 Subject: [PATCH] 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/e2e_test/src/fake_s3_target/README.md | 2 +- crates/e2e_test/src/fake_s3_target/mod.rs | 31 +++ .../src/on_demand_migration/fault_test.rs | 139 ++++++++--- crates/ecstore/src/api/mod.rs | 7 +- .../ecstore/src/bucket/bucket_target_sys.rs | 23 +- .../src/bucket/on_demand_migration/mod.rs | 4 +- .../src/bucket/on_demand_migration/pull.rs | 224 +++++++++++++++--- .../on_demand_migration/source_client.rs | 27 +-- .../src/bucket/on_demand_migration/sys.rs | 13 +- crates/ecstore/src/bucket/remote_s3_client.rs | 104 +++++++- docs/operations/on-demand-migration.md | 6 +- .../src/admin/handlers/on_demand_migration.rs | 7 +- rustfs/src/admin/storage_api.rs | 1 + rustfs/src/app/object/get.rs | 7 +- 14 files changed, 486 insertions(+), 109 deletions(-) diff --git a/crates/e2e_test/src/fake_s3_target/README.md b/crates/e2e_test/src/fake_s3_target/README.md index 533f97fb5..bcd2e8ca6 100644 --- a/crates/e2e_test/src/fake_s3_target/README.md +++ b/crates/e2e_test/src/fake_s3_target/README.md @@ -12,6 +12,6 @@ Supported data operations are HeadBucket, GetBucketVersioning, ListObjectsV2, PU ListObjectsV2 lists current versions only (a key whose newest version is a delete marker is hidden) in byte order and supports `prefix`, `delimiter`, `max-keys` (clamped to 1000), `start-after`, and `continuation-token`; common prefixes count toward `max-keys`, `IsTruncated` / `NextContinuationToken` / `KeyCount` follow S3, and continuation tokens are opaque. `encoding-type` and `fetch-owner` are accepted but ignored, and ListObjects (v1) is not implemented. GET and HEAD honor `Range` in the `bytes=first-last`, `bytes=first-`, and `bytes=-suffix` forms with a 206 status, exact `Content-Range`, and `Accept-Ranges: bytes`; unsatisfiable ranges answer 416 `InvalidRange` with `Content-Range: bytes */`. PUT and CreateMultipartUpload accept `Content-Type`, `Content-Encoding`, `Content-Disposition`, `Content-Language`, `Cache-Control`, `Expires`, and `x-amz-meta-*` (names stored lowercased), and HEAD/GET replay them verbatim together with `Last-Modified` and the ETag (hex MD5 for single PUTs, `-` for multipart objects). `put_seed_object` stores an object directly, bypassing the wire, the fault script, and the journal, so a source can be seeded without polluting the assertions a scenario later makes. -Fault actions cover HTTP 401/403/503 responses (`Status`), any 4xx/5xx status paired with the matching S3 error code (`ResponseStatus`), pre-dispatch delay, holding a fully computed successful response before its first byte (`Stall`), connection abort when a logical request-body threshold is reached, GetObject bodies cut off after N bytes while `Content-Length` announces the full size (`TruncateBodyAt`), streaming slow drain, and a deliberately wrong response ETag (including multipart-complete XML). `requests()` returns the ordered, credential-free request journal for assertions and `count_requests(operation, key)` counts entries for one exact key. Each record journals the `Range` and `User-Agent` request headers, the ListObjectsV2 `prefix` and `continuation-token` query values, a `TransportSnapshot` — whether the body was announced as `aws-chunked`, the verbatim `Content-MD5`, the sorted `x-amz-checksum-*` / `x-amz-sdk-checksum-algorithm` header names, and whether any `x-amz-object-lock-*` header was present — and a `ProxyHeaderSnapshot` — the read-proxy anti-loop marker (`x-{rustfs,minio}-source-proxy-request`), the replication-check exemption header, and the client SSE-C header family (algorithm and key-MD5 values; for the key itself only its presence) — so proxy tests can pin the exact wire contract. +Fault actions cover HTTP 401/403/503 responses (`Status`), any 4xx/5xx status paired with the matching S3 error code (`ResponseStatus`), pre-dispatch delay, holding a fully computed successful response before its first byte (`Stall`), connection abort when a logical request-body threshold is reached, GetObject bodies cut off after N bytes while `Content-Length` announces the full size (`TruncateBodyAt`), GetObject bodies delivered in fixed slices with a pause between them (`SlowSendBody`, a mid-body stall rather than a first-byte one), streaming slow drain, and a deliberately wrong response ETag (including multipart-complete XML). `requests()` returns the ordered, credential-free request journal for assertions and `count_requests(operation, key)` counts entries for one exact key. Each record journals the `Range` and `User-Agent` request headers, the ListObjectsV2 `prefix` and `continuation-token` query values, a `TransportSnapshot` — whether the body was announced as `aws-chunked`, the verbatim `Content-MD5`, the sorted `x-amz-checksum-*` / `x-amz-sdk-checksum-algorithm` header names, and whether any `x-amz-object-lock-*` header was present — and a `ProxyHeaderSnapshot` — the read-proxy anti-loop marker (`x-{rustfs,minio}-source-proxy-request`), the replication-check exemption header, and the client SSE-C header family (algorithm and key-MD5 values; for the key itself only its presence) — so proxy tests can pin the exact wire contract. The listener is loopback-only. It admits at most 64 active connections and two concurrently buffered request bodies; authenticated multipart-complete XML collection and assembly take both body permits. Keep-alive is disabled, request-header reads are bounded to 30 seconds, a parsed request is bounded to 65 seconds, and the complete connection lifetime is bounded to 100 seconds. It retains at most 256 buckets, 4,096 journal entries, 4,096 scripted faults, 4,096 object versions, 256 multipart uploads, and 10,000 multipart parts. Retained identifiers are capped at 1 KiB, user metadata at 2 KiB, and content type and each standard object header at 1 KiB. By default a PUT or uploaded part is capped at 64 MiB and a completed multipart object and all stored object/part data are capped at 128 MiB; `FakeS3Target::start_with_options(FakeS3TargetOptions { max_object_bytes })` raises the object cap up to 256 MiB, and the total budget then becomes twice the object cap (never below 128 MiB). Body drain, body-permit waits, delay, stall, and slow-drain execution are bounded to 30 seconds; each slow-drain slice delay must be below that bound. diff --git a/crates/e2e_test/src/fake_s3_target/mod.rs b/crates/e2e_test/src/fake_s3_target/mod.rs index 7a04419cf..65f916fd9 100644 --- a/crates/e2e_test/src/fake_s3_target/mod.rs +++ b/crates/e2e_test/src/fake_s3_target/mod.rs @@ -266,6 +266,12 @@ pub enum FaultAction { /// body bytes, then abort the connection so the client observes a short /// read. Ignored by every other operation. TruncateBodyAt(usize), + /// GetObject only: deliver the response body in `chunk_bytes` slices, + /// sleeping `delay` between them. The head and the first slice leave + /// immediately, so this is a mid-body stall rather than a first-byte one; + /// a `delay` above the reader's idle budget is what a source that stops + /// pushing bytes looks like. Ignored by every other operation. + SlowSendBody { chunk_bytes: usize, delay: Duration }, /// Apply the request normally, then hold the complete response (status /// line included) for the duration before the first byte is written — /// the first-byte-timeout scenario. Error responses are not held. @@ -994,6 +1000,9 @@ fn validate_fault_action(action: &FaultAction) { if let FaultAction::SlowDrain { chunk_bytes: 0, .. } = action { panic!("slow-drain chunk size must be non-zero"); } + if let FaultAction::SlowSendBody { chunk_bytes: 0, .. } = action { + panic!("slow-send chunk size must be non-zero"); + } match action { FaultAction::Delay(duration) if *duration > MAX_FAULT_DURATION => { panic!("fault delay must not exceed 30 seconds"); @@ -1004,6 +1013,9 @@ fn validate_fault_action(action: &FaultAction) { FaultAction::SlowDrain { delay, .. } if *delay >= MAX_FAULT_DURATION => { panic!("slow-drain slice delay must be below 30 seconds"); } + FaultAction::SlowSendBody { delay, .. } if *delay >= MAX_FAULT_DURATION => { + panic!("slow-send slice delay must be below 30 seconds"); + } FaultAction::ResponseStatus(code) if !(400..=599).contains(code) => { panic!("scripted response status must be a 4xx or 5xx code"); } @@ -1409,6 +1421,7 @@ async fn apply_non_body_fault(fault: Option<&RequestFault>, control: &Mutex Ok(()), } @@ -1457,6 +1470,7 @@ async fn collect_stream( Some(FaultAction::WrongEtag) | Some(FaultAction::DisconnectAfterResponse) | Some(FaultAction::TruncateBodyAt(_)) + | Some(FaultAction::SlowSendBody { .. }) | Some(FaultAction::Stall(_)) | None => {} } @@ -1524,6 +1538,22 @@ fn truncated_body(body: Bytes, truncate_at: usize) -> StreamingBlob { })) } +/// GetObject body delivered in `chunk_bytes` slices with `delay` between +/// them. The head and the first slice are written immediately, so the client +/// starts reading and then observes the source going quiet mid-body. +fn slow_sent_body(body: Bytes, chunk_bytes: usize, delay: Duration) -> StreamingBlob { + StreamingBlob::wrap(futures::stream::unfold((body, true), move |(mut rest, first)| async move { + if rest.is_empty() { + return None; + } + if !first { + sleep(delay).await; + } + let chunk = rest.split_to(chunk_bytes.min(rest.len())); + Some((Ok::(chunk), (rest, false))) + })) +} + async fn assemble_multipart( parts: Vec<(Bytes, [u8; 16])>, total_len: usize, @@ -2225,6 +2255,7 @@ impl S3 for FakeBackend { let body = version.body.slice(served.range.clone()); let body = match fault.as_ref().map(|fault| &fault.action) { Some(FaultAction::TruncateBodyAt(truncate_at)) => truncated_body(body, *truncate_at), + Some(FaultAction::SlowSendBody { chunk_bytes, delay }) => slow_sent_body(body, *chunk_bytes, *delay), _ => StreamingBlob::from(body), }; let mut response = S3Response::new(GetObjectOutput { diff --git a/crates/e2e_test/src/on_demand_migration/fault_test.rs b/crates/e2e_test/src/on_demand_migration/fault_test.rs index b51ea370b..1d9f5f9b8 100644 --- a/crates/e2e_test/src/on_demand_migration/fault_test.rs +++ b/crates/e2e_test/src/on_demand_migration/fault_test.rs @@ -14,8 +14,8 @@ //! Source-failure scenarios for on-demand migration (rustfs/backlog#2158): //! access denied, the circuit breaker, first-byte and mid-body stream -//! failures, ETag integrity, the negative cache, and an unsupported -//! (SSE-C) source object. +//! failures (a cut body and a stalled one), ETag integrity, the negative +//! cache, and an unsupported (SSE-C) source object. //! //! Every case asserts what the source was asked for, not only what the //! client received: a fault that silently turned into a second source @@ -162,10 +162,10 @@ async fn test_odm_source_access_denied_propagates_without_opening_the_breaker() /// window closes it again. The open window is a compiled-in 30 s constant /// (`BREAKER_OPEN_DURATION`), so this case waits in real time. /// -/// One ODM source call is several wire requests: the SDK retries a 503 on -/// its own, and only the exhausted call counts as one breaker failure. The -/// script is therefore deep enough to cover every retry, and the open state -/// is waited for instead of being predicted from a request count. +/// The source client disables SDK retries, so one logical source call is +/// exactly one wire request: the script is exactly as deep as the number of +/// breaker failures it has to produce, and the scripted fault count and the +/// observed source request count must agree. #[tokio::test] async fn test_odm_repeated_source_errors_open_the_breaker_and_recover() -> TestResult { let bucket = "odm-fault-breaker"; @@ -175,9 +175,8 @@ async fn test_odm_repeated_source_errors_open_the_breaker_and_recover() -> TestR env.seed_source(SOURCE_BUCKET, &[SeedObject::new(key, body.clone())]); env.source - .inject_for_key(Operation::HeadObject, key, FaultAction::ResponseStatus(503), 200); - let mut opened = false; - for attempt in 1..=BREAKER_FAILURE_THRESHOLD * 2 { + .inject_for_key(Operation::HeadObject, key, FaultAction::ResponseStatus(503), BREAKER_FAILURE_THRESHOLD); + for attempt in 1..=BREAKER_FAILURE_THRESHOLD { let response = env.raw_get(bucket, key).await?; assert_eq!( response.status, @@ -185,21 +184,19 @@ async fn test_odm_repeated_source_errors_open_the_breaker_and_recover() -> TestR "attempt {attempt}: {}", String::from_utf8_lossy(&response.body) ); - if env - .status_json(bucket) + } + assert_eq!( + env.status_json(bucket) .await? .pointer("/breaker/state") - .and_then(|v| v.as_str()) - == Some("open") - { - opened = true; - break; - } - } - assert!(opened, "consecutive source failures must open the breaker"); - assert!( - env.source.count_requests(Operation::HeadObject, key) >= BREAKER_FAILURE_THRESHOLD, - "each counted failure is at least one source request" + .and_then(|v| v.as_str()), + Some("open"), + "the threshold of consecutive source failures must open the breaker" + ); + assert_eq!( + env.source.count_requests(Operation::HeadObject, key), + BREAKER_FAILURE_THRESHOLD, + "every counted failure is exactly one source request" ); // With the script cleared, the only thing that can still fail a read is @@ -251,8 +248,8 @@ async fn test_odm_repeated_source_errors_open_the_breaker_and_recover() -> TestR } /// Case 3: a source that holds the response past `first_byte_ms` is a -/// timeout, and the client never sees a 200 head. Every attempt the SDK -/// makes on its own is stalled too, so the ODM call really does give up. +/// timeout, and the client never sees a 200 head. One logical source call is +/// one wire request, so a single scripted stall is enough to fail the read. #[tokio::test] async fn test_odm_source_stall_times_out_before_the_first_byte() -> TestResult { let bucket = "odm-fault-stall"; @@ -264,20 +261,19 @@ async fn test_odm_source_stall_times_out_before_the_first_byte() -> TestResult { env.seed_source(SOURCE_BUCKET, &[SeedObject::new(key, payload(4096))]); env.source - .inject_for_key(Operation::HeadObject, key, FaultAction::Stall(Duration::from_secs(5)), 8); + .inject_for_key(Operation::HeadObject, key, FaultAction::Stall(Duration::from_secs(5)), 1); let started = Instant::now(); let response = env.raw_get(bucket, key).await?; let elapsed = started.elapsed(); assert_eq!(response.status, SOURCE_UNAVAILABLE_STATUS, "{}", String::from_utf8_lossy(&response.body)); - assert!( - elapsed < Duration::from_secs(30), - "the read timeout must cut every attempt short, took {elapsed:?}" + assert_eq!( + env.source.count_requests(Operation::HeadObject, key), + 1, + "the stalled HEAD is the only source request" ); - let attempts = env.source.count_requests(Operation::HeadObject, key); - assert!(attempts >= 1, "the stalled HEAD is the only source request"); assert!( - elapsed < Duration::from_secs(5) * u32::try_from(attempts).unwrap_or(1), - "no attempt waited the stall out ({attempts} attempts in {elapsed:?})" + elapsed < Duration::from_secs(5), + "the read timeout must cut the attempt short, took {elapsed:?}" ); assert_eq!( env.source.count_requests(Operation::GetObject, key), @@ -336,7 +332,78 @@ async fn test_odm_inline_pull_aborts_when_the_source_body_is_cut() -> TestResult Ok(()) } -/// Case 5: the background pull of a large object hits a cut body, counts the +/// Case 5: the source answers, sends part of the body and then goes quiet +/// for longer than `source_timeout.idle_ms`. The inline tee must end both +/// ends: the client gets a short read rather than a silently truncated 200, +/// the pull is counted as a source timeout, and nothing (object or multipart +/// upload) is left behind locally. +#[tokio::test] +async fn test_odm_inline_pull_aborts_when_the_source_body_stalls() -> TestResult { + let bucket = "odm-fault-inline-stall"; + const IDLE_MS: u64 = 1_000; + let idle = Duration::from_millis(IDLE_MS); + let env = start_configured_env(bucket, SOURCE_BUCKET, |spec| { + spec.policy.source_timeout.idle_ms = IDLE_MS; + }) + .await?; + let key = "stall/inline.bin"; + let body = payload(256 * 1024); + env.seed_source(SOURCE_BUCKET, &[SeedObject::new(key, body.clone())]); + + // The head and the first slice arrive at once; the source then pauses for + // four times the idle budget, which is what a stalled source looks like. + env.source.inject_for_key( + Operation::GetObject, + key, + FaultAction::SlowSendBody { + chunk_bytes: 32 * 1024, + delay: idle * 4, + }, + 1, + ); + let started = Instant::now(); + env.raw_get(bucket, key) + .await + .expect_err("a stalled source body must not read back as a complete object"); + let elapsed = started.elapsed(); + assert!( + elapsed < idle * 4, + "the idle budget, not the source's own pause, must end the read (took {elapsed:?})" + ); + + assert_eq!(env.source.count_requests(Operation::HeadObject, key), 1); + assert_eq!( + env.source.count_requests(Operation::GetObject, key), + 1, + "an aborted inline pull is not retried on the same request" + ); + env.wait_for_status_counter(bucket, "/counters/pull_failures_total/source_timeout", 1, SETTLE) + .await?; + // The leader releases its slot just after it records the failure. + let deadline = Instant::now() + SETTLE; + loop { + let inflight = env + .status_json(bucket) + .await? + .pointer("/inflight_pulls") + .and_then(|value| value.as_u64()); + if inflight == Some(0) { + break; + } + assert!(Instant::now() < deadline, "the aborted pull never released its slot: {inflight:?}"); + tokio::time::sleep(Duration::from_millis(200)).await; + } + env.assert_local_absent(bucket, key).await; + let uploads = env.client.list_multipart_uploads().bucket(bucket).send().await?; + assert!( + uploads.uploads().is_empty(), + "a stalled pull leaves no multipart upload: {:?}", + uploads.uploads() + ); + Ok(()) +} + +/// Case 6: the background pull of a large object hits a cut body, counts the /// failure, and the retry stores the object. #[tokio::test] async fn test_odm_background_pull_retries_a_truncated_source_body() -> TestResult { @@ -387,7 +454,7 @@ async fn test_odm_background_pull_retries_a_truncated_source_body() -> TestResul Ok(()) } -/// Case 6: the source advertises an ETag its bytes do not match. The client +/// Case 7: the source advertises an ETag its bytes do not match. The client /// still gets every byte; the write-back is discarded as an integrity /// failure and nothing is stored. #[tokio::test] @@ -415,7 +482,7 @@ async fn test_odm_wrong_source_etag_discards_the_write_back() -> TestResult { Ok(()) } -/// Case 7: a source miss is remembered for `negative_cache_ttl_secs`, and +/// Case 8: a source miss is remembered for `negative_cache_ttl_secs`, and /// re-checked once the entry expires. #[tokio::test] async fn test_odm_source_not_found_is_negative_cached_for_the_ttl() -> TestResult { @@ -455,7 +522,7 @@ async fn test_odm_source_not_found_is_negative_cached_for_the_ttl() -> TestResul Ok(()) } -/// Case 8: an SSE-C source object cannot be migrated (the key belongs to the +/// Case 9: an SSE-C source object cannot be migrated (the key belongs to the /// source's client), so the read fails as unsupported without a body read. #[tokio::test] async fn test_odm_ssec_source_object_is_unsupported() -> TestResult { diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 1e63ed571..9febf4071 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/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, }; } diff --git a/crates/ecstore/src/bucket/bucket_target_sys.rs b/crates/ecstore/src/bucket/bucket_target_sys.rs index 6932784f9..dca41cf72 100644 --- a/crates/ecstore/src/bucket/bucket_target_sys.rs +++ b/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"); diff --git a/crates/ecstore/src/bucket/on_demand_migration/mod.rs b/crates/ecstore/src/bucket/on_demand_migration/mod.rs index 705a3f51e..1672bf52b 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/mod.rs +++ b/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, diff --git a/crates/ecstore/src/bucket/on_demand_migration/pull.rs b/crates/ecstore/src/bucket/on_demand_migration/pull.rs index f46b9c2b5..60f7145a2 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/pull.rs +++ b/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> + 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, +} + +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>, Arc) { + 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, key: &str, head: SourceHead, tags: Option>, body: WriteBackBody, + idle: &SourceIdleGuard, ) -> Result { 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>, body: WriteBackBody, + idle: &SourceIdleGuard, ) -> Result { 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::>(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::>(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::>(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 = mock.clone(); + sys.set_write_back(mock_dyn); + let state = enabled_state(&sys, &config()).await; + + let (tx, rx) = mpsc::channel::>(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::>(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); diff --git a/crates/ecstore/src/bucket/on_demand_migration/source_client.rs b/crates/ecstore/src/bucket/on_demand_migration/source_client.rs index 7164ef7f0..c35040349 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/source_client.rs +++ b/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, 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, @@ -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), } diff --git a/crates/ecstore/src/bucket/on_demand_migration/sys.rs b/crates/ecstore/src/bucket/on_demand_migration/sys.rs index 5ee654618..e40683f73 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/sys.rs +++ b/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; diff --git a/crates/ecstore/src/bucket/remote_s3_client.rs b/crates/ecstore/src/bucket/remote_s3_client.rs index 58a311a04..20d3d0365 100644 --- a/crates/ecstore/src/bucket/remote_s3_client.rs +++ b/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, pub connect_timeout: Option, pub read_timeout: Option, + /// 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, + } + + 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()); diff --git a/docs/operations/on-demand-migration.md b/docs/operations/on-demand-migration.md index 196d2e810..178aab57d 100644 --- a/docs/operations/on-demand-migration.md +++ b/docs/operations/on-demand-migration.md @@ -165,6 +165,7 @@ Behaviour a client can observe. The "Test" column names the case that pins it: ` | Source answers 404 | 404 to the client and the key is negative-cached for `negative_cache_ttl_secs` | `get_basic_test.rs::get_source_not_found_is_404_and_negative_cached`, `fault_test.rs::test_odm_source_not_found_is_negative_cached_for_the_ttl` | | Source answers 403 | 424 (or 404 under `not_found`); the breaker is **not** opened — a credential problem is not a transient one | `fault_test.rs::test_odm_source_access_denied_propagates_without_opening_the_breaker` | | Repeated source 5xx / timeouts | The breaker opens, GETs answer per `source_error` and HEADs answer 404; a probe closes it again after the open window | `fault_test.rs::test_odm_repeated_source_errors_open_the_breaker_and_recover`, `head.rs::odm_head_source_errors_follow_policy_and_open_the_breaker` | +| Source stalls mid-body on an inline pull | `source_timeout.idle_ms` ends both tee ends: the client's read fails instead of receiving a silently truncated 200, the pull counts under `source_timeout`, and no object or multipart upload is left behind | `fault_test.rs::test_odm_inline_pull_aborts_when_the_source_body_stalls`, `pull.rs::commit_inline_reports_a_stalled_source_as_a_source_timeout` | | Bucket default SSE | The pulled object is stored encrypted and reads back in plaintext; the ETag override is dropped, and the source ETag is kept in metadata | `interaction_test.rs::test_odm_pulled_object_uses_bucket_default_encryption`, `on_demand_migration_put.rs::write_back_under_bucket_default_sse_stores_ciphertext_and_records_source_etag` | | Object Lock bucket | The pulled object inherits the bucket's default retention | `interaction_test.rs::test_odm_pulled_object_inherits_object_lock_retention` | | Bucket quota exceeded | The client is still served from the source; the write-back is rejected, nothing is stored, and the failure counts under `quota` | `interaction_test.rs::test_odm_write_back_respects_the_bucket_quota`, `on_demand_migration_put.rs::write_back_reports_a_full_bucket_quota` | @@ -210,7 +211,8 @@ Five provenance keys are written on every pulled object under both internal pref | Concurrency limit | Local write amplification | `max_concurrent_pulls` permits shared by inline and background pulls | | Bounded queue | Unbounded memory on a burst | `pull_queue_capacity` waiting jobs; overflow is counted as `queue_full` and never fails a client response | | Bandwidth limit | Source and network saturation | `bandwidth_limit_bytes_per_sec` (minimum 64 KiB/s) on the source client | -| Retry budget | Transient source blips | Background pulls retry a retryable failure up to 3 times (1 s / 4 s / 16 s plus jitter). Inline pulls never retry: the bytes are already on their way to the client | +| Retry budget | Transient source blips | Background pulls retry a retryable failure up to 3 times (1 s / 4 s / 16 s plus jitter). Inline pulls never retry: the bytes are already on their way to the client. The SDK retry policy on the source client is disabled (`RemoteS3RetryPolicy::Disabled`), so this is the only retry budget and one logical source call is exactly one wire request — replication targets keep the SDK's three attempts, declared on their own spec | +| Idle timeout | A source that answers and then goes quiet mid-body | `source_timeout.idle_ms` per body chunk on both paths. The budget measures the source read, upstream of the inline tee, so a slow client is never mistaken for an idle source; when it fires the client stream ends in an error and the write-back is discarded | | Anti-loop marker | Migration chains between RustFS/MinIO deployments | Every source request carries `x-rustfs-source-proxy-request` and `x-minio-source-proxy-request`; a request carrying it is always answered locally | | Outbound endpoint policy | SSRF | See [outbound-connection-policy.md](outbound-connection-policy.md) | @@ -291,7 +293,7 @@ sum by (bucket, reason) (rate(rustfs_on_demand_migration_pull_failures_total[5m] **`NoSuchBucket` from the source, or a 301/307 redirect.** The addressing style is wrong: a virtual-host request against a path-style-only server looks like a missing bucket, and a path-style request against AWS gets redirected. Set `source.path_style` explicitly instead of relying on `auto`. -**Large objects are served but never stored.** Check `pull_failures_total`: `queue_full` means bursts exceed `pull_queue_capacity` (raise it, or raise `max_concurrent_pulls`); `quota` means the bucket quota rejected the write-back and `local_write` a genuine local write failure; `etag_mismatch` means the source body did not match the ETag the source advertised. The client response is served in all three cases — only the local copy is missing. +**Large objects are served but never stored.** Check `pull_failures_total`: `queue_full` means bursts exceed `pull_queue_capacity` (raise it, or raise `max_concurrent_pulls`); `quota` means the bucket quota rejected the write-back and `local_write` a genuine local write failure; `etag_mismatch` means the source body did not match the ETag the source advertised. The client response is served in all these cases — only the local copy is missing. `source_timeout` is different: the source stopped sending mid-body, so the client's own read fails too and nothing is stored. **The first read of a key is slow, later ones are fast.** Expected: the first read pays a source HEAD plus a source GET. Watch `source_latency_seconds_*` for the source's contribution and lower `inline_max_bytes` if teeing large objects is hurting first-byte latency. diff --git a/rustfs/src/admin/handlers/on_demand_migration.rs b/rustfs/src/admin/handlers/on_demand_migration.rs index ea2017f36..5fdf3b2f4 100644 --- a/rustfs/src/admin/handlers/on_demand_migration.rs +++ b/rustfs/src/admin/handlers/on_demand_migration.rs @@ -47,7 +47,9 @@ use crate::admin::storage_api::bucket::on_demand_migration::source_client::{ use crate::admin::storage_api::bucket::on_demand_migration::{ OdmBucketSnapshot, OnDemandMigrationConfig, OnDemandMigrationConfigError, OnDemandMigrationSys, PathStyle, ValidationContext, }; -use crate::admin::storage_api::bucket::remote_s3_client::{PathStyle as RemotePathStyle, RemoteCredentials, RemoteS3ClientError}; +use crate::admin::storage_api::bucket::remote_s3_client::{ + PathStyle as RemotePathStyle, RemoteCredentials, RemoteS3ClientError, RemoteS3RetryPolicy, +}; use crate::admin::storage_api::contract::bucket::{BucketOperations as _, BucketOptions}; use crate::admin::storage_api::error::StorageError; use crate::admin::storage_api::s3::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, error as admin_s3_error}; @@ -615,6 +617,9 @@ pub(crate) fn source_client_spec(config: &OnDemandMigrationConfig) -> SourceClie connect: Duration::from_millis(timeout.connect_ms), read: Duration::from_millis(timeout.first_byte_ms), }, + // The probe reports the source's own answer; an SDK retry would hide + // a flapping source behind a success and triple the probe's cost. + retry: RemoteS3RetryPolicy::Disabled, bandwidth_limit: config.policy.bandwidth_limit_bytes_per_sec.and_then(NonZeroU64::new), } } diff --git a/rustfs/src/admin/storage_api.rs b/rustfs/src/admin/storage_api.rs index de8154d93..8e23fa0e0 100644 --- a/rustfs/src/admin/storage_api.rs +++ b/rustfs/src/admin/storage_api.rs @@ -315,6 +315,7 @@ pub(crate) mod remote_s3_client { pub(crate) type PathStyle = super::ecstore_bucket::remote_s3_client::PathStyle; pub(crate) type RemoteCredentials = super::ecstore_bucket::remote_s3_client::RemoteCredentials; pub(crate) type RemoteS3ClientError = super::ecstore_bucket::remote_s3_client::RemoteS3ClientError; + pub(crate) type RemoteS3RetryPolicy = super::ecstore_bucket::remote_s3_client::RemoteS3RetryPolicy; } pub(crate) mod metadata_sys { diff --git a/rustfs/src/app/object/get.rs b/rustfs/src/app/object/get.rs index 490394217..b9158afda 100644 --- a/rustfs/src/app/object/get.rs +++ b/rustfs/src/app/object/get.rs @@ -4706,12 +4706,13 @@ async fn odm_get_inline( // The inline path has no background pump, so `source_timeout.idle_ms` is // applied to the teed body here; without it a stalled source would hold // both the client stream and the write-back open until the SDK read - // timeout fires. + // timeout fires. The guard wraps the source read, upstream of the tee, so + // a slow client throttles the tee instead of ageing the source's budget. let source_body: SourceBody = Box::pin(tokio_util::io::ReaderStream::with_capacity( body.into_async_read(), ODM_SOURCE_BODY_CHUNK_BYTES, )); - let guarded = idle_guarded_body(source_body, Duration::from_millis(policy.source_timeout.idle_ms)); + let (guarded, idle) = idle_guarded_body(source_body, Duration::from_millis(policy.source_timeout.idle_ms)); let (primary, secondary) = tee_reader_with_options(Box::pin(tokio_util::io::StreamReader::new(guarded)), ODM_INLINE_TEE_BUFFER_BYTES, options); let output = Box::new(odm_get_output(&head, content_length, content_range, odm_inline_client_body(primary))); @@ -4719,7 +4720,7 @@ async fn odm_get_inline( let commit_key = key.to_string(); spawn_background_with_context(request_context, async move { let body: WriteBackBody = Box::pin(secondary.into_stream()); - let result = commit_inline(&commit_state, &commit_key, head, tags, body).await; + let result = commit_inline(&commit_state, &commit_key, head, tags, body, &idle).await; leader.complete(result.map(|outcome| PullOutcome { etag: outcome.etag, size: outcome.size,