fix(odm): bound empty pagination chains with staged tokens (#7197)

* fix(odm): add staged cross-request pagination progress budgets

* fix(ecstore): remove duplicate local rename implementation

Keep the canonical commit module after concurrent storage changes merged.
The control-write and rollback changes are already present there.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(app): simplify absent SSE configuration matching

* fix(tests): satisfy new clippy lints

---------

Co-authored-by: houseme <housemecn@gmail.com>
Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>

rustfs/src/app/bucket_list_through.rs

@@ -26,8 +26,8 @@ use super::storage_api::bucket_usecase::ECStore;
 use super::storage_api::bucket_usecase::StorageObjectInfo as ObjectInfo;
 use super::storage_api::bucket_usecase::StorageObjectOptions;
 use super::storage_api::bucket_usecase::bucket::on_demand_migration::{
-    BucketOdmState, ListEntryKey, ListThroughCursor, ListThroughMerger, ListThroughToken, ListThroughTokenError, MergeSide,
-    OnDemandMigrationSys, SOURCE_LIST_MAX_RATE_WAIT, SourceClient, SourceError, SourceErrorPolicy, SourceListPlan,
+    BucketOdmState, ListEntryKey, ListPageError, ListThroughCursor, ListThroughMerger, ListThroughToken, ListThroughTokenError,
+    MergeSide, OnDemandMigrationSys, SOURCE_LIST_MAX_RATE_WAIT, SourceClient, SourceError, SourceErrorPolicy, SourceListPlan,
     SourceListRequest, SourceObject, SourcePage, decode_continuation_token, source_list_plan,
 };
 use super::storage_api::bucket_usecase::bucket::versioning_sys::BucketVersioningSys;
@@ -51,6 +51,9 @@ type ListObjectsV2Info = StorageListObjectsV2Info<ObjectInfo>;
 /// yet, so the only class RustFS can vouch for is the default one.
 const SOURCE_STORAGE_CLASS: &str = "STANDARD";
 
+/// Enable only after every node serving continuation requests can read v2.
+const ENV_LIST_PROGRESS_TOKENS: &str = "RUSTFS_ON_DEMAND_MIGRATION_LIST_V2_TOKENS";
+
 /// Concurrent local metadata probes when a versioned bucket has to check
 /// source-only keys for a shadowing delete marker.
 const DELETE_MARKER_PROBE_CONCURRENCY: usize = 32;
@@ -264,7 +267,18 @@ pub(crate) async fn merged_list_objects_v2(
         buffers[usize::from(fetch.side == MergeSide::Source)].extend(kept.into_iter().map(Some));
     }
 
-    let outcome = merger.finish();
+    let issue_progress_tokens = rustfs_utils::get_env_bool(ENV_LIST_PROGRESS_TOKENS, false);
+    let outcome = match merger.finish(issue_progress_tokens) {
+        Ok(outcome) => outcome,
+        Err(ListPageError::NoProgress(MergeSide::Source)) => {
+            degrade_or_fail(&mut merger, &mut degraded, policy.source_error, "invalid_pagination")?;
+            merger
+                .finish(issue_progress_tokens)
+                .map_err(|error| S3Error::with_message(S3ErrorCode::InternalError, error.to_string()))?
+        }
+        Err(error) => return Err(S3Error::with_message(S3ErrorCode::InternalError, error.to_string())),
+    };
+    drop(merger);
     let mut objects = Vec::with_capacity(outcome.picks.len());
     let mut prefixes = Vec::new();
     let mut source_only_keys = Vec::new();
@@ -430,7 +444,8 @@ mod tests {
     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,
+        FilterConfig, MAX_LIST_NO_PROGRESS_PAGES, 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;
@@ -448,6 +463,7 @@ mod tests {
             source: Some("source-2".to_string()),
             source_done: false,
             last_key: Some("k".to_string()),
+            no_progress: None,
         }
     }
 
@@ -515,6 +531,24 @@ mod tests {
         assert!(matches!(local_cursor(Some(&encoded), decoded.as_ref()), LocalListCursor::Exhausted));
     }
 
+    #[test]
+    fn a_v2_token_keeps_the_local_cursor_when_list_through_is_turned_off() {
+        let mut resume = token(Some("local-2"), false);
+        resume.v = 2;
+        resume.no_progress = Some(MAX_LIST_NO_PROGRESS_PAGES - 1);
+        let encoded = resume.encode();
+        let decoded = decode_list_cursor(Some(&encoded)).expect("a v2 envelope decodes");
+        assert_eq!(decoded.as_ref(), Some(&resume));
+        assert!(matches!(
+            local_cursor(Some(&encoded), decoded.as_ref()),
+            LocalListCursor::Token(Some(local)) if local == "local-2"
+        ));
+        resume.local_done = true;
+        let encoded = resume.encode();
+        let decoded = decode_list_cursor(Some(&encoded)).expect("v2 with local EOF decodes");
+        assert!(matches!(local_cursor(Some(&encoded), decoded.as_ref()), LocalListCursor::Exhausted));
+    }
+
     #[test]
     fn a_plain_local_token_is_passed_through_and_a_tampered_one_is_rejected() {
         assert!(
@@ -551,14 +585,30 @@ mod tests {
     /// 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>) -> (String, tokio_util::task::AbortOnDropHandle<Vec<String>>) {
+        let (endpoint, server, _) = list_source(pages.into_iter()).await;
+        (endpoint, server)
+    }
+
+    async fn list_source(
+        pages: impl Iterator<Item = String> + Send + 'static,
+    ) -> (
+        String,
+        tokio_util::task::AbortOnDropHandle<Vec<String>>,
+        tokio_util::sync::CancellationToken,
+    ) {
         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 stop = tokio_util::sync::CancellationToken::new();
+        let server_stop = stop.clone();
         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 stream, _) = tokio::select! {
+                    _ = server_stop.cancelled() => break,
+                    accepted = listener.accept() => accepted.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") {
@@ -588,7 +638,7 @@ mod tests {
             }
             requests
         });
-        (format!("http://{address}"), tokio_util::task::AbortOnDropHandle::new(server))
+        (format!("http://{address}"), tokio_util::task::AbortOnDropHandle::new(server), stop)
     }
 
     fn source_xml(next: Option<&str>, truncated: bool, key: Option<&str>) -> String {
@@ -616,12 +666,12 @@ mod tests {
         }
     }
 
-    async fn source_policy_request(
-        pages: Vec<String>,
+    async fn source_policy_input(
+        endpoint: String,
         policy: SourceErrorPolicy,
         resume_source: Option<&str>,
         filter_prefix: Option<&str>,
-    ) -> (S3Result<S3Response<ListObjectsV2Output>>, Vec<String>) {
+    ) -> (ListThroughTestState, ListObjectsV2Input) {
         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());
@@ -638,9 +688,8 @@ mod tests {
             )
             .await
             .expect("seed real local listing");
-        let (endpoint, server) = scripted_list_source(pages).await;
         let sys = OnDemandMigrationSys::get();
-        let _state_guard = ListThroughTestState {
+        let state_guard = ListThroughTestState {
             bucket: bucket.clone(),
             module_enabled: sys.is_module_enabled(),
         };
@@ -685,6 +734,7 @@ mod tests {
                 source: Some(source.into()),
                 source_done: false,
                 last_key: None,
+                no_progress: None,
             };
             base64_simd::STANDARD.encode_to_string(token.encode().as_bytes())
         });
@@ -701,6 +751,10 @@ mod tests {
             request_payer: None,
             start_after: None,
         };
+        (state_guard, input)
+    }
+
+    async fn execute_source_list(input: ListObjectsV2Input) -> S3Result<S3Response<ListObjectsV2Output>> {
         let request = S3Request {
             input,
             method: http::Method::GET,
@@ -712,12 +766,23 @@ mod tests {
             service: None,
             trailing_headers: None,
         };
-        let result = tokio::time::timeout(
+        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");
+        .expect("listing must complete within its bounded source budget")
+    }
+
+    async fn source_policy_request(
+        pages: Vec<String>,
+        policy: SourceErrorPolicy,
+        resume_source: Option<&str>,
+        filter_prefix: Option<&str>,
+    ) -> (S3Result<S3Response<ListObjectsV2Output>>, Vec<String>) {
+        let (endpoint, server) = scripted_list_source(pages).await;
+        let (_state_guard, input) = source_policy_input(endpoint, policy, resume_source, filter_prefix).await;
+        let result = execute_source_list(input).await;
         let requests = tokio::time::timeout(Duration::from_secs(5), server)
             .await
             .expect("source connections must finish")
@@ -842,6 +907,244 @@ mod tests {
         });
     }
 
+    #[test]
+    #[serial_test::serial]
+    fn list_through_cross_request_empty_cursor_cycle_obeys_policy() {
+        run_large_stack_test("list-through-cross-request-cursor-cycle", || async {
+            temp_env::async_with_vars(
+                [
+                    (ENV_LIST_PROGRESS_TOKENS, Some("true")),
+                    ("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] {
+                        let pages = ["B", "C", "A"].map(|next| source_xml(Some(next), true, None));
+                        let (endpoint, server, stop) = list_source(pages.into_iter().cycle()).await;
+                        let (_state_guard, mut input) = source_policy_input(endpoint, policy, Some("A"), None).await;
+                        let mut seen = std::collections::HashSet::from([input
+                            .continuation_token
+                            .clone()
+                            .expect("the first request resumes source cursor A")]);
+                        let mut client_requests = 0;
+                        let mut empty_pages = 0;
+                        let terminal = tokio::time::timeout(Duration::from_secs(30), async {
+                            loop {
+                                client_requests += 1;
+                                let response = match execute_source_list(input.clone()).await {
+                                    Ok(response) => response,
+                                    Err(error) => break Err(error),
+                                };
+                                if response.headers.contains_key("x-rustfs-on-demand-migration-list") {
+                                    break Ok(response);
+                                }
+                                let output = response.output;
+                                assert!(output.contents.as_ref().is_none_or(Vec::is_empty));
+                                assert!(output.common_prefixes.as_ref().is_none_or(Vec::is_empty));
+                                assert_eq!(output.key_count, Some(0));
+                                assert_eq!(output.is_truncated, Some(true));
+                                let next = output
+                                    .next_continuation_token
+                                    .expect("a truncated page must carry its cursor");
+                                assert!(
+                                    seen.insert(next.clone()),
+                                    "a cross-request source cursor cycle must not return an identical empty merged token"
+                                );
+                                empty_pages += 1;
+                                input.continuation_token = Some(next);
+                            }
+                        })
+                        .await
+                        .expect("a source cursor cycle must terminate within a bounded client pagination chain");
+                        assert_eq!(empty_pages, usize::from(MAX_LIST_NO_PROGRESS_PAGES - 1));
+                        assert_eq!(client_requests, usize::from(MAX_LIST_NO_PROGRESS_PAGES));
+                        assert_source_policy_result(terminal, policy);
+                        stop.cancel();
+                        let requests = tokio::time::timeout(Duration::from_secs(5), server)
+                            .await
+                            .expect("cyclic source server must stop")
+                            .expect("cyclic source server must not panic");
+                        assert_eq!(requests.len(), 2 * client_requests, "the sixteenth empty page exhausts the budget");
+                        for (index, request) in requests.iter().enumerate() {
+                            let source_cursor = ["A", "B", "C"][index % 3];
+                            assert!(
+                                request.contains(&format!("continuation-token={source_cursor}")),
+                                "the real SDK must follow the returned source cursor: {request}"
+                            );
+                        }
+                    }
+                },
+            )
+            .await;
+        });
+    }
+
+    #[test]
+    #[serial_test::serial]
+    fn list_through_default_rollout_continues_v2_without_issuing_it_from_v1() {
+        run_large_stack_test("list-through-reader-first-rollout", || async {
+            temp_env::async_with_vars(
+                [
+                    (ENV_LIST_PROGRESS_TOKENS, None),
+                    ("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] {
+                        let pages = ["B", "C", "A"].map(|next| source_xml(Some(next), true, None));
+                        let (endpoint, server, stop) = list_source(pages.into_iter().cycle()).await;
+                        let (_state_guard, mut input) = source_policy_input(endpoint, policy, Some("A"), None).await;
+                        let original = input.continuation_token.clone();
+                        for _ in 0..3 {
+                            let response = execute_source_list(input.clone()).await.expect("reader-only v1 behavior");
+                            assert!(!response.headers.contains_key("x-rustfs-on-demand-migration-list"));
+                            assert_eq!(response.output.key_count, Some(0));
+                            assert_eq!(response.output.is_truncated, Some(true));
+                            let next = response.output.next_continuation_token.expect("resumable empty page");
+                            let raw = base64_simd::STANDARD.decode_to_vec(&next).expect("base64 continuation token");
+                            let decoded = std::str::from_utf8(&raw).expect("JSON token");
+                            let token = decode_list_cursor(Some(decoded)).expect("v1 reader").expect("merged token");
+                            assert_eq!(token.v, 1, "the default rollout cannot begin issuing v2");
+                            assert_eq!(token.no_progress, None);
+                            assert!(!decoded.contains("no_progress"), "ordinary v1 wire shape stays unchanged");
+                            input.continuation_token = Some(next);
+                        }
+                        assert_eq!(input.continuation_token, original, "default rollout retains the known v1 limitation");
+
+                        let raw = base64_simd::STANDARD
+                            .decode_to_vec(input.continuation_token.as_ref().expect("v1 token"))
+                            .expect("base64 continuation token");
+                        let mut token = decode_list_cursor(Some(std::str::from_utf8(&raw).expect("JSON token")))
+                            .expect("v1 reader")
+                            .expect("merged token");
+                        token.v = 2;
+                        token.no_progress = Some(MAX_LIST_NO_PROGRESS_PAGES - 2);
+                        input.continuation_token = Some(base64_simd::STANDARD.encode_to_string(token.encode().as_bytes()));
+                        let response = execute_source_list(input.clone()).await.expect("reader-only node resumes v2");
+                        assert_eq!(response.output.key_count, Some(0));
+                        assert_eq!(response.output.is_truncated, Some(true));
+                        let next = response.output.next_continuation_token.expect("last allowed empty cursor");
+                        let raw = base64_simd::STANDARD.decode_to_vec(&next).expect("base64 continuation token");
+                        let token = decode_list_cursor(Some(std::str::from_utf8(&raw).expect("JSON token")))
+                            .expect("v2 reader")
+                            .expect("merged token");
+                        assert_eq!(token.v, 2);
+                        assert_eq!(token.no_progress, Some(MAX_LIST_NO_PROGRESS_PAGES - 1));
+                        input.continuation_token = Some(next);
+                        assert_source_policy_result(execute_source_list(input).await, policy);
+                        stop.cancel();
+                        let requests = tokio::time::timeout(Duration::from_secs(5), server)
+                            .await
+                            .expect("cyclic source server must stop")
+                            .expect("source server must not panic");
+                        assert_eq!(requests.len(), 10, "five handler requests each fetched two source pages");
+                        for (index, request) in requests.iter().enumerate() {
+                            let cursor = ["A", "B", "C"][index % 3];
+                            assert!(request.contains(&format!("continuation-token={cursor}")), "{request}");
+                        }
+                    }
+                },
+            )
+            .await;
+        });
+    }
+
+    #[test]
+    #[serial_test::serial]
+    fn list_through_empty_advancing_pages_resume_across_handler_requests() {
+        run_large_stack_test("list-through-resumable-empty-pages", || async {
+            temp_env::async_with_vars(
+                [
+                    (ENV_LIST_PROGRESS_TOKENS, Some("true")),
+                    ("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 = filter_prefix.map_or("a-source", |_| "photos/2024/a-source");
+                        let (endpoint, server) = scripted_list_source(vec![
+                            source_xml(Some("A"), true, None),
+                            source_xml(Some("B"), true, None),
+                            source_xml(Some("C"), true, None),
+                            source_xml(None, false, Some(source_key)),
+                        ])
+                        .await;
+                        let (_state_guard, mut input) =
+                            source_policy_input(endpoint, SourceErrorPolicy::Propagate, None, filter_prefix).await;
+                        let first = execute_source_list(input.clone())
+                            .await
+                            .expect("valid empty pages must remain resumable");
+                        assert!(!first.headers.contains_key("x-rustfs-on-demand-migration-list"));
+                        assert!(first.output.contents.as_ref().is_none_or(Vec::is_empty));
+                        assert!(first.output.common_prefixes.as_ref().is_none_or(Vec::is_empty));
+                        assert_eq!(first.output.key_count, Some(0));
+                        assert_eq!(first.output.is_truncated, Some(true));
+                        input.continuation_token = Some(first.output.next_continuation_token.expect("empty advancing cursor"));
+
+                        let second = execute_source_list(input)
+                            .await
+                            .expect("a progressing empty chain must reach its data");
+                        assert!(!second.headers.contains_key("x-rustfs-on-demand-migration-list"));
+                        let output = second.output;
+                        let objects = output
+                            .contents
+                            .unwrap_or_default()
+                            .into_iter()
+                            .map(|object| object.key.expect("listed object key"))
+                            .collect::<Vec<_>>();
+                        let prefixes = output
+                            .common_prefixes
+                            .unwrap_or_default()
+                            .into_iter()
+                            .map(|prefix| prefix.prefix.expect("listed common prefix"))
+                            .collect::<Vec<_>>();
+                        if filter_prefix.is_some() {
+                            assert_eq!(objects, vec!["z-local"]);
+                            assert_eq!(prefixes, vec!["photos/"]);
+                        } else {
+                            assert_eq!(objects, vec!["a-source", "z-local"]);
+                            assert!(prefixes.is_empty());
+                        }
+                        assert_eq!(output.key_count, Some(2));
+                        assert_eq!(output.is_truncated, Some(false));
+                        assert!(output.next_continuation_token.is_none());
+                        let requests = tokio::time::timeout(Duration::from_secs(5), server)
+                            .await
+                            .expect("finite source connections must finish")
+                            .expect("finite source server must not panic");
+                        assert_eq!(requests.len(), 4);
+                        assert!(!requests[0].contains("continuation-token="));
+                        for (request, cursor) in requests[1..].iter().zip(["A", "B", "C"]) {
+                            assert!(request.contains(&format!("continuation-token={cursor}")), "{request}");
+                        }
+                    }
+                },
+            )
+            .await;
+        });
+    }
+
     fn assert_source_policy_result(result: S3Result<S3Response<ListObjectsV2Output>>, policy: SourceErrorPolicy) {
         match policy {
             SourceErrorPolicy::Propagate => {

rustfs/src/app/storage_api.rs

@@ -634,8 +634,8 @@ pub(crate) mod bucket {
         };
         #[cfg(test)]
         pub(crate) use crate::storage::storage_api::ecstore_bucket::on_demand_migration::{
-            BREAKER_FAILURE_THRESHOLD, BreakerState, FilterConfig, OnDemandMigrationConfig, PathStyle, Provider, SourceConfig,
-            SourceCredentials, TlsConfig,
+            BREAKER_FAILURE_THRESHOLD, BreakerState, FilterConfig, MAX_LIST_NO_PROGRESS_PAGES, OnDemandMigrationConfig,
+            PathStyle, Provider, SourceConfig, SourceCredentials, TlsConfig,
         };
         pub(crate) use crate::storage::storage_api::ecstore_bucket::on_demand_migration::{
             BucketOdmState, HeadPolicy, OdmLookup, OdmOp, OdmOutcome, OdmStateError, OnDemandMigrationSys, PolicyConfig,
@@ -643,8 +643,8 @@ pub(crate) mod bucket {
             commit_inline, idle_guarded_body,
         };
         pub(crate) use crate::storage::storage_api::ecstore_bucket::on_demand_migration::{
-            ListEntryKey, ListThroughCursor, ListThroughMerger, ListThroughToken, ListThroughTokenError, MergeSide,
-            SOURCE_LIST_MAX_RATE_WAIT, SourceListPlan, decode_continuation_token, source_list_plan,
+            ListEntryKey, ListPageError, ListThroughCursor, ListThroughMerger, ListThroughToken, ListThroughTokenError,
+            MergeSide, SOURCE_LIST_MAX_RATE_WAIT, SourceListPlan, decode_continuation_token, source_list_plan,
         };
     }