From 65bd2d575b2a5ae92cd606d37d3d315eb31d3973 Mon Sep 17 00:00:00 2001 From: Chris An Date: Mon, 26 Oct 2026 21:03:12 +0800 Subject: [PATCH] refactor(listing): make cursors plain keys and opaque V2 tokens (#8950) * refactor(listing): make cursors plain keys and opaque V2 tokens RustFS inherited MinIO's metacache cursor tag and appended `[rustfs_cache:v2,id:,src:walker,gen:live]` to every internal continuation marker, although nothing ever consumed the id: four parsers had drifted apart, V1 NextMarker had to be stripped on the way out, and ListObjectVersions leaked the tag onto the wire as NextKeyMarker. ecstore stops assigning `list_id` in `ECStore::list_path` and `SetDisks::list_path_result`, drops `append_list_cache_id_to_marker` and the `cache_id` parameters of the pagination helpers, and removes the unused `ListPathOptions::encode_marker`; internal markers are the plain last key, last common prefix or last raw scanned key. The API layer builds the V2 NextContinuationToken as base64 of the server-side cursor in one place and decodes it in one place; V1 NextMarker is the decoded cursor and the `public_list_marker` tag stripper is gone; NextKeyMarker is the plain key. Base64 of the plain key is what nodes from earlier releases already read, so a rolling upgrade needs no switch. Compatibility, registered as `listing-cursor-legacy-tag`: `parse_marker` and `list_marker_key` still strip a trailing legacy tag from incoming markers so a client that paginates across the upgrade keeps its position. * fix(listing): preserve literal cache-tag cursor boundaries --- .../src/list_objects_v2_pagination_test.rs | 14 +- .../bucket/lifecycle/bucket_lifecycle_ops.rs | 8 +- crates/ecstore/src/store/list_objects.rs | 393 +++++++++++------- docs/architecture/compat-cleanup-register.md | 1 + rustfs/src/app/bucket_list_through.rs | 6 +- rustfs/src/app/bucket_usecase.rs | 6 +- rustfs/src/app/storage_api.rs | 4 +- rustfs/src/storage/s3_api/bucket.rs | 237 ++++++----- rustfs/src/storage/storage_api.rs | 4 +- 9 files changed, 395 insertions(+), 278 deletions(-) diff --git a/crates/e2e_test/src/list_objects_v2_pagination_test.rs b/crates/e2e_test/src/list_objects_v2_pagination_test.rs index 9b7c0bace..dbde76f4e 100644 --- a/crates/e2e_test/src/list_objects_v2_pagination_test.rs +++ b/crates/e2e_test/src/list_objects_v2_pagination_test.rs @@ -1129,12 +1129,14 @@ mod tests { /// and the marker plus the cursor tag. /// /// Bug Reference: backlog#1047 - /// The V2 continuation token appends a `[rustfs_cache:...]` cursor tag to the - /// last returned key. The orchestration layer passed that raw string to - /// `forward_past`, so any key whose byte after the shared stem sorts below - /// `[` (0x5B) — `.`, `/`, `-`, digits, uppercase — was silently dropped on - /// the next page: with keys `a`, `a.txt`, `zz` and max_keys=1, page 2 - /// returned `zz` and `a.txt` was never listed. + /// Earlier releases appended a `[rustfs_cache:...]` cursor tag to the last + /// returned key inside the V2 continuation token. The orchestration layer + /// passed that raw string to `forward_past`, so any key whose byte after the + /// shared stem sorts below `[` (0x5B) — `.`, `/`, `-`, digits, uppercase — + /// was silently dropped on the next page: with keys `a`, `a.txt`, `zz` and + /// max_keys=1, page 2 returned `zz` and `a.txt` was never listed. The token + /// is now base64 of the plain key, and this test keeps guarding the + /// continuation boundary. #[tokio::test] async fn test_list_objects_v2_continuation_keeps_keys_after_marker_stem() { init_logging(); diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index ed3590f44..b4b8cbba8 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -11314,13 +11314,7 @@ mod tests { let marker = marker .as_deref() .expect("production checkpoint should persist an object marker"); - let marker_suffix = marker - .strip_prefix(&expected_marker) - .expect("production checkpoint should stop at the first page boundary"); - assert!( - marker_suffix.is_empty() || marker_suffix.starts_with("[rustfs_cache:v2,"), - "unexpected production checkpoint marker suffix: {marker_suffix}" - ); + assert_eq!(marker, expected_marker, "production checkpoint should stop at the first page boundary"); assert_eq!(version_marker.as_deref(), Some("null")); } diff --git a/crates/ecstore/src/store/list_objects.rs b/crates/ecstore/src/store/list_objects.rs index 6f1bafb2d..adfd5d012 100644 --- a/crates/ecstore/src/store/list_objects.rs +++ b/crates/ecstore/src/store/list_objects.rs @@ -426,7 +426,6 @@ const LIST_CURSOR_SOURCE_WALKER: &str = "walker"; const LIST_CURSOR_SOURCE_INDEX_KEY_ONLY: &str = "index_key_only"; const LIST_CURSOR_SOURCE_INDEX_VERIFIED_PAGE: &str = "index_verified_page"; const LIST_CURSOR_SOURCE_INDEX_METADATA_FAST: &str = "index_metadata_fast"; -const LIST_CURSOR_GENERATION_LIVE: &str = "live"; const DEFAULT_API_LIST_QUORUM: &str = "strict"; const ENV_API_LIST_OBJECTS_QUORUM: &str = "RUSTFS_LIST_OBJECTS_QUORUM"; const DEFAULT_API_LIST_OBJECTS_QUORUM: &str = "optimal"; @@ -446,15 +445,6 @@ pub enum ListSourceMode { } impl ListSourceMode { - fn cursor_value(self) -> &'static str { - match self { - Self::Walker => LIST_CURSOR_SOURCE_WALKER, - Self::IndexKeyOnly => LIST_CURSOR_SOURCE_INDEX_KEY_ONLY, - Self::IndexVerifiedPage => LIST_CURSOR_SOURCE_INDEX_VERIFIED_PAGE, - Self::IndexMetadataFast => LIST_CURSOR_SOURCE_INDEX_METADATA_FAST, - } - } - fn from_cursor_value(source: &str) -> Option { match source { LIST_CURSOR_SOURCE_WALKER => Some(Self::Walker), @@ -497,38 +487,6 @@ struct ListContinuationV2 { } impl ListContinuationV2 { - fn encode_marker(&self, marker: &str) -> String { - let mut marker_tag = String::with_capacity(marker.len() + 64); - marker_tag.push_str(marker); - marker_tag.push_str(LIST_CACHE_MARKER_PREFIX); - marker_tag.push_str(self.version); - match &self.id { - Some(id) => { - marker_tag.push_str(",id:"); - marker_tag.push_str(id); - } - None => marker_tag.push_str(",return:"), - } - if let Some(pool_idx) = self.pool_idx { - marker_tag.push_str(",p:"); - marker_tag.push_str(&pool_idx.to_string()); - } - if let Some(set_idx) = self.set_idx { - marker_tag.push_str(",s:"); - marker_tag.push_str(&set_idx.to_string()); - } - if let Some(source) = &self.source { - marker_tag.push_str(",src:"); - marker_tag.push_str(source.cursor_value()); - } - if let Some(generation) = &self.generation { - marker_tag.push_str(",gen:"); - marker_tag.push_str(generation); - } - marker_tag.push(']'); - marker_tag - } - fn parse(marker_tag: &str) -> Option<(Self, bool)> { let mut parsed = Self { version: MARKER_TAG_VERSION, @@ -597,6 +555,11 @@ impl ListContinuationV2 { } } +/// Returns the key part of a marker that may still carry the legacy cursor tag. +/// The data-usage scanner compares successive version-listing markers through +/// it so two differently tagged markers for the same key never count as progress. +// RUSTFS_COMPAT_TODO(listing-cursor-legacy-tag): markers tagged by 1.0.x servers. Remove after every supported +// direct-upgrade release issues plain key markers; callers can then compare markers directly. pub(crate) fn list_marker_key(marker: &str) -> &str { marker .rfind(LIST_CACHE_MARKER_PREFIX) @@ -624,22 +587,6 @@ fn list_objects_quorum_from_env() -> String { normalize_list_quorum(&value).to_owned() } -fn append_list_cache_id_to_marker(marker: String, cache_id: Option<&str>) -> String { - let Some(id) = cache_id else { - return marker; - }; - - ListContinuationV2 { - version: MARKER_TAG_VERSION, - id: Some(id.to_owned()), - pool_idx: None, - set_idx: None, - source: Some(ListSourceMode::Walker), - generation: Some(LIST_CURSOR_GENERATION_LIVE.to_owned()), - } - .encode_marker(&marker) -} - /// Name of the last RAW entry the walker returned for this page (after /// `forward_past`), captured before delimiter folding collapses many keys into a /// few common prefixes. Used as a continuation marker so a fully re-folded page @@ -683,21 +630,31 @@ fn versioned_prefix_page_boundary_key<'a>( 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)) +/// Ordinary cursors are plain keys. Cache-shaped literal keys retain an +/// old-readable escape until legacy tag parsing can be removed. +// RUSTFS_COMPAT_TODO(listing-cursor-legacy-tag): Remove after the minimum upgrade source stops issuing tagged cursors. +fn escape_list_marker(marker: String) -> String { + // Both the legacy request parser and scanner cursor comparison recognize + // cache-shaped suffixes. One old-readable outer tag protects the literal key. + if marker + .rfind(LIST_CACHE_MARKER_PREFIX) + .is_some_and(|start| marker[start..].contains(']')) + { + format!("{marker}[rustfs_cache:v2,return:]") } else { - prefixes - .last() - .map(|marker| append_list_cache_id_to_marker(marker.clone(), cache_id)) + marker } } -fn build_list_versions_next_marker( - objects: &[ObjectInfo], - prefixes: &[String], - cache_id: Option<&str>, -) -> (Option, Option) { +fn build_list_next_marker(objects: &[ObjectInfo], prefixes: &[String]) -> Option { + objects + .last() + .map(|last| last.name.clone()) + .or_else(|| prefixes.last().cloned()) + .map(escape_list_marker) +} + +fn build_list_versions_next_marker(objects: &[ObjectInfo], prefixes: &[String]) -> (Option, Option) { if let Some(last) = objects.last() { // A null version carries the synthesized `Some(Uuid::nil())` identity // here; advertise it as the literal `null` marker so a resumed listing @@ -715,9 +672,9 @@ fn build_list_versions_next_marker( .unwrap_or_else(|| "null".to_string()), ) }; - (Some(append_list_cache_id_to_marker(last.name.clone(), cache_id)), version_marker) + (Some(escape_list_marker(last.name.clone())), version_marker) } else if let Some(last_prefix) = prefixes.last() { - (Some(append_list_cache_id_to_marker(last_prefix.clone(), cache_id)), None) + (Some(escape_list_marker(last_prefix.clone())), None) } else { (None, None) } @@ -728,7 +685,6 @@ fn list_objects_paginate( delimiter: &Option, max_keys: i32, disk_has_more: bool, - cache_id: Option<&str>, include_version_id: bool, last_scanned_key: Option<&str>, ) -> (Vec, Vec, bool, Option, Option) { @@ -746,9 +702,9 @@ fn list_objects_paginate( if is_truncated { if include_version_id { - (next_marker, next_version_idmarker) = build_list_versions_next_marker(&get_objects, &[], cache_id); + (next_marker, next_version_idmarker) = build_list_versions_next_marker(&get_objects, &[]); } else { - next_marker = build_list_next_marker(&get_objects, &[], cache_id); + next_marker = build_list_next_marker(&get_objects, &[]); } } @@ -781,13 +737,13 @@ fn list_objects_paginate( && 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)); + next_marker = Some(escape_list_marker(last_scanned.to_owned())); } 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); + (next_marker, next_version_idmarker) = build_list_versions_next_marker(&objects, &prefixes); } else { - next_marker = build_list_next_marker(&objects, &prefixes, cache_id); + next_marker = build_list_next_marker(&objects, &prefixes); } } else if delimiter.is_some() && let Some(last_scanned) = last_scanned_key @@ -806,7 +762,7 @@ fn list_objects_paginate( // forever. A real scanned key strictly advances past the already-scanned // window and guarantees finite pagination. is_truncated = true; - next_marker = Some(append_list_cache_id_to_marker(last_scanned.to_owned(), cache_id)); + next_marker = Some(escape_list_marker(last_scanned.to_owned())); // A bare key marker with no version id makes `forward_past` skip the whole // scanned window on the next page; its folded prefixes are already emitted. } @@ -819,7 +775,6 @@ struct VersionedListPagination<'a, I> { delimiter: &'a Option, max_keys: i32, disk_has_more: bool, - cache_id: Option<&'a str>, marker: Option<&'a str>, last_scanned_key: Option<&'a str>, raw_keys: I, @@ -846,12 +801,11 @@ where pagination.delimiter, pagination.max_keys, pagination.disk_has_more, - pagination.cache_id, true, pagination.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, pagination.cache_id)); + next_marker = Some(escape_list_marker(raw_key)); next_version_idmarker = None; } @@ -1481,6 +1435,21 @@ impl ListPathOptions { } } + /// Strips a trailing legacy cursor tag from the request marker. + /// + /// Releases up to 1.0.x appended `[rustfs_cache:v2,id:...,src:walker,gen:live]` + /// to every internal continuation marker and exposed it through the V2 + /// continuation token and the ListObjectVersions `NextKeyMarker`. Current + /// servers emit plain keys, but a client that paginates across the upgrade + /// still sends the tagged form back as `continuation-token`, `key-marker` + /// or `marker`, and the opt-in index provider still carries its `src`/`gen` + /// cursor fields in the same tag. + /// + /// Newly issued cursors protect cache-shaped literal keys with one legacy + /// outer tag. Parse only at the listing entry point; parsing the resulting + /// literal key again would move the walker back to its stem. + // RUSTFS_COMPAT_TODO(listing-cursor-legacy-tag): accept markers tagged by 1.0.x servers. Remove after every supported + // direct-upgrade release issues plain key markers, so no client can still hold a tagged cursor. pub fn parse_marker(&mut self) { let Some(marker) = self.marker.clone() else { return; @@ -1509,18 +1478,6 @@ impl ListPathOptions { self.create = true; } } - - pub fn encode_marker(&mut self, marker: &str) -> String { - ListContinuationV2 { - version: MARKER_TAG_VERSION, - id: self.id.clone(), - pool_idx: self.pool_idx, - set_idx: self.set_idx, - source: Some(ListSourceMode::Walker), - generation: Some(LIST_CURSOR_GENERATION_LIVE.to_owned()), - } - .encode_marker(marker) - } } impl ECStore { @@ -1589,7 +1546,7 @@ impl ECStore { ask_disks: list_objects_quorum_from_env(), ..Default::default() }; - // The request marker still carries the `[rustfs_cache:...]` cursor tag; + // A marker from a 1.0.x client may still carry the legacy cursor tag; // strip it before any name comparison (notably `forward_past`), or every // key sorting between `` and `[` is silently skipped on // the continuation page (backlog#1047). @@ -1599,7 +1556,6 @@ impl ECStore { .list_path(&opts) .await .map_err(|err| to_object_err(err, vec![bucket, prefix]))?; - let next_cache_id = page.entries.list_id.clone(); let disk_has_more = !page.exhausted; @@ -1612,15 +1568,8 @@ impl ECStore { let get_objects = ObjectInfo::from_meta_cache_entries_sorted_infos(&page.entries, bucket, prefix, delimiter.clone()).await; - let (objects, prefixes, is_truncated, next_marker, next_version_idmarker) = list_objects_paginate( - get_objects, - &delimiter, - max_keys, - disk_has_more, - next_cache_id.as_deref(), - false, - last_scanned_key.as_deref(), - ); + let (objects, prefixes, is_truncated, next_marker, next_version_idmarker) = + list_objects_paginate(get_objects, &delimiter, max_keys, disk_has_more, false, last_scanned_key.as_deref()); let _ = next_version_idmarker; let result = ListObjectsInfo { @@ -1812,15 +1761,14 @@ impl ECStore { include_marker: has_version_marker, ..Default::default() }; - // Strip the `[rustfs_cache:...]` cursor tag before any name comparison - // (notably `forward_past`) — see backlog#1047. + // Strip a legacy cursor tag before any name comparison (notably + // `forward_past`), see backlog#1047. opts.parse_marker(); let mut page = self .list_path(&opts) .await .map_err(|err| to_object_err(err, vec![bucket, prefix]))?; - let next_cache_id = page.entries.list_id.clone(); let disk_has_more = !page.exhausted; @@ -1853,7 +1801,6 @@ impl ECStore { delimiter: &delimiter, max_keys, disk_has_more, - cache_id: next_cache_id.as_deref(), marker: opts.marker.as_deref(), last_scanned_key: last_scanned_key.as_deref(), raw_keys: entries.entries().iter().map(|entry| entry.name.as_str()), @@ -1908,8 +1855,6 @@ impl ECStore { o.skip_hidden_prefix_check = can_skip_hidden_prefix_check(&o).await; - o.parse_marker(); - if o.base_dir.is_empty() { o.base_dir = base_dir_from_prefix(&o.prefix); } @@ -1940,9 +1885,6 @@ impl ECStore { page.entries.reuse = true; page.entries.o.0.truncate(o.limit as usize); - if !o.transient { - page.entries.list_id = Some(o.id.unwrap_or_else(|| Uuid::new_v4().to_string())); - } rustfs_io_metrics::record_stage_duration( "store_list_objects_list_path", @@ -3133,7 +3075,6 @@ impl SetDisks { .list_path_result(&opts) .await .map_err(|err| to_object_err(err, vec![bucket, prefix]))?; - let next_cache_id = page.entries.list_id.clone(); let disk_has_more = !page.exhausted; if !has_version_marker { page.entries.forward_past(opts.marker.clone()); @@ -3149,15 +3090,8 @@ impl SetDisks { version_marker, ) .await?; - let (objects, prefixes, is_truncated, next_marker, next_version_idmarker) = list_objects_paginate( - get_objects, - &None, - max_keys, - disk_has_more, - next_cache_id.as_deref(), - true, - last_scanned_key.as_deref(), - ); + let (objects, prefixes, is_truncated, next_marker, next_version_idmarker) = + list_objects_paginate(get_objects, &None, max_keys, disk_has_more, true, last_scanned_key.as_deref()); Ok(ListObjectVersionsInfo { is_truncated, next_marker, @@ -3226,8 +3160,6 @@ impl SetDisks { o.skip_hidden_prefix_check = can_skip_hidden_prefix_check(&o).await; - o.parse_marker(); - if o.base_dir.is_empty() { o.base_dir = base_dir_from_prefix(&o.prefix); } @@ -3256,9 +3188,6 @@ impl SetDisks { page.entries.reuse = true; page.entries.o.0.truncate(o.limit as usize); - if !o.transient { - page.entries.list_id = Some(o.id.unwrap_or_else(|| Uuid::new_v4().to_string())); - } rustfs_io_metrics::record_stage_duration( "sets_list_objects_list_path", @@ -5171,8 +5100,9 @@ mod test { assert_eq!(parsed.pool_idx, Some(3)); assert_eq!(parsed.set_idx, Some(7)); + // A 1.0.x server re-tagged the stripped key the same way on replay. let base_marker = parsed.marker.clone().expect("marker should parse"); - let replay_marker = parsed.encode_marker(base_marker.as_str()); + let replay_marker = format!("{base_marker}[rustfs_cache:v2,id:list-cache-id,p:3,s:7]"); let mut replay = ListPathOptions { marker: Some(replay_marker), ..Default::default() @@ -5245,14 +5175,191 @@ mod test { ..Default::default() }; - let (next_marker, next_version_idmarker) = super::build_list_versions_next_marker(&[null_version], &[], None); + let (next_marker, next_version_idmarker) = super::build_list_versions_next_marker(&[null_version], &[]); assert_eq!(next_marker.as_deref(), Some("obj-a")); assert_eq!(next_version_idmarker.as_deref(), Some("null")); - let (_, next_version_idmarker) = super::build_list_versions_next_marker(&[real_version], &[], None); + let (_, next_version_idmarker) = super::build_list_versions_next_marker(&[real_version], &[]); assert_eq!(next_version_idmarker.as_deref(), Some(uuid::Uuid::from_u128(7).to_string().as_str())); } + // The internal continuation marker is the plain last key. Earlier releases + // appended a metacache tag that no server component consumed and that leaked + // onto the wire as the ListObjectVersions NextKeyMarker. + #[test] + fn list_objects_paginate_next_marker_is_the_plain_last_key() { + let get_objects = vec![plain_object("obj-0001"), plain_object("obj-0002"), plain_object("obj-0003")]; + + let (objects, prefixes, is_truncated, next_marker, next_version_idmarker) = + list_objects_paginate(get_objects, &None, 2, false, false, None); + + assert_eq!(objects.len(), 2); + assert!(prefixes.is_empty()); + assert!(is_truncated); + assert_eq!(next_marker.as_deref(), Some("obj-0002")); + assert!(next_version_idmarker.is_none()); + } + + #[test] + fn list_objects_paginate_versions_next_key_marker_is_the_plain_last_key() { + let version = uuid::Uuid::from_u128(7); + let mut get_objects = vec![plain_object("obj-0001"), plain_object("obj-0002"), plain_object("obj-0003")]; + get_objects[1].version_id = Some(version); + + let (_, _, is_truncated, next_marker, next_version_idmarker) = + list_objects_paginate(get_objects, &None, 2, false, true, None); + + assert!(is_truncated); + assert_eq!(next_marker.as_deref(), Some("obj-0002")); + assert_eq!(next_version_idmarker.as_deref(), Some(version.to_string().as_str())); + } + + #[test] + fn list_objects_paginate_preserves_literal_cache_tag_boundaries() { + for key in [ + "b[rustfs_cache:v1,return:]", + "b[rustfs_cache:v2,id:x]suffix", + "b[rustfs_cache:v2,id:x][rustfs_cache:v1,return:]", + "b[rustfs_cache:future]", + ] { + for include_version_id in [false, true] { + let (_, _, truncated, marker, _) = + list_objects_paginate(vec![plain_object(key), plain_object("zz")], &None, 1, false, include_version_id, None); + assert!(truncated); + let marker = marker.expect("truncated page needs a cursor"); + assert_eq!(list_marker_key(&marker), key, "scanner must compare the complete literal key"); + let mut opts = ListPathOptions { + marker: Some(marker), + ..Default::default() + }; + opts.parse_marker(); + assert_eq!(opts.marker.as_deref(), Some(key), "the next page must start after the literal key"); + } + } + } + + #[cfg(unix)] + #[tokio::test] + async fn literal_cache_tag_cursors_advance_through_real_objects_and_versions() { + use crate::storage_api_contracts::bucket::BucketOperations as _; + use crate::storage_api_contracts::object::ObjectIO as _; + + let bucket = "literal-cache-tag-cursors"; + let keys = [ + "b0", + "b1", + "b2", + "b[rustfs_cache:v1,return:]", + "c[rustfs_cache:v2,id:x]suffix", + "d[rustfs_cache:v2,id:x][rustfs_cache:v1,return:]", + "zz", + ]; + let (_dirs, store) = crate::bucket::metadata_sys::test_support::isolated_store_over_temp_disks().await; + crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await; + store + .make_bucket(bucket, &crate::storage_api_contracts::bucket::MakeBucketOptions::default()) + .await + .expect("create bucket"); + for key in keys { + store + .put_object( + bucket, + key, + &mut crate::object_api::PutObjReader::from_vec(vec![1]), + &crate::object_api::ObjectOptions::default(), + ) + .await + .expect("seed literal object"); + } + let mut marker = None; + let mut key_marker = None; + let mut version_marker = None; + for (index, key) in keys.into_iter().enumerate() { + let page = store + .clone() + .inner_list_objects_v2(bucket, "", marker, None, 1, false, None, false) + .await + .expect("list next object"); + assert_eq!(page.objects.iter().map(|object| object.name.as_str()).collect::>(), [key]); + assert_eq!(page.is_truncated, index + 1 < keys.len()); + marker = page.next_continuation_token; + + let versions = store + .clone() + .inner_list_object_versions(bucket, "", key_marker, version_marker, None, 1) + .await + .expect("list next version"); + assert_eq!(versions.objects.iter().map(|object| object.name.as_str()).collect::>(), [key]); + assert_eq!(versions.is_truncated, index + 1 < keys.len()); + key_marker = versions.next_marker; + version_marker = versions.next_version_idmarker; + } + assert!(marker.is_none()); + assert!(key_marker.is_none()); + } + + #[tokio::test] + async fn list_object_versions_next_key_marker_is_the_plain_last_key() { + use crate::storage_api_contracts::bucket::BucketOperations as _; + use crate::storage_api_contracts::object::ObjectIO as _; + + let bucket = "plain-key-marker"; + let (_dirs, store) = crate::bucket::metadata_sys::test_support::isolated_store_over_temp_disks().await; + crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await; + store + .make_bucket(bucket, &crate::storage_api_contracts::bucket::MakeBucketOptions::default()) + .await + .expect("create bucket"); + for key in ["obj-a", "obj-b", "obj-c"] { + store + .put_object( + bucket, + key, + &mut crate::object_api::PutObjReader::from_vec(vec![1]), + &crate::object_api::ObjectOptions::default(), + ) + .await + .expect("seed object"); + } + + let versions = store + .clone() + .inner_list_object_versions(bucket, "", None, None, None, 2) + .await + .expect("list versions"); + assert!(versions.is_truncated); + assert_eq!(versions.next_marker.as_deref(), Some("obj-b")); + assert_eq!(versions.next_version_idmarker.as_deref(), Some("null")); + + let page = store + .clone() + .inner_list_objects_v2(bucket, "", None, None, 2, false, None, false) + .await + .expect("list objects"); + assert!(page.is_truncated); + assert_eq!(page.next_continuation_token.as_deref(), Some("obj-b")); + + // A client that paginated across the upgrade still hands back the + // marker with the legacy tag appended; the walker resumes after `obj-b`. + let resumed = store + .clone() + .inner_list_objects_v2( + bucket, + "", + Some("obj-b[rustfs_cache:v2,id:list-cache-id,src:walker,gen:live]".to_string()), + None, + 2, + false, + None, + false, + ) + .await + .expect("resume from a legacy tagged marker"); + assert_eq!(resumed.objects.iter().map(|o| o.name.as_str()).collect::>(), vec!["obj-c"]); + assert!(!resumed.is_truncated); + assert!(resumed.next_continuation_token.is_none()); + } + // ECA-03 / #944: a page whose raw keys fully collapse into fewer than max_keys // common prefixes must still report truncation and carry a continuation marker, // otherwise every key beyond the scan window is silently dropped. @@ -5266,7 +5373,6 @@ mod test { &delimiter, 1000, true, // disk_has_more: walker filled its raw candidate limit - None, false, Some("data-1001"), // last RAW scanned key ); @@ -5287,7 +5393,7 @@ mod test { let get_objects = vec![folded_prefix_object("data-")]; let (_objects, prefixes, is_truncated, next_marker, _v) = - list_objects_paginate(get_objects, &delimiter, 1000, false, None, false, Some("data-1001")); + list_objects_paginate(get_objects, &delimiter, 1000, false, false, Some("data-1001")); assert_eq!(prefixes, vec!["data-".to_string()]); assert!(!is_truncated); @@ -5301,7 +5407,7 @@ mod test { let get_objects = vec![plain_object("obj-0001"), plain_object("obj-0002")]; let (objects, prefixes, is_truncated, next_marker, _v) = - list_objects_paginate(get_objects, &None, 1000, true, None, false, Some("zzz-unrelated")); + list_objects_paginate(get_objects, &None, 1000, true, false, Some("zzz-unrelated")); assert_eq!(objects.len(), 2); assert!(prefixes.is_empty()); @@ -5323,7 +5429,6 @@ mod test { &delimiter, 1000, true, // disk_has_more: walker filled its raw candidate limit - None, true, Some("data-1001"), // last RAW scanned key ); @@ -5347,7 +5452,7 @@ mod test { 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")); + list_objects_paginate(get_objects, &delimiter, 1, true, true, Some("data-0002")); assert!(objects.is_empty()); assert_eq!(prefixes, vec!["data-".to_string()]); @@ -5404,7 +5509,6 @@ mod test { delimiter: &delimiter, max_keys, disk_has_more, - cache_id: None, marker: marker.as_deref(), last_scanned_key: last_scanned.as_deref(), raw_keys: window.iter().map(String::as_str), @@ -5485,7 +5589,6 @@ mod test { delimiter: &delimiter, max_keys, disk_has_more, - cache_id: None, marker: marker.as_deref(), last_scanned_key: last_scanned.as_deref(), raw_keys: window.iter().map(String::as_str), @@ -5570,7 +5673,7 @@ mod test { let get_objects = fold_delimiter_page(&window, "", "-"); let (objects, prefixes, is_truncated, next_marker, _v) = - list_objects_paginate(get_objects, &delimiter, max_keys, disk_has_more, None, false, last_scanned.as_deref()); + list_objects_paginate(get_objects, &delimiter, max_keys, disk_has_more, false, last_scanned.as_deref()); assert!(objects.is_empty(), "every key folds into a common prefix"); for prefix in prefixes { @@ -7425,19 +7528,11 @@ mod test { #[test] fn list_path_marker_round_trip_preserves_set_index() { - let mut opts = ListPathOptions { - id: Some("list-cache-id".to_string()), - pool_idx: Some(3), - set_idx: Some(7), - ..Default::default() - }; - - let marker = opts.encode_marker("photos/2026/image.jpg"); - let expected_marker = format!( + // The exact tag shape a 1.0.x walker appended to its markers. + let marker = format!( "photos/2026/image.jpg[rustfs_cache:{},id:list-cache-id,p:3,s:7,src:walker,gen:live]", super::MARKER_TAG_VERSION ); - assert_eq!(marker, expected_marker); let mut parsed = ListPathOptions { marker: Some(marker), diff --git a/docs/architecture/compat-cleanup-register.md b/docs/architecture/compat-cleanup-register.md index 2a3173f98..6a65a4dd9 100644 --- a/docs/architecture/compat-cleanup-register.md +++ b/docs/architecture/compat-cleanup-register.md @@ -62,6 +62,7 @@ - `backlog-2097-tier-mutation-v4-error-text` tier-mutation Prepare rejection classification: a v3 server rejects a v4 request before store/runtime dispatch with the FailedPrecondition status and an authenticated, byte-exact unsupported-version message. A v4 coordinator recognizes only that exact code/message/requested-version tuple as definitely not persisted and fails the mutation without sending that peer an incompatible Abort; Unimplemented, near-text, missing/unknown failure classes, timeouts, and every other transport outcome remain ambiguous and stay in identity-bound Abort fanout. There is no automatic v3 retry. New servers retain v3 request/proof decoding for older coordinators, while operators must pause tier edit/remove/clear during a mixed v3/v4 rollout. Remove the text classifier after the minimum supported RustFS peer version returns the signed v4 PreDispatchRejected failure class. - `backlog-2097-tier-delete-journal-v6` tier-delete sole-owner recovery: v6 distinguishes transactions that may replace the xl.meta free-version owner and therefore require the all-pool live-source proof. v5-and-older readers reject and retain v6 records during rolling upgrades instead of performing an unsafe remote delete. Keep v1-v5 readers for upgrade recovery and keep the downgrade prohibition while any v6 record exists; retire the fence only after every supported rollback release understands and enforces v6 proof semantics. - `multipart-compression-default-off-window` staged multipart disk-compression rollout: releases before the resumable legacy decompressor fail transient reads of compressed objects under mid-payload suspension, so multipart uploads advertise the compression marker only when RUSTFS_COMPRESSION_MULTIPART_ENABLED is set in addition to RUSTFS_COMPRESSION_ENABLED, keeping rolling upgrades from creating new compressed multipart objects while pre-fix nodes may still serve reads. Flip the default to enabled (and retire the extra switch) after the minimum supported direct-upgrade release ships the resumable decompressor. +- `listing-cursor-legacy-tag` ListObjects cursor compatibility: releases up to 1.0.x appended a metacache tag ([rustfs_cache:v2,id:...,src:walker,gen:live]) to every internal continuation marker and exposed it through the base64 V2 continuation token and the ListObjectVersions NextKeyMarker; no server component consumed the tag, and it caused keys sorting between the marker and the tag to be skipped (backlog#1047). Current servers emit plain markers for ordinary keys; cache-shaped literal keys retain one legacy outer tag so both old and new readers resume after the complete key. V2 tokens remain base64 of the cursor. ListPathOptions::parse_marker decodes the cursor once at the listing entry point, and list_marker_key removes the outer tag for scanner comparisons. V1 responses remove only the conditional outer escape; the pre-existing ambiguity of incoming raw V1 markers remains during the compatibility window. Remove the tag stripping after every supported direct-upgrade release has issued plain key markers for a full release, so no client can still hold a tagged cursor. - `s3gate-trailer-adapter` aws-chunked trailer handle bridge: s3s decodes aws-chunked request bodies and publishes their x-amz-checksum-* trailers through its own handle, while rio consumes trailers only through its framework-neutral TrailerSource trait. The RustFS application crate adapts the s3s handle at the object write path so rio carries no s3s dependency. Remove the adapter after the gateway stack replaces s3s as the request body decoder and publishes trailers through its own TrailerSource implementation. ## Review Checklist diff --git a/rustfs/src/app/bucket_list_through.rs b/rustfs/src/app/bucket_list_through.rs index 40713b952..5c282a467 100644 --- a/rustfs/src/app/bucket_list_through.rs +++ b/rustfs/src/app/bucket_list_through.rs @@ -2193,9 +2193,9 @@ mod tests { }; if let Some((expected_second, expected_replay)) = second_page { let first_local = token.local.clone().expect("first local page was fully consumed"); - assert!( - first_local.starts_with(&format!("{}[rustfs_cache:", expected_first[1])), - "expected a real opaque cache cursor: {first_local}" + assert_eq!( + first_local, expected_first[1], + "the local cursor is the plain last key, with no cache tag" ); input.continuation_token = Some(cursor); let second = execute_source_list(input.clone()).await.expect("second merged page").output; diff --git a/rustfs/src/app/bucket_usecase.rs b/rustfs/src/app/bucket_usecase.rs index c07cfbe82..7b7d0fa32 100644 --- a/rustfs/src/app/bucket_usecase.rs +++ b/rustfs/src/app/bucket_usecase.rs @@ -57,8 +57,8 @@ use super::storage_api::bucket_usecase::helper::{OperationHelper, spawn_backgrou use super::storage_api::bucket_usecase::object_utils::to_s3s_etag; use super::storage_api::bucket_usecase::s3_api::bucket::{ ListObjectVersionsParams, ListObjectsV2Params, build_list_buckets_output, build_list_object_versions_output, - build_list_objects_output, build_list_objects_v2_output, parse_list_object_versions_params, parse_list_objects_v2_params, - rustfs_owner, + build_list_objects_output, build_list_objects_v2_output, encode_list_objects_v2_continuation_token, + parse_list_object_versions_params, parse_list_objects_v2_params, rustfs_owner, }; use super::storage_api::bucket_usecase::{ get_validated_store, process_lambda_configurations, process_queue_configurations, process_topic_configurations, @@ -1238,7 +1238,7 @@ fn build_list_objects_v2_metadata_output( let key_count = metadata_count_to_i32(contents.len() + common_prefixes.len()); let next_continuation_token = object_infos .next_continuation_token - .map(|token| base64_simd::STANDARD.encode_to_string(token.as_bytes())); + .map(|cursor| encode_list_objects_v2_continuation_token(&cursor)); ListObjectsV2MetadataOutput { output: ListObjectsV2Output { diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index bcc848e99..b183571c8 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -1138,8 +1138,8 @@ pub(crate) mod s3_api { pub(crate) mod bucket { pub(crate) use crate::storage::storage_api::s3_api_consumer::bucket::{ ListObjectVersionsParams, ListObjectsV2Params, build_list_buckets_output, build_list_object_versions_output, - build_list_objects_output, build_list_objects_v2_output, parse_list_object_versions_params, - parse_list_objects_v2_params, + build_list_objects_output, build_list_objects_v2_output, encode_list_objects_v2_continuation_token, + parse_list_object_versions_params, parse_list_objects_v2_params, }; pub(crate) use crate::storage::storage_api::s3_api_consumer::common::rustfs_owner; } diff --git a/rustfs/src/storage/s3_api/bucket.rs b/rustfs/src/storage/s3_api/bucket.rs index a73cdff43..ccea63bb7 100644 --- a/rustfs/src/storage/s3_api/bucket.rs +++ b/rustfs/src/storage/s3_api/bucket.rs @@ -31,8 +31,31 @@ use uuid::Uuid; use crate::storage::storage_api::s3_api_consumer::bucket::StorageObjectInfo as ObjectInfo; const S3_MAX_KEYS: i32 = 1000; -const LIST_CACHE_MARKER_TAG_PREFIX: &str = "[rustfs_cache:"; -const LIST_CACHE_MARKER_TAG_VERSIONS: &[&str] = &["v1", "v2"]; + +fn invalid_continuation_token() -> S3Error { + S3Error::with_message(S3ErrorCode::InvalidArgument, "Invalid continuation token".to_string()) +} + +/// Encodes the server-side cursor into the wire `NextContinuationToken`. +/// +/// AWS defines the token as opaque, so nothing in it is a promise to clients. +/// The cursor is the plain last key of an ordinary listing, or the envelope +/// another listing layer (on-demand migration list-through, table isolation) +/// hands out; base64 keeps it URL-safe and is the only transform applied, so +/// nodes from earlier releases read the same token. +pub(crate) fn encode_list_objects_v2_continuation_token(cursor: &str) -> String { + base64_simd::STANDARD.encode_to_string(cursor.as_bytes()) +} + +/// Decodes a wire `ContinuationToken` into the server-side cursor it carries. +/// A cursor issued by a 1.0.x server may still end in the metacache tag; +/// ecstore strips it on the way in. +fn decode_list_objects_v2_continuation_token(token: &str) -> Result { + let bytes = base64_simd::STANDARD + .decode_to_vec(token.as_bytes()) + .map_err(|_| invalid_continuation_token())?; + String::from_utf8(bytes).map_err(|_| invalid_continuation_token()) +} type ListObjectVersionsInfo = StorageListObjectVersionsInfo; type ListObjectsV2Info = StorageListObjectsV2Info; @@ -76,6 +99,8 @@ pub(crate) struct ListObjectsV2Params { pub response_start_after: Option, pub start_after_for_query: Option, pub response_continuation_token: Option, + /// Server-side cursor carried by the request token; see + /// [`decode_list_objects_v2_continuation_token`]. pub decoded_continuation_token: Option, } @@ -173,19 +198,9 @@ pub(crate) fn parse_list_objects_v2_params( // Save original continuation_token for response (per S3 API spec, must echo back if provided) let response_continuation_token = continuation_token.clone(); - // Decode continuation_token from base64 for internal use let decoded_continuation_token = continuation_token .filter(|token| !token.is_empty()) - .map(|token| { - base64_simd::STANDARD - .decode_to_vec(token.as_bytes()) - .map_err(|_| S3Error::with_message(S3ErrorCode::InvalidArgument, "Invalid continuation token".to_string())) - .and_then(|bytes| { - String::from_utf8(bytes).map_err(|_| { - S3Error::with_message(S3ErrorCode::InvalidArgument, "Invalid continuation token".to_string()) - }) - }) - }) + .map(|token| decode_list_objects_v2_continuation_token(&token)) .transpose()?; Ok(ListObjectsV2Params { @@ -344,10 +359,9 @@ pub(crate) fn build_list_objects_v2_output( // KeyCount should include both objects and common prefixes per S3 API spec. let key_count = (objects.len() + common_prefixes.len()) as i32; - // Encode next_continuation_token to base64. let next_continuation_token = object_infos .next_continuation_token - .map(|token| base64_simd::STANDARD.encode_to_string(token.as_bytes())); + .map(|cursor| encode_list_objects_v2_continuation_token(&cursor)); ListObjectsV2Output { is_truncated: Some(object_infos.is_truncated), @@ -374,11 +388,17 @@ fn calculate_next_marker(v2: &ListObjectsV2Output) -> Option { return None; } - if let Some(marker) = decoded_next_continuation_marker(v2) { - return Some(encode_list_output_value( - public_list_marker(&marker).to_owned(), - v2.encoding_type.as_ref(), - )); + // The token was built by this server around the internal cursor, which is + // the last raw key the walker scanned: on a delimiter page that folded into + // fewer prefixes than max-keys it lies past the last visible entry. + if let Some(marker) = v2 + .next_continuation_token + .as_deref() + .and_then(|token| decode_list_objects_v2_continuation_token(token).ok()) + { + // Only cache-shaped literal keys carry this server-generated escape. + let marker = marker.strip_suffix("[rustfs_cache:v2,return:]").unwrap_or(&marker); + return Some(encode_list_output_value(marker.to_owned(), v2.encoding_type.as_ref())); } let last_key = v2 @@ -419,54 +439,12 @@ fn calculate_next_marker(v2: &ListObjectsV2Output) -> Option { } } -fn public_list_marker(marker: &str) -> &str { - let Some(start_idx) = marker.rfind(LIST_CACHE_MARKER_TAG_PREFIX) else { - return marker; - }; - let Some(tag_body) = marker[start_idx + 1..].strip_suffix(']') else { - return marker; - }; - - if is_list_cache_marker_tag(tag_body) { - &marker[..start_idx] - } else { - marker - } -} - -fn is_list_cache_marker_tag(tag_body: &str) -> bool { - let mut has_supported_version = false; - let mut has_continuation_field = false; - - for tag in tag_body.split(',') { - let Some((key, value)) = tag.split_once(':') else { - continue; - }; - - match key { - "rustfs_cache" => has_supported_version = LIST_CACHE_MARKER_TAG_VERSIONS.contains(&value), - "id" if !value.is_empty() => has_continuation_field = true, - "return" => has_continuation_field = true, - "p" | "s" if value.parse::().is_err() => return false, - _ => {} - } - } - - has_supported_version && has_continuation_field -} - -fn decoded_next_continuation_marker(v2: &ListObjectsV2Output) -> Option { - let token = v2.next_continuation_token.as_ref()?; - let bytes = base64_simd::STANDARD.decode_to_vec(token.as_bytes()).ok()?; - String::from_utf8(bytes).ok() -} - #[cfg(test)] mod tests { use super::{ ListObjectVersionsInfo, ListObjectVersionsParams, ListObjectsV2Info, build_list_buckets_output, build_list_object_versions_output, build_list_objects_output, build_list_objects_v2_output, - parse_list_object_versions_params, parse_list_objects_v2_params, to_s3s_etag, + encode_list_objects_v2_continuation_token, parse_list_object_versions_params, parse_list_objects_v2_params, to_s3s_etag, }; use crate::storage::s3_api::common::rustfs_owner; use crate::storage::storage_api::s3_api_consumer::bucket::StorageObjectInfo as ObjectInfo; @@ -551,30 +529,30 @@ mod tests { assert_eq!(output.next_marker, Some("zebra/".to_string())); } + // The V1 NextMarker is the cursor carried by the V2 token: the last raw key + // the walker scanned, which on a re-folded delimiter page lies past the last + // visible common prefix (ECA-03 / #944). #[test] - fn test_list_objects_next_marker_removes_internal_continuation_marker() { - let marker = "key-998[rustfs_cache:v2,id:list-cache-id,p:0,s:0]"; + fn test_list_objects_next_marker_uses_the_cursor_inside_the_v2_token() { let v2 = ListObjectsV2Output { is_truncated: Some(true), - next_continuation_token: Some(base64_simd::STANDARD.encode_to_string(marker.as_bytes())), - contents: Some(vec![Object { - key: Some("key-998".to_string()), - ..Default::default() + next_continuation_token: Some(encode_list_objects_v2_continuation_token("data-1001")), + common_prefixes: Some(vec![CommonPrefix { + prefix: Some("data-".to_string()), }]), ..Default::default() }; let output = build_list_objects_output(v2, None); - assert_eq!(output.next_marker.as_deref(), Some("key-998")); + assert_eq!(output.next_marker.as_deref(), Some("data-1001")); } #[test] - fn test_list_objects_next_marker_strips_internal_marker_when_url_encoded() { - let marker = "dir a/key[rustfs_cache:v2,id:list-cache-id,p:0,s:0]"; + fn test_list_objects_next_marker_url_encodes_the_v2_cursor() { let v2 = ListObjectsV2Output { is_truncated: Some(true), encoding_type: Some(EncodingType::from_static(EncodingType::URL)), - next_continuation_token: Some(base64_simd::STANDARD.encode_to_string(marker.as_bytes())), + next_continuation_token: Some(encode_list_objects_v2_continuation_token("dir a/key")), contents: Some(vec![Object { key: Some("dir%20a/key".to_string()), ..Default::default() @@ -586,37 +564,36 @@ mod tests { assert_eq!(output.next_marker.as_deref(), Some("dir%20a/key")); } + // A key that happens to end in the shape of the retired metacache tag is a + // key: the marker never carried a tag to strip, so nothing is stripped. #[test] - fn test_list_objects_next_marker_uses_visible_common_prefix_with_internal_token() { - let marker = "asdf[rustfs_cache:v2,id:list-cache-id,p:0,s:0]"; + fn test_list_objects_next_marker_keeps_a_legacy_looking_key_suffix() { + let key = "key[rustfs_cache:v2,id:list-cache-id,src:walker,gen:live]"; let v2 = ListObjectsV2Output { is_truncated: Some(true), - next_continuation_token: Some(base64_simd::STANDARD.encode_to_string(marker.as_bytes())), - common_prefixes: Some(vec![CommonPrefix { - prefix: Some("asdf".to_string()), - }]), - ..Default::default() - }; - - let output = build_list_objects_output(v2, None); - assert_eq!(output.next_marker.as_deref(), Some("asdf")); - } - - #[test] - fn test_list_objects_next_marker_preserves_cache_like_key_suffix() { - let marker = "key[rustfs_cache:v2]"; - let v2 = ListObjectsV2Output { - is_truncated: Some(true), - next_continuation_token: Some(base64_simd::STANDARD.encode_to_string(marker.as_bytes())), + next_continuation_token: Some(encode_list_objects_v2_continuation_token(key)), contents: Some(vec![Object { - key: Some(marker.to_string()), + key: Some(key.to_string()), ..Default::default() }]), ..Default::default() }; let output = build_list_objects_output(v2, None); - assert_eq!(output.next_marker.as_deref(), Some(marker)); + assert_eq!(output.next_marker.as_deref(), Some(key)); + } + + #[test] + fn test_list_objects_next_marker_unwraps_only_the_literal_key_escape() { + let key = "key[rustfs_cache:v2,return:]"; + let cursor = format!("{key}[rustfs_cache:v2,return:]"); + let v2 = ListObjectsV2Output { + is_truncated: Some(true), + next_continuation_token: Some(encode_list_objects_v2_continuation_token(&cursor)), + ..Default::default() + }; + + assert_eq!(build_list_objects_output(v2, None).next_marker.as_deref(), Some(key)); } #[test] @@ -895,7 +872,7 @@ mod tests { } #[test] - fn test_list_objects_v2_next_continuation_token_is_base64_encoded() { + fn test_list_objects_v2_next_continuation_token_is_base64_of_the_cursor() { let object_infos = ListObjectsV2Info { next_continuation_token: Some("token-123".to_string()), ..Default::default() @@ -915,10 +892,25 @@ mod tests { assert_eq!(output.continuation_token, Some("echo-token".to_string())); assert_eq!(output.start_after, Some("start-after".to_string())); - assert_eq!( - output.next_continuation_token, - Some(base64_simd::STANDARD.encode_to_string("token-123".as_bytes())) - ); + let wire = output.next_continuation_token.expect("truncated page carries a token"); + assert_eq!(wire, base64_simd::STANDARD.encode_to_string("token-123".as_bytes())); + assert!(!wire.contains("rustfs_cache")); + + let parsed = parse_list_objects_v2_params(None, None, Some(1000), Some(wire.clone()), None).expect("round trip"); + assert_eq!(parsed.response_continuation_token, Some(wire)); + assert_eq!(parsed.decoded_continuation_token.as_deref(), Some("token-123")); + } + + // A key that literally ends in the retired metacache tag survives the V2 + // round trip unchanged: the API layer no longer strips anything. + #[test] + fn test_parse_list_objects_v2_params_round_trips_a_legacy_looking_literal_key() { + let key = "b[rustfs_cache:v1,return:]"; + let wire = encode_list_objects_v2_continuation_token(key); + + let parsed = parse_list_objects_v2_params(None, None, Some(2), Some(wire), None).expect("parse should succeed"); + + assert_eq!(parsed.decoded_continuation_token.as_deref(), Some(key)); } #[test] @@ -986,16 +978,19 @@ mod tests { assert_eq!(parsed.decoded_continuation_token, None); } + // A token issued by a 1.0.x server is base64 of the marker with the metacache + // tag appended; it still paginates, and the tag is left for ecstore to strip. #[test] - fn test_parse_list_objects_v2_params_decodes_continuation_token() { - let raw = "token-123"; - let encoded = base64_simd::STANDARD.encode_to_string(raw.as_bytes()); + fn test_parse_list_objects_v2_params_decodes_plain_and_legacy_tagged_tokens() { + for raw in ["token-123", "key-998[rustfs_cache:v2,id:list-cache-id,src:walker,gen:live]"] { + let encoded = base64_simd::STANDARD.encode_to_string(raw.as_bytes()); - let parsed = - parse_list_objects_v2_params(None, None, Some(1000), Some(encoded.clone()), None).expect("parse should succeed"); + let parsed = + parse_list_objects_v2_params(None, None, Some(1000), Some(encoded.clone()), None).expect("parse should succeed"); - assert_eq!(parsed.response_continuation_token, Some(encoded)); - assert_eq!(parsed.decoded_continuation_token, Some(raw.to_string())); + assert_eq!(parsed.response_continuation_token, Some(encoded)); + assert_eq!(parsed.decoded_continuation_token.as_deref(), Some(raw)); + } } #[test] @@ -1099,6 +1094,36 @@ mod tests { ); } + // ListObjectVersions NextKeyMarker is the plain last key of the page, as AWS + // defines it; earlier releases leaked the internal metacache tag here. + #[test] + fn test_build_list_object_versions_output_next_key_marker_is_the_plain_last_key() { + let object_infos = ListObjectVersionsInfo { + is_truncated: true, + next_marker: Some("obj-b".to_string()), + next_version_idmarker: Some("null".to_string()), + objects: vec![object_info("obj-a"), object_info("obj-b")], + prefixes: Vec::new(), + }; + + let output = build_list_object_versions_output( + object_infos, + "bucket-a".to_string(), + &ListObjectVersionsParams { + prefix: String::new(), + delimiter: None, + key_marker: None, + version_id_marker: None, + max_keys: 2, + }, + None, + ); + + assert_eq!(output.is_truncated, Some(true)); + assert_eq!(output.next_key_marker.as_deref(), Some("obj-b")); + assert_eq!(output.next_version_id_marker.as_deref(), Some("null")); + } + #[test] fn test_build_list_object_versions_output_url_encodes_response_fields() { let object_infos = ListObjectVersionsInfo { diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index e89b67fe7..2705d9721 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -329,8 +329,8 @@ pub(crate) mod s3_api_consumer { pub(crate) mod bucket { pub(crate) use super::super::super::s3_api::bucket::{ ListObjectVersionsParams, ListObjectsV2Params, build_list_buckets_output, build_list_object_versions_output, - build_list_objects_output, build_list_objects_v2_output, parse_list_object_versions_params, - parse_list_objects_v2_params, + build_list_objects_output, build_list_objects_v2_output, encode_list_objects_v2_continuation_token, + parse_list_object_versions_params, parse_list_objects_v2_params, }; pub(crate) mod contract {