fix(s3): keep ListObjects v1 local during migration (#7220)

* fix(s3): keep ListObjects v1 local during migration

* test(s3): use the v1 listing request DTO directly

rustfs/src/app/bucket_list_through.rs

@@ -451,6 +451,7 @@ mod tests {
     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 s3s::dto::ListObjectsInput;
     use std::time::Duration;
     use tokio::io::{AsyncReadExt, AsyncWriteExt};
 
@@ -795,6 +796,97 @@ mod tests {
         (result, requests)
     }
 
+    #[test]
+    #[serial_test::serial]
+    fn list_objects_v1_stays_local_with_xml_safe_key_markers() {
+        run_large_stack_test("list-through-v1-local-markers", || 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 {
+                    let (endpoint, server, stop) =
+                        list_source(std::iter::repeat(source_xml(None, false, Some("a-source")))).await;
+                    let (_state_guard, source_input) =
+                        source_policy_input(endpoint, SourceErrorPolicy::Propagate, None, None).await;
+                    let store = shared_gating_ecstore().await;
+                    store
+                        .put_object(
+                            &source_input.bucket,
+                            "a&local",
+                            &mut StoragePutObjReader::from_vec(vec![1]),
+                            &StorageObjectOptions::default(),
+                        )
+                        .await
+                        .expect("seed a second local object");
+
+                    for delimiter in [None, Some("/".to_string())] {
+                        let mut input = ListObjectsInput {
+                            bucket: source_input.bucket.clone(),
+                            max_keys: Some(1),
+                            delimiter,
+                            ..Default::default()
+                        };
+                        for (index, expected_key) in ["a&local", "z-local"].into_iter().enumerate() {
+                            let request_marker = input.marker.clone().unwrap_or_default();
+                            let response = tokio::time::timeout(
+                                Duration::from_secs(10),
+                                DefaultBucketUsecase::from_global().execute_list_objects(S3Request {
+                                    input: input.clone(),
+                                    method: http::Method::GET,
+                                    uri: http::Uri::from_static("/"),
+                                    headers: HeaderMap::new(),
+                                    extensions: http::Extensions::new(),
+                                    credentials: None,
+                                    region: None,
+                                    service: None,
+                                    trailing_headers: None,
+                                }),
+                            )
+                            .await
+                            .expect("v1 pagination must finish")
+                            .expect("list-through must not change v1 listing");
+                            assert!(!response.headers.contains_key("x-rustfs-on-demand-migration-list"));
+                            let output = response.output;
+                            let contents = output.contents.as_ref().expect("local page contents");
+                            assert_eq!(contents.len(), 1);
+                            assert_eq!(contents[0].key.as_deref(), Some(expected_key));
+                            assert_eq!(output.marker.as_deref(), Some(request_marker.as_str()));
+                            assert_eq!(output.is_truncated, Some(index == 0));
+                            assert_eq!(output.next_marker.as_deref(), (index == 0).then_some(expected_key));
+
+                            let mut xml = Vec::new();
+                            s3s::xml::Serialize::serialize(&output, &mut s3s::xml::Serializer::new(&mut xml))
+                                .expect("serialize the real v1 response");
+                            assert!(!xml.contains(&0), "XML 1.0 forbids NUL in NextMarker");
+                            let mut reader = quick_xml::Reader::from_reader(xml.as_slice());
+                            loop {
+                                if reader.read_event().expect("v1 response must be well-formed XML")
+                                    == quick_xml::events::Event::Eof
+                                {
+                                    break;
+                                }
+                            }
+                            input.marker = output.next_marker;
+                        }
+                    }
+                    stop.cancel();
+                    let requests = server.await.expect("source server must not panic");
+                    assert!(requests.is_empty(), "ListObjects v1 must issue no remote LIST requests: {requests:?}");
+                },
+            )
+            .await;
+        });
+    }
+
     #[test]
     #[serial_test::serial]
     fn list_through_invalid_source_pagination_obeys_policy_on_the_handler_path() {

rustfs/src/app/bucket_usecase.rs

@@ -2724,7 +2724,14 @@ impl DefaultBucketUsecase {
 
     #[instrument(level = "trace", skip(self, req))]
     pub async fn execute_list_objects_v2(&self, req: S3Request<ListObjectsV2Input>) -> S3Result<S3Response<ListObjectsV2Output>> {
-        // warn!("list_objects_v2 req {:?}", &req.input);
+        self.execute_list_objects_v2_inner(req, true).await
+    }
+
+    async fn execute_list_objects_v2_inner(
+        &self,
+        req: S3Request<ListObjectsV2Input>,
+        allow_list_through: bool,
+    ) -> S3Result<S3Response<ListObjectsV2Output>> {
         let ListObjectsV2Input {
             bucket,
             continuation_token,
@@ -2750,8 +2757,15 @@ impl DefaultBucketUsecase {
         // The on-demand migration envelope is decoded whether or not this
         // bucket still merges: a token handed out under `list_through` must keep
         // paginating after the policy is turned off (rustfs/backlog#2164).
-        let merged_token = list_through::decode_list_cursor(params.decoded_continuation_token.as_deref())?;
-        let (object_infos, degraded) = match list_through::list_through_state(&bucket, &req.headers) {
+        let (merged_token, source_state) = if allow_list_through {
+            (
+                list_through::decode_list_cursor(params.decoded_continuation_token.as_deref())?,
+                list_through::list_through_state(&bucket, &req.headers),
+            )
+        } else {
+            (None, None)
+        };
+        let (object_infos, degraded) = match source_state {
             Some(state) => {
                 let outcome = list_through::merged_list_objects_v2(
                     &store,
@@ -2938,7 +2952,9 @@ impl DefaultBucketUsecase {
     #[instrument(level = "debug", skip(self, req))]
     pub async fn execute_list_objects(&self, req: S3Request<ListObjectsInput>) -> S3Result<S3Response<ListObjectsOutput>> {
         let request_marker = req.input.marker.clone();
-        let v2_resp = self.execute_list_objects_v2(req.map_input(Into::into)).await?;
+        // V1 markers are object keys, so they cannot carry the opaque merged
+        // pagination state used by V2 list-through.
+        let v2_resp = self.execute_list_objects_v2_inner(req.map_input(Into::into), false).await?;
 
         Ok(v2_resp.map_output(|v2| build_list_objects_output(v2, request_marker)))
     }