diff --git a/crates/ecstore/src/store/list_objects.rs b/crates/ecstore/src/store/list_objects.rs index 038e436e0..91f2aca5d 100644 --- a/crates/ecstore/src/store/list_objects.rs +++ b/crates/ecstore/src/store/list_objects.rs @@ -2194,6 +2194,41 @@ fn last_scanned_entry_name(entries: Option<&MetaCacheEntriesSorted>) -> Option, delimiter: Option<&str>, marker: Option<&str>) { + if delimiter.is_some() + && let Some(marker) = marker + { + // S3 excludes CommonPrefixes that are not lexicographically greater than KeyMarker. + // Filter before applying max-keys so a resumed prefix cannot consume the page budget. + objects.retain(|object| !(object.is_dir && object.mod_time.is_none()) || object.name.as_str() > marker); + } +} + +fn last_raw_key_for_prefix<'a>(raw_keys: impl Iterator, prefix: &str) -> Option<&'a str> { + raw_keys.filter(|key| key.starts_with(prefix)).last() +} + +fn versioned_prefix_page_boundary_key<'a>( + objects: &[ObjectInfo], + raw_keys: impl Iterator, + delimiter: Option<&str>, + max_keys: i32, + disk_has_more: bool, +) -> Option { + if delimiter.is_none() || !disk_has_more || max_keys <= 0 || objects.len() < max_keys as usize { + return None; + } + + let last_visible = objects.get(max_keys as usize - 1)?; + if !last_visible.is_dir || last_visible.mod_time.is_some() { + return None; + } + + // A folded prefix cannot advance the raw versioned walker. Continue at the last + // raw key represented by the final returned prefix, not at the next prefix's key. + last_raw_key_for_prefix(raw_keys, &last_visible.name).map(str::to_owned) +} + fn build_list_next_marker(objects: &[ObjectInfo], prefixes: &[String], cache_id: Option<&str>) -> Option { if let Some(last) = objects.last() { Some(append_list_cache_id_to_marker(last.name.clone(), cache_id)) @@ -2280,14 +2315,20 @@ fn list_objects_paginate( } } - if !is_truncated && disk_has_more && !include_version_id { + if !is_truncated && disk_has_more { let visible_count = objects.len() + prefixes.len(); let should_truncate = if delimiter.is_none() { visible_count > 0 } else { visible_count >= max_keys as usize }; - if should_truncate { + if delimiter.is_some() + && include_version_id + && let Some(last_scanned) = last_scanned_key + { + is_truncated = true; + next_marker = Some(append_list_cache_id_to_marker(last_scanned.to_owned(), cache_id)); + } else if should_truncate { is_truncated = true; if include_version_id { (next_marker, next_version_idmarker) = build_list_versions_next_marker(&objects, &prefixes, cache_id); @@ -2320,6 +2361,30 @@ fn list_objects_paginate( (objects, prefixes, is_truncated, next_marker, next_version_idmarker) } +fn list_objects_paginate_versions<'a>( + mut get_objects: Vec, + delimiter: &Option, + max_keys: i32, + disk_has_more: bool, + cache_id: Option<&str>, + marker: Option<&str>, + last_scanned_key: Option<&str>, + raw_keys: impl Iterator, +) -> (Vec, Vec, bool, Option, Option) { + filter_versions_common_prefixes_after_marker(&mut get_objects, delimiter.as_deref(), marker); + let prefix_at_page_boundary = + versioned_prefix_page_boundary_key(&get_objects, raw_keys, delimiter.as_deref(), max_keys, disk_has_more); + + let (objects, prefixes, is_truncated, mut next_marker, mut next_version_idmarker) = + list_objects_paginate(get_objects, delimiter, max_keys, disk_has_more, cache_id, true, last_scanned_key); + if is_truncated && let Some(raw_key) = prefix_at_page_boundary { + next_marker = Some(append_list_cache_id_to_marker(raw_key, cache_id)); + next_version_idmarker = None; + } + + (objects, prefixes, is_truncated, next_marker, next_version_idmarker) +} + enum VerifiedIndexVisibleEntry { Object(Box), Prefix(String), @@ -4259,14 +4324,15 @@ impl ECStore { ObjectInfo::from_meta_cache_entries_sorted_versions(&entries, bucket, prefix, delimiter.clone(), version_marker).await }; - let (objects, prefixes, is_truncated, next_marker, next_version_idmarker) = list_objects_paginate( + let (objects, prefixes, is_truncated, next_marker, next_version_idmarker) = list_objects_paginate_versions( get_objects, &delimiter, max_keys, disk_has_more, next_cache_id.as_deref(), - true, + opts.marker.as_deref(), last_scanned_key.as_deref(), + entries.entries().iter().map(|entry| entry.name.as_str()), ); Ok(ListObjectVersionsInfo { @@ -5681,23 +5747,20 @@ impl Sets { // Last RAW scanned key, captured before folding (ECA-03 / #944). let last_scanned_key = last_scanned_entry_name(list_result.entries.as_ref()); - let get_objects = ObjectInfo::from_meta_cache_entries_sorted_versions( - &list_result.entries.unwrap_or_default(), - bucket, - prefix, - delimiter.clone(), - version_marker, - ) - .await; + let entries = list_result.entries.unwrap_or_default(); + let get_objects = + ObjectInfo::from_meta_cache_entries_sorted_versions(&entries, bucket, prefix, delimiter.clone(), version_marker) + .await; - let (objects, prefixes, is_truncated, next_marker, next_version_idmarker) = list_objects_paginate( + let (objects, prefixes, is_truncated, next_marker, next_version_idmarker) = list_objects_paginate_versions( get_objects, &delimiter, max_keys, disk_has_more, next_cache_id.as_deref(), - true, + opts.marker.as_deref(), last_scanned_key.as_deref(), + entries.entries().iter().map(|entry| entry.name.as_str()), ); Ok(ListObjectVersionsInfo { @@ -6498,23 +6561,20 @@ impl SetDisks { // Last RAW scanned key, captured before folding (ECA-03 / #944). let last_scanned_key = last_scanned_entry_name(list_result.entries.as_ref()); - let get_objects = ObjectInfo::from_meta_cache_entries_sorted_versions( - &list_result.entries.unwrap_or_default(), - bucket, - prefix, - delimiter.clone(), - version_marker, - ) - .await; + let entries = list_result.entries.unwrap_or_default(); + let get_objects = + ObjectInfo::from_meta_cache_entries_sorted_versions(&entries, bucket, prefix, delimiter.clone(), version_marker) + .await; - let (objects, prefixes, is_truncated, next_marker, next_version_idmarker) = list_objects_paginate( + let (objects, prefixes, is_truncated, next_marker, next_version_idmarker) = list_objects_paginate_versions( get_objects, &delimiter, max_keys, disk_has_more, next_cache_id.as_deref(), - true, + opts.marker.as_deref(), last_scanned_key.as_deref(), + entries.entries().iter().map(|entry| entry.name.as_str()), ); Ok(ListObjectVersionsInfo { @@ -8697,6 +8757,219 @@ mod test { assert_eq!(next_marker.as_deref(), Some("obj-0002")); } + // ECA-03 / #944 (versioned variant): a versions listing whose raw keys fully + // collapse into fewer than max_keys common prefixes must still report + // truncation and carry a continuation marker, otherwise every key and every + // version past the scan window is silently dropped. + #[test] + fn list_objects_paginate_versioned_delimiter_refold_reports_truncation() { + let delimiter = Some("-".to_string()); + let get_objects = vec![folded_prefix_object("data-")]; + + let (objects, prefixes, is_truncated, next_marker, next_version_idmarker) = list_objects_paginate( + get_objects, + &delimiter, + 1000, + true, // disk_has_more: walker filled its raw candidate limit + None, + true, + Some("data-1001"), // last RAW scanned key + ); + + assert!(objects.is_empty()); + assert_eq!(prefixes, vec!["data-".to_string()]); + assert!(is_truncated, "versioned re-folded page with more on disk must be truncated"); + // The marker must be the last RAW key, NOT the folded prefix "data-": + // versioned walks are excluded from the gather-level common-prefix + // collector, so a prefix marker could never advance `forward_past`. + assert_eq!(next_marker.as_deref(), Some("data-1001")); + assert!(next_version_idmarker.is_none(), "a folded prefix page carries no version marker"); + } + + // Same contract when the page is FILLED by folded common prefixes + // (`visible_count >= max_keys`): the versioned path must still prefer the + // raw scanned key over `prefixes.last()`, which would loop forever. + #[test] + fn list_objects_paginate_versioned_delimiter_full_prefix_page_uses_raw_marker() { + let delimiter = Some("-".to_string()); + let get_objects = vec![folded_prefix_object("data-")]; + + let (objects, prefixes, is_truncated, next_marker, next_version_idmarker) = + list_objects_paginate(get_objects, &delimiter, 1, true, None, true, Some("data-0002")); + + assert!(objects.is_empty()); + assert_eq!(prefixes, vec!["data-".to_string()]); + assert!(is_truncated, "a prefix-filled versioned page with more on disk must be truncated"); + assert_eq!(next_marker.as_deref(), Some("data-0002")); + assert!(next_version_idmarker.is_none()); + } + + // End-to-end simulation of versioned pagination with `max_keys=1`, where the + // next page resumes via a bare key marker (`forward_past` keeps `name > marker` + // only, matching `inner_list_object_versions_with_projection` without a version + // marker). Before the raw-key marker the same window replayed forever, so + // `other-`/`zzz` were never listed. + #[test] + fn list_objects_paginate_versioned_delimiter_prefix_filled_pages_terminate() { + let all: Vec = [ + "data-0001", + "data-0002", + "data-0003", + "data-0004", + "data-0005", + "other-0001", + "zzz", + ] + .into_iter() + .map(str::to_string) + .collect(); + let delimiter = Some("-".to_string()); + let max_keys = 1i32; + let limit = max_keys_plus_one(max_keys, true) as usize; + + let mut marker: Option = None; + let mut seen_prefixes = Vec::new(); + let mut seen_objects = Vec::new(); + let mut page_prefixes = Vec::new(); + let mut pages = 0; + + loop { + pages += 1; + assert!(pages <= 16, "versioned pagination did not terminate (possible infinite loop)"); + + let start = match &marker { + Some(m) => all.partition_point(|k| k.as_str() <= m.as_str()), + None => 0, + }; + let window: Vec = all[start..].iter().take(limit).cloned().collect(); + let disk_has_more = window.len() == limit; + let last_scanned = window.last().cloned(); + + let get_objects = fold_delimiter_page(&window, "", "-"); + let (objects, prefixes, is_truncated, next_marker, _v) = super::list_objects_paginate_versions( + get_objects, + &delimiter, + max_keys, + disk_has_more, + None, + marker.as_deref(), + last_scanned.as_deref(), + window.iter().map(String::as_str), + ); + + assert!( + prefixes.windows(2).all(|pair| pair[0] != pair[1]), + "a page must not contain duplicate common prefixes" + ); + assert!( + prefixes + .iter() + .all(|prefix| marker.as_deref().is_none_or(|marker| prefix.as_str() > marker)), + "a resumed page must not repeat a prefix at or before its key marker" + ); + page_prefixes.push(prefixes.clone()); + seen_objects.extend(objects.into_iter().map(|object| object.name)); + seen_prefixes.extend(prefixes); + + if !is_truncated { + assert!(next_marker.is_none()); + break; + } + + let next = next_marker.expect("truncated versioned page must carry a continuation marker"); + if let Some(prev) = &marker { + assert!( + next.as_str() > prev.as_str(), + "next_marker must strictly advance to stay finite, got {next} after {prev}" + ); + } + marker = Some(next); + } + + assert_eq!( + page_prefixes, + vec![vec!["data-".to_string()], vec![], vec!["other-".to_string()], vec![]], + "versioned pagination must not repeat a folded prefix while scanning later raw keys" + ); + assert_eq!( + seen_prefixes, + vec!["data-".to_string(), "other-".to_string()], + "versioned pagination must preserve the ordered visible result sequence" + ); + assert_eq!(seen_objects, vec!["zzz".to_string()]); + } + + #[test] + fn list_objects_paginate_versioned_prefix_boundary_uses_last_raw_key_in_returned_prefix() { + let all: Vec = ["alpha-0001", "beta-0001", "charlie-0001"] + .into_iter() + .map(str::to_string) + .collect(); + let delimiter = Some("-".to_string()); + let max_keys = 1; + let limit = max_keys_plus_one(max_keys, true) as usize; + let mut marker: Option = None; + let mut page_prefixes = Vec::new(); + let mut pages = 0; + + loop { + pages += 1; + assert!(pages <= 8, "versioned prefix pagination did not terminate"); + + let start = marker + .as_deref() + .map(|marker| all.partition_point(|key| key.as_str() <= marker)) + .unwrap_or(0); + let window: Vec = all[start..].iter().take(limit).cloned().collect(); + let disk_has_more = window.len() == limit; + let last_scanned = window.last().cloned(); + + let get_objects = fold_delimiter_page(&window, "", "-"); + let (objects, prefixes, is_truncated, next_marker, _) = super::list_objects_paginate_versions( + get_objects, + &delimiter, + max_keys, + disk_has_more, + None, + marker.as_deref(), + last_scanned.as_deref(), + window.iter().map(String::as_str), + ); + + assert_eq!(objects.len() + prefixes.len(), 1, "a page must respect max-keys"); + page_prefixes.push(prefixes); + + if !is_truncated { + assert!(next_marker.is_none()); + break; + } + + let next = next_marker.expect("truncated page must carry a continuation key marker"); + if let Some(previous) = &marker { + assert!(next > *previous, "the continuation marker must strictly advance"); + } + marker = Some(next); + } + + assert_eq!( + page_prefixes, + vec![ + vec!["alpha-".to_string()], + vec!["beta-".to_string()], + vec!["charlie-".to_string()] + ] + ); + } + + #[test] + fn versioned_marker_filter_does_not_change_no_delimiter_directories() { + let mut objects = vec![folded_prefix_object("data-")]; + super::filter_versions_common_prefixes_after_marker(&mut objects, None, Some("z")); + + assert_eq!(objects.len(), 1); + assert!(super::versioned_prefix_page_boundary_key(&objects, std::iter::once("data-"), None, 1, true).is_none()); + } + // End-to-end pagination over 5000 `data-*` + 100 `other-*` with delimiter '-'. // Every full page re-folds to a single common prefix (< max_keys), the exact // trigger for ECA-03 / #944. Asserts pagination terminates in a bounded number