diff --git a/crates/ecstore/src/bucket/on_demand_migration/breaker.rs b/crates/ecstore/src/bucket/on_demand_migration/breaker.rs index c41305f1d..46d25fca2 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/breaker.rs +++ b/crates/ecstore/src/bucket/on_demand_migration/breaker.rs @@ -86,7 +86,12 @@ impl BreakerVerdict { Some(SourceError::Throttled | SourceError::Timeout | SourceError::Connect(_) | SourceError::ServerError(_)) => { BreakerVerdict::Failure } - Some(SourceError::AccessDenied | SourceError::Unsupported(_) | SourceError::Other(_)) => BreakerVerdict::Neutral, + Some( + SourceError::AccessDenied + | SourceError::Unsupported(_) + | SourceError::InvalidPagination(_) + | SourceError::Other(_), + ) => BreakerVerdict::Neutral, } } } diff --git a/crates/ecstore/src/bucket/on_demand_migration/list_through.rs b/crates/ecstore/src/bucket/on_demand_migration/list_through.rs index 8ff71196a..2720c7718 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/list_through.rs +++ b/crates/ecstore/src/bucket/on_demand_migration/list_through.rs @@ -189,8 +189,8 @@ pub enum SourceListPlan { /// delimiter — the source's own roll-up boundary matches the request's. Page { prefix: String }, /// `filter.prefix` reaches past a delimiter, so every key the source could - /// contribute rolls into this one common prefix. One bounded probe listing - /// decides whether it exists; there is nothing to paginate. + /// contribute rolls into this one common prefix. Bounded probes follow + /// empty progressing pages until a key proves existence or the source ends. Folded { probe_prefix: String, common_prefix: String }, } @@ -279,6 +279,29 @@ pub struct FetchRequest { pub token: Option, } +/// Invalid pagination metadata. Opaque cursor values are never included in errors. +#[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)] +pub enum ListPageError { + #[error("truncated listing has no continuation token")] + Missing, + #[error("truncated listing has an empty continuation token")] + Empty, + #[error("truncated listing repeats a continuation token")] + Repeated, +} + +pub(crate) fn validate_list_page(is_truncated: bool, token: Option<&str>, next_token: Option<&str>) -> Result<(), ListPageError> { + if is_truncated { + match next_token { + None => return Err(ListPageError::Missing), + Some("") => return Err(ListPageError::Empty), + Some(next) if Some(next) == token => return Err(ListPageError::Repeated), + Some(_) => {} + } + } + Ok(()) +} + #[derive(Debug, Default)] struct SideState { start: SideCursor, @@ -364,6 +387,11 @@ impl ListThroughMerger { /// or `filter.prefix` excludes it. pub fn disable_source(&mut self) { self.source.disabled = true; + // A refill can fail after a valid first page. A local-only response + // must discard both that source payload and its ordering horizon. + self.source.entries.clear(); + self.source.pages.clear(); + self.source.more = false; } pub fn next_fetch(&self) -> Option { @@ -378,7 +406,13 @@ impl ListThroughMerger { /// Records one fetched page. `entries` must be sorted by `name` and already /// filtered with [`Self::accepts`]; the caller keeps the matching payloads /// in the same order. - pub fn push_page(&mut self, side: MergeSide, entries: Vec, is_truncated: bool, next_token: Option) { + pub fn push_page( + &mut self, + side: MergeSide, + entries: Vec, + is_truncated: bool, + next_token: Option, + ) -> Result<(), ListPageError> { let state = match side { MergeSide::Local => &mut self.local, MergeSide::Source => &mut self.source, @@ -387,15 +421,19 @@ impl ListThroughMerger { Some(last) => last.next_token.clone(), None => state.start.token.clone(), }; - // A truncated page without a cursor cannot be continued; treating the - // side as finished is the only alternative to looping on it forever. - state.more = is_truncated && next_token.is_some(); + validate_list_page(is_truncated, token.as_deref(), next_token.as_deref())?; + // Also reject a cycle through an earlier page in this bounded fetch. + if is_truncated && state.pages.iter().any(|page| page.token == next_token) { + return Err(ListPageError::Repeated); + } + state.more = is_truncated; state.pages.push(FetchedPage { token, count: entries.len(), next_token: is_truncated.then_some(next_token).flatten(), }); state.entries.extend(entries); + Ok(()) } pub fn finish(self) -> MergeOutcome { @@ -599,9 +637,15 @@ mod tests { let (entries, truncated, next) = reference_page(keys, prefix, delimiter, fetch.token.as_deref(), max_keys); let kept: Vec = entries.into_iter().filter(|entry| merger.accepts(&entry.name)).collect(); buffers[usize::from(fetch.side == MergeSide::Source)].extend(kept.iter().cloned()); - merger.push_page(fetch.side, kept, truncated, next); + merger + .push_page(fetch.side, kept, truncated, next) + .expect("reference provider pages must advance"); } let outcome = merger.finish(); + assert_eq!(outcome.is_truncated, outcome.next_token.is_some()); + if outcome.is_truncated { + assert_ne!(outcome.next_token, token, "every truncated merged page must make progress"); + } page_sizes.push(outcome.picks.len()); for pick in &outcome.picks { let entry = buffers[usize::from(pick.side == MergeSide::Source)][pick.index].clone(); @@ -616,11 +660,25 @@ mod tests { } fn expected(local: &[String], source: &[String], prefix: &str, delimiter: Option<&str>) -> Vec { - let mut all: Vec = local.iter().chain(source.iter()).cloned().collect(); - all.sort(); - all.dedup(); - let (entries, _, _) = reference_page(&all, prefix, delimiter, None, usize::MAX); - entries + // This oracle builds the complete namespace independently of the + // provider's page/marker helper and the production merger. + let mut namespace = std::collections::BTreeMap::new(); + for key in local.iter().chain(source) { + let Some(suffix) = key.strip_prefix(prefix) else { + continue; + }; + if let Some(delimiter) = delimiter.filter(|delimiter| !delimiter.is_empty()) + && let Some((directory, _)) = suffix.split_once(delimiter) + { + namespace.insert(format!("{prefix}{directory}{delimiter}"), true); + continue; + } + namespace.insert(key.clone(), false); + } + namespace + .into_iter() + .map(|(name, is_prefix)| ListEntryKey { name, is_prefix }) + .collect() } #[test] @@ -662,7 +720,9 @@ mod tests { token: None }) ); - merger.push_page(MergeSide::Local, vec![ListEntryKey::object("a")], false, None); + merger + .push_page(MergeSide::Local, vec![ListEntryKey::object("a")], false, None) + .expect("local EOF is valid"); assert_eq!(merger.next_fetch(), None); let outcome = merger.finish(); assert_eq!(outcome.picks.len(), 1); @@ -683,12 +743,14 @@ mod tests { }; let mut merger = ListThroughMerger::new(1, Some(&resume)); merger.disable_source(); - merger.push_page( - MergeSide::Local, - vec![ListEntryKey::object("b"), ListEntryKey::object("c")], - true, - Some("local-2".to_string()), - ); + merger + .push_page( + MergeSide::Local, + vec![ListEntryKey::object("b"), ListEntryKey::object("c")], + true, + Some("local-2".to_string()), + ) + .expect("local cursor advances"); let outcome = merger.finish(); assert!(outcome.is_truncated); let token = outcome.next_token.expect("truncated page carries a token"); @@ -698,6 +760,212 @@ mod tests { assert_eq!(token.local.as_deref(), Some("local-1"), "a partly read page is re-listed"); } + #[test] + fn truncated_pages_require_a_nonempty_advancing_cursor() { + for side in [MergeSide::Local, MergeSide::Source] { + for entries in [vec![], vec![ListEntryKey::object("a")]] { + for (next, expected) in [ + (None, Err(ListPageError::Missing)), + (Some(""), Err(ListPageError::Empty)), + (Some("stuck"), Err(ListPageError::Repeated)), + (Some("advances"), Ok(())), + ] { + let resume = ListThroughToken::new( + SideCursor { + token: Some("stuck".into()), + done: false, + }, + SideCursor { + token: Some("stuck".into()), + done: false, + }, + None, + ); + let mut merger = ListThroughMerger::new(2, Some(&resume)); + let result = merger.push_page(side, entries.clone(), true, next.map(str::to_string)); + assert_eq!(result, expected, "{side:?}, {entries:?}, {next:?}"); + let state = if side == MergeSide::Local { + &merger.local + } else { + &merger.source + }; + assert_eq!(state.pages.len(), usize::from(result.is_ok()), "invalid page must not be accepted"); + } + } + } + } + + #[test] + fn repeated_empty_cursor_is_rejected_before_an_identical_page_can_escape() { + let resume = ListThroughToken::new( + SideCursor { token: None, done: true }, + SideCursor { + token: Some("stuck".into()), + done: false, + }, + None, + ); + let mut merger = ListThroughMerger::new(2, Some(&resume)); + assert_eq!( + merger.next_fetch(), + Some(FetchRequest { + side: MergeSide::Source, + token: Some("stuck".into()) + }) + ); + assert_eq!( + merger.push_page(MergeSide::Source, vec![], true, Some("stuck".into())), + Err(ListPageError::Repeated) + ); + } + + #[test] + fn empty_pages_may_advance_within_the_fetch_budget_until_eof() { + let mut merger = ListThroughMerger::new(2, None); + merger.push_page(MergeSide::Local, vec![], false, None).expect("local EOF"); + for next in ["opaque-z", "opaque-a"] { + assert_eq!(merger.next_fetch().expect("bounded source fetch").side, MergeSide::Source); + merger + .push_page(MergeSide::Source, vec![], true, Some(next.into())) + .expect("opaque cursor advances regardless of sort order"); + } + assert!(merger.next_fetch().is_none(), "two source fetches exhaust the request budget"); + let outcome = merger.finish(); + assert!(outcome.picks.is_empty()); + assert!(outcome.is_truncated); + let token = outcome.next_token.expect("empty progressing page has a cursor"); + assert_eq!(token.source.as_deref(), Some("opaque-a")); + let mut merger = ListThroughMerger::new(2, Some(&token)); + assert_eq!(merger.next_fetch().expect("source resumes").token.as_deref(), Some("opaque-a")); + merger + .push_page(MergeSide::Source, vec![ListEntryKey::object("result")], false, None) + .expect("source EOF"); + let outcome = merger.finish(); + assert_eq!( + outcome.picks, + vec![MergePick { + side: MergeSide::Source, + index: 0 + }] + ); + assert!(!outcome.is_truncated); + assert!(outcome.next_token.is_none()); + } + + #[test] + fn a_cursor_cycle_inside_the_fetch_budget_is_rejected() { + let resume = ListThroughToken::new( + SideCursor { token: None, done: true }, + SideCursor { + token: Some("first".into()), + done: false, + }, + None, + ); + let mut merger = ListThroughMerger::new(2, Some(&resume)); + merger + .push_page(MergeSide::Source, vec![], true, Some("second".into())) + .expect("first page advances"); + assert_eq!( + merger.push_page(MergeSide::Source, vec![], true, Some("first".into())), + Err(ListPageError::Repeated) + ); + } + + #[test] + fn source_refill_failure_discards_buffered_source_entries_and_horizon() { + let mut merger = ListThroughMerger::new(2, None); + merger + .push_page(MergeSide::Local, vec![ListEntryKey::object("z")], false, None) + .expect("local EOF"); + merger + .push_page(MergeSide::Source, vec![ListEntryKey::object("a")], true, Some("stuck".into())) + .expect("first source page advances"); + assert_eq!(merger.next_fetch().expect("source refill is required").token.as_deref(), Some("stuck")); + assert_eq!( + merger.push_page(MergeSide::Source, vec![], true, Some("stuck".into())), + Err(ListPageError::Repeated) + ); + merger.disable_source(); + let outcome = merger.finish(); + assert_eq!( + outcome.picks, + vec![MergePick { + side: MergeSide::Local, + index: 0 + }] + ); + assert!(!outcome.is_truncated); + assert!(outcome.next_token.is_none()); + } + + #[test] + fn list_through_static_namespace_boundary_matrix() { + let corpus = [ + "a", + "a/", + "a/b", + "a/b/child", + "a0", + "b", + "b/leaf", + "quote\"&<", + "space key", + "z", + "é", + "中/文", + ]; + for count in [0, 1, 3, 4, corpus.len()] { + let keys: Vec = corpus[..count].iter().map(|key| (*key).to_string()).collect(); + for placement in 0..3 { + let (local, source): (Vec<_>, Vec<_>) = + keys.iter() + .enumerate() + .fold((vec![], vec![]), |(mut local, mut source), (index, key)| { + if placement != 1 || index % 2 == 0 { + local.push(key.clone()); + } + if placement != 0 || index % 2 == 0 { + source.push(key.clone()); + } + (local, source) + }); + for prefix in ["", "a", "a/", "中/"] { + for delimiter in [None, Some("/")] { + for max_keys in [1, 3, 4] { + let oracle = expected(&local, &source, prefix, delimiter); + let (emitted, sizes) = walk(&local, &source, prefix, delimiter, max_keys); + assert_eq!( + emitted.iter().map(|(entry, _)| entry.clone()).collect::>(), + oracle, + "count={count}, placement={placement}, prefix={prefix}, delimiter={delimiter:?}, max={max_keys}" + ); + let expected_sizes: Vec<_> = if oracle.is_empty() { + vec![0] + } else { + oracle.chunks(max_keys).map(<[ListEntryKey]>::len).collect() + }; + assert_eq!(sizes, expected_sizes, "exact max and max+1 boundaries must agree"); + } + } + } + } + } + } + + #[test] + fn list_through_large_overlap_walk_keeps_all_5300_keys() { + let source: Vec<_> = (0..5000).map(|index| format!("k{index:05}")).collect(); + let local: Vec<_> = (4800..5300).map(|index| format!("k{index:05}")).collect(); + let (emitted, sizes) = walk(&local, &source, "", None, 333); + assert_eq!(emitted.len(), 5300); + for (index, (entry, side)) in emitted.iter().enumerate() { + assert_eq!(entry.name, format!("k{index:05}")); + assert_eq!(*side, if index >= 4800 { MergeSide::Local } else { MergeSide::Source }); + } + assert_eq!(sizes, [vec![333; 15], vec![305]].concat()); + } + #[test] fn token_round_trips_and_rejects_tampering() { let token = ListThroughToken::new( @@ -796,7 +1064,10 @@ mod tests { } proptest! { - #![proptest_config(ProptestConfig::with_cases(256))] + #![proptest_config(ProptestConfig { + rng_seed: proptest::test_runner::RngSeed::Fixed(0xec5706), + ..ProptestConfig::with_cases(256) + })] /// Full pagination of a merged listing equals the sorted, deduplicated /// union of both sides, with every shared key served by local, and no 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 4782e05f0..f06301036 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/source_client.rs +++ b/crates/ecstore/src/bucket/on_demand_migration/source_client.rs @@ -25,6 +25,7 @@ //! Client-supplied `If-*`, `Authorization`, `Host` and SSE-C headers are never //! forwarded: v1 rejects SSE-C source objects outright. +use super::list_through::{ListPageError, validate_list_page}; use crate::bucket::remote_s3_client::{ PathStyle, RemoteCredentials, RemoteS3ClientError, RemoteS3EndpointSpec, RemoteS3RetryPolicy, build_remote_s3_config, }; @@ -223,6 +224,8 @@ pub enum SourceError { ServerError(u16), #[error("unsupported source object: {0}")] Unsupported(String), + #[error("invalid source listing: {0}")] + InvalidPagination(#[from] ListPageError), #[error("source request failed: {0}")] Other(String), } @@ -245,6 +248,7 @@ impl SourceError { SourceError::Connect(_) => "connect", SourceError::ServerError(_) => "server_error", SourceError::Unsupported(_) => "unsupported", + SourceError::InvalidPagination(_) => "invalid_pagination", SourceError::Other(_) => "other", } } @@ -714,6 +718,7 @@ impl SourceClient { ..*request }) .await?; + validate_list_page(page.is_truncated, request.continuation_token, page.next_continuation_token.as_deref())?; page.objects = page .objects .into_iter() @@ -800,11 +805,6 @@ impl SourceBackend for S3SourceBackend { let is_truncated = output.is_truncated.unwrap_or(false); let next_continuation_token = output.next_continuation_token; - if is_truncated && next_continuation_token.is_none() { - return Err(SourceError::Other( - "source reported a truncated listing without a continuation token".to_string(), - )); - } let objects = output .contents .unwrap_or_default() @@ -1274,7 +1274,9 @@ mod tests { data/photos/ outside/ "#; - let (client, requests) = scripted_client(&spec(Some("data/")), vec![ok(Vec::new(), body), ok(Vec::new(), body)]).await; + let next_body = body.replace("data/opaque", "data/next"); + let (client, requests) = + scripted_client(&spec(Some("data/")), vec![ok(Vec::new(), body), ok(Vec::new(), &next_body)]).await; let first = client .list_page(&SourceListRequest { prefix: Some("photos/"), @@ -1336,7 +1338,104 @@ mod tests { .list_objects_v2(None, None, 10) .await .expect_err("truncated page without token is corrupt"); - assert!(matches!(err, SourceError::Other(_)), "{err:?}"); + assert!(matches!(err, SourceError::InvalidPagination(ListPageError::Missing)), "{err:?}"); + } + + #[tokio::test] + async fn list_page_validates_s3_cursor_progress_before_mapping_entries() { + for contents in ["", "data/a1"] { + for (truncated, next, expected) in [ + (true, None, Some(ListPageError::Missing)), + (true, Some(""), Some(ListPageError::Empty)), + (true, Some("stuck"), Some(ListPageError::Repeated)), + (true, Some("opaque-next"), None), + (false, None, None), + (false, Some("stuck"), None), + ] { + let next_xml = next + .map(|next| format!("{next}")) + .unwrap_or_default(); + let body = format!( + "{truncated}{next_xml}{contents}" + ); + let (client, requests) = scripted_client(&spec(Some("data/")), vec![ok(Vec::new(), &body)]).await; + let result = client + .list_page(&SourceListRequest { + continuation_token: Some("stuck"), + max_keys: 2, + ..Default::default() + }) + .await; + match expected { + Some(expected) => { + let error = result.expect_err("malformed pagination must fail at the provider boundary"); + assert!( + matches!(&error, SourceError::InvalidPagination(actual) if *actual == expected), + "{error:?}" + ); + assert_eq!(error.class_label(), "invalid_pagination"); + assert!(!error.is_retryable()); + assert!(!error.to_string().contains("stuck"), "errors must not echo opaque tokens"); + } + None => { + let page = result.expect("progressing empty/nonempty pages and EOF are valid"); + assert_eq!(page.is_truncated, truncated); + assert_eq!(page.next_continuation_token.as_deref(), next); + assert_eq!(page.objects.len(), usize::from(!contents.is_empty())); + if let Some(object) = page.objects.first() { + assert_eq!(object.key, "a"); + } + } + } + let requests = recorded(&requests); + assert_eq!(requests.len(), 1, "invalid pagination must not be retried"); + assert!(requests[0].uri.contains("continuation-token=stuck")); + } + } + } + + struct ListOnlyBackend(SourcePage); + + #[async_trait::async_trait] + impl SourceBackend for ListOnlyBackend { + async fn list(&self, request: &SourceListRequest<'_>) -> Result { + assert_eq!(request.continuation_token, Some("stuck"), "opaque cursors reach every provider unchanged"); + Ok(self.0.clone()) + } + + async fn head(&self, _key: &str) -> Result { + panic!("unexpected HEAD in list test") + } + async fn get(&self, _key: &str, _range: Option<&HTTPRangeSpec>) -> Result { + panic!("unexpected GET in list test") + } + async fn tagging(&self, _key: &str) -> Result, SourceError> { + panic!("unexpected tagging in list test") + } + async fn probe(&self) -> Result<(), SourceError> { + panic!("unexpected probe in list test") + } + } + + #[tokio::test] + async fn list_page_validates_non_s3_provider_cursors_at_the_common_boundary() { + for (next, expected) in [ + (None, ListPageError::Missing), + (Some(""), ListPageError::Empty), + (Some("stuck"), ListPageError::Repeated), + ] { + let mut client = prefix_client(Some("data/".into())); + client.backend = Box::new(ListOnlyBackend(SourcePage { + is_truncated: true, + next_continuation_token: next.map(str::to_string), + ..Default::default() + })); + let error = client + .list_objects_v2(None, Some("stuck"), 2) + .await + .expect_err("all providers must advance pagination"); + assert!(matches!(error, SourceError::InvalidPagination(actual) if actual == expected)); + } } const TAGGING_BODY: &str = r#" diff --git a/crates/ecstore/src/bucket/on_demand_migration/stats.rs b/crates/ecstore/src/bucket/on_demand_migration/stats.rs index ed3a7de2c..a483abde0 100644 --- a/crates/ecstore/src/bucket/on_demand_migration/stats.rs +++ b/crates/ecstore/src/bucket/on_demand_migration/stats.rs @@ -177,7 +177,7 @@ impl From<&SourceError> for PullFailureReason { SourceError::Connect(_) => PullFailureReason::SourceConnect, SourceError::ServerError(_) => PullFailureReason::SourceServerError, SourceError::Unsupported(_) => PullFailureReason::SourceUnsupported, - SourceError::Other(_) => PullFailureReason::SourceOther, + SourceError::InvalidPagination(_) | SourceError::Other(_) => PullFailureReason::SourceOther, } } } diff --git a/rustfs/src/app/bucket_list_through.rs b/rustfs/src/app/bucket_list_through.rs index 02c4c7420..a97470f9a 100644 --- a/rustfs/src/app/bucket_list_through.rs +++ b/rustfs/src/app/bucket_list_through.rs @@ -252,8 +252,16 @@ pub(crate) async fn merged_list_objects_v2( .filter(|entry| merger.accepts(&entry.key().name)) .collect(); let keys: Vec = kept.iter().map(SideEntry::key).collect(); + if let Err(error) = merger.push_page(fetch.side, keys, is_truncated, next_token) { + match fetch.side { + MergeSide::Source => { + degrade_or_fail(&mut merger, &mut degraded, policy.source_error, "invalid_pagination")?; + continue; + } + MergeSide::Local => return Err(S3Error::with_message(S3ErrorCode::InternalError, error.to_string())), + } + } buffers[usize::from(fetch.side == MergeSide::Source)].extend(kept.into_iter().map(Some)); - merger.push_page(fetch.side, keys, is_truncated, next_token); } let outcome = merger.finish(); @@ -340,10 +348,12 @@ async fn fetch_source_page( continuation_token: token, max_keys: params.max_keys, }, - // Everything under `filter.prefix` rolls into one common prefix, so a - // single bounded listing settles whether it exists. + // Everything under `filter.prefix` rolls into one common prefix. An + // empty truncated probe must still follow its cursor before declaring + // that prefix absent. SourceListPlan::Folded { probe_prefix, .. } => SourceListRequest { prefix: Some(probe_prefix.as_str()), + continuation_token: token, max_keys: 1, ..Default::default() }, @@ -368,8 +378,8 @@ async fn fetch_source_page( } else { Vec::new() }, - false, - None, + !exists && page.is_truncated, + if exists { None } else { page.next_continuation_token }, )) } _ => { @@ -417,6 +427,17 @@ async fn local_delete_markers(store: &Arc, bucket: &str, keys: &[String #[cfg(test)] mod tests { use super::*; + use crate::app::bucket_usecase::DefaultBucketUsecase; + use crate::app::gating_test_env::{run_large_stack_test, shared_gating_ecstore}; + use crate::app::storage_api::bucket_usecase::bucket::on_demand_migration::{ + FilterConfig, OnDemandMigrationConfig, PathStyle, PolicyConfig, Provider, SourceConfig, SourceCredentials, TlsConfig, + }; + use crate::app::storage_api::bucket_usecase::s3::{ListObjectsV2Input, ListObjectsV2Output, S3Request, S3Response}; + use crate::app::storage_api::test::StoragePutObjReader; + use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions}; + use crate::app::storage_api::test::contract::object::ObjectIO as _; + use std::time::Duration; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; fn token(local: Option<&str>, local_done: bool) -> ListThroughToken { ListThroughToken { @@ -526,4 +547,332 @@ mod tests { assert!(degraded); assert_eq!(merger.next_fetch().map(|fetch| fetch.side), Some(MergeSide::Local)); } + + /// Serves exactly the scripted S3 pages and joins every connection before + /// returning. A source retry or unexpected operation fails the test. + async fn scripted_list_source(pages: Vec) -> (String, tokio_util::task::AbortOnDropHandle>) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind listing source"); + let address = listener.local_addr().expect("listing source address"); + let server = tokio::spawn(async move { + let mut requests = Vec::new(); + for body in pages { + let (mut stream, _) = listener.accept().await.expect("accept source listing"); + let mut request = Vec::new(); + let mut chunk = [0; 4096]; + while !request.windows(4).any(|window| window == b"\r\n\r\n") { + let count = stream.read(&mut chunk).await.expect("read signed listing request"); + assert!(count > 0, "source request must include complete headers"); + request.extend_from_slice(&chunk[..count]); + assert!(request.len() <= 32 * 1024, "listing request headers must be bounded"); + } + let first_line = String::from_utf8_lossy(&request) + .lines() + .next() + .expect("request line") + .to_string(); + // The SDK joins the bucket endpoint with the LIST operation's `/` path. + assert!( + first_line.starts_with("GET /source-bucket/?"), + "expected a path-style bucket-root LIST request, got {first_line:?}" + ); + assert!(first_line.contains("list-type=2"), "expected a ListObjectsV2 query, got {first_line:?}"); + requests.push(first_line); + let response = format!( + "HTTP/1.1 200 OK\r\ncontent-type: application/xml\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", + body.len() + ); + stream.write_all(response.as_bytes()).await.expect("write source page"); + stream.shutdown().await.expect("finish source response"); + } + requests + }); + (format!("http://{address}"), tokio_util::task::AbortOnDropHandle::new(server)) + } + + fn source_xml(next: Option<&str>, truncated: bool, key: Option<&str>) -> String { + let next = next + .map(|token| format!("{token}")) + .unwrap_or_default(); + let contents = key + .map(|key| format!("{key}1")) + .unwrap_or_default(); + format!( + "{truncated}{next}{contents}" + ) + } + + struct ListThroughTestState { + bucket: String, + module_enabled: bool, + } + + impl Drop for ListThroughTestState { + fn drop(&mut self) { + let sys = OnDemandMigrationSys::get(); + sys.remove(&self.bucket); + sys.set_module_enabled(self.module_enabled); + } + } + + async fn source_policy_request( + pages: Vec, + policy: SourceErrorPolicy, + resume_source: Option<&str>, + filter_prefix: Option<&str>, + ) -> (S3Result>, Vec) { + let store = shared_gating_ecstore().await; + crate::app::runtime_sources::install_test_app_context(Arc::clone(&store)).await; + let bucket = format!("odm-list-{}", uuid::Uuid::new_v4().simple()); + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("create list-through bucket"); + store + .put_object( + &bucket, + "z-local", + &mut StoragePutObjReader::from_vec(vec![1]), + &StorageObjectOptions::default(), + ) + .await + .expect("seed real local listing"); + let (endpoint, server) = scripted_list_source(pages).await; + let sys = OnDemandMigrationSys::get(); + let _state_guard = ListThroughTestState { + bucket: bucket.clone(), + module_enabled: sys.is_module_enabled(), + }; + sys.set_module_enabled(true); + let config = OnDemandMigrationConfig { + version: 1, + enabled: true, + source: SourceConfig { + provider: Provider::Minio, + endpoint: Some(endpoint), + region: "us-east-1".into(), + bucket: "source-bucket".into(), + path_style: PathStyle::Path, + credentials: Some(SourceCredentials { + access_key: "test-access".into(), + secret_key: "test-secret".into(), + session_token: None, + }), + tls: TlsConfig::default(), + }, + filter: FilterConfig { + prefix: filter_prefix.map(str::to_string), + ..Default::default() + }, + policy: PolicyConfig { + list_through: true, + source_error: policy, + ..Default::default() + }, + }; + sys.apply(&bucket, Some(&config)).await; + assert!( + sys.state(&bucket).expect("ODM state installed").client().is_ok(), + "fake source client must build" + ); + let continuation_token = resume_source.map(|source| { + let token = ListThroughToken { + t: "odm-list".into(), + v: 1, + local: None, + local_done: false, + source: Some(source.into()), + source_done: false, + last_key: None, + }; + base64_simd::STANDARD.encode_to_string(token.encode().as_bytes()) + }); + let input = ListObjectsV2Input { + bucket, + max_keys: Some(2), + continuation_token, + delimiter: filter_prefix.map(|_| "/".to_string()), + encoding_type: None, + expected_bucket_owner: None, + fetch_owner: None, + optional_object_attributes: None, + prefix: None, + request_payer: None, + start_after: None, + }; + let request = S3Request { + input, + method: http::Method::GET, + uri: http::Uri::from_static("/?list-type=2"), + headers: HeaderMap::new(), + extensions: http::Extensions::new(), + credentials: None, + region: None, + service: None, + trailing_headers: None, + }; + let result = tokio::time::timeout( + Duration::from_secs(10), + DefaultBucketUsecase::from_global().execute_list_objects_v2(request), + ) + .await + .expect("listing must complete within its bounded source budget"); + let requests = tokio::time::timeout(Duration::from_secs(5), server) + .await + .expect("source connections must finish") + .expect("source server must not panic"); + (result, requests) + } + + #[test] + #[serial_test::serial] + fn list_through_invalid_source_pagination_obeys_policy_on_the_handler_path() { + run_large_stack_test("list-through-source-policy", || async { + temp_env::async_with_vars( + [ + ("RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET", Some("true")), + ("HTTP_PROXY", None), + ("HTTPS_PROXY", None), + ("ALL_PROXY", None), + ("http_proxy", None), + ("https_proxy", None), + ("all_proxy", None), + ("NO_PROXY", Some("*")), + ("no_proxy", Some("*")), + ], + async { + for policy in [SourceErrorPolicy::Propagate, SourceErrorPolicy::NotFound] { + for next in [None, Some(""), Some("stuck")] { + for key in [None, Some("a-source")] { + let (result, requests) = + source_policy_request(vec![source_xml(next, true, key)], policy, Some("stuck"), None).await; + assert_eq!(requests.len(), 1, "a malformed source page must not be retried"); + assert!(requests[0].contains("continuation-token=stuck")); + assert_source_policy_result(result, policy); + } + } + let (result, requests) = source_policy_request( + vec![ + source_xml(Some("stuck"), true, Some("a-source")), + source_xml(Some("stuck"), true, None), + ], + policy, + None, + None, + ) + .await; + assert_eq!(requests.len(), 2, "the failure must occur during a real refill"); + assert!(!requests[0].contains("continuation-token=")); + assert!(requests[1].contains("continuation-token=stuck")); + assert_source_policy_result(result, policy); + } + }, + ) + .await; + }); + } + + #[test] + #[serial_test::serial] + fn list_through_empty_advancing_source_pages_reach_eof_on_the_handler_path() { + run_large_stack_test("list-through-empty-source-pages", || async { + temp_env::async_with_vars( + [ + ("RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET", Some("true")), + ("HTTP_PROXY", None), + ("HTTPS_PROXY", None), + ("ALL_PROXY", None), + ("http_proxy", None), + ("https_proxy", None), + ("all_proxy", None), + ("NO_PROXY", Some("*")), + ("no_proxy", Some("*")), + ], + async { + for filter_prefix in [None, Some("photos/2024/")] { + let source_key = if filter_prefix.is_some() { + "photos/2024/a-source" + } else { + "a-source" + }; + let (result, requests) = source_policy_request( + vec![ + source_xml(Some("opaque-next"), true, None), + source_xml(None, false, Some(source_key)), + ], + SourceErrorPolicy::Propagate, + None, + filter_prefix, + ) + .await; + assert_eq!(requests.len(), 2, "an empty truncated source page must reach its successor"); + assert!(requests[1].contains("continuation-token=opaque-next")); + let response = result.expect("empty progressing source page is valid"); + assert!(!response.headers.contains_key("x-rustfs-on-demand-migration-list")); + let output = response.output; + let objects: Vec<_> = output + .contents + .unwrap_or_default() + .into_iter() + .map(|object| object.key.expect("listed object key")) + .collect(); + if filter_prefix.is_some() { + assert_eq!(objects, vec!["z-local"]); + assert_eq!( + output + .common_prefixes + .unwrap_or_default() + .into_iter() + .map(|prefix| prefix.prefix.expect("rolled-up prefix")) + .collect::>(), + vec!["photos/"] + ); + } else { + assert_eq!(objects, vec!["a-source", "z-local"]); + assert!(output.common_prefixes.unwrap_or_default().is_empty()); + } + assert_eq!(output.key_count, Some(2)); + assert_eq!(output.is_truncated, Some(false)); + assert!(output.next_continuation_token.is_none()); + } + }, + ) + .await; + }); + } + + fn assert_source_policy_result(result: S3Result>, policy: SourceErrorPolicy) { + match policy { + SourceErrorPolicy::Propagate => { + let error = result.expect_err("propagate must expose malformed pagination"); + assert_eq!(error.status_code(), Some(http::StatusCode::FAILED_DEPENDENCY)); + assert_eq!(error.code(), &S3ErrorCode::Custom("SourceUnavailable".into())); + assert_eq!(error.message(), Some("invalid_pagination")); + } + SourceErrorPolicy::NotFound => { + let response = result.expect("not_found must preserve the local listing"); + assert_eq!( + response + .headers + .get("x-rustfs-on-demand-migration-list") + .expect("local_only header"), + "local_only" + ); + let output = response.output; + assert_eq!( + output + .contents + .unwrap_or_default() + .into_iter() + .map(|object| object.key.expect("local key")) + .collect::>(), + vec!["z-local"] + ); + assert_eq!(output.is_truncated, Some(false)); + assert_eq!(output.key_count, Some(1)); + assert!(output.next_continuation_token.is_none()); + } + } + } } diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index 19548aa66..9f543ac95 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -29,11 +29,13 @@ pub(crate) fn EndpointServerPools( pub(crate) mod s3 { #[cfg(test)] pub(crate) use s3s::dto::{ - BucketVersioningStatus, DeleteMarkerReplication, DeleteMarkerReplicationStatus, Destination, ReplicationConfiguration, - ReplicationRule, ReplicationRuleFilter, ReplicationRuleStatus, ServerSideEncryptionByDefault, - ServerSideEncryptionConfiguration, ServerSideEncryptionRule, Tag, VersioningConfiguration, + BucketVersioningStatus, DeleteMarkerReplication, DeleteMarkerReplicationStatus, Destination, ListObjectsV2Input, + ListObjectsV2Output, ReplicationConfiguration, ReplicationRule, ReplicationRuleFilter, ReplicationRuleStatus, + ServerSideEncryptionByDefault, ServerSideEncryptionConfiguration, ServerSideEncryptionRule, Tag, VersioningConfiguration, }; pub(crate) use s3s::{S3Error, S3ErrorCode, S3Result}; + #[cfg(test)] + pub(crate) use s3s::{S3Request, S3Response}; } pub(crate) mod admin {