fix(odm): decode encoded Azure blob names exactly once (#7251)

* fix(odm): decode encoded Azure blob names exactly once

* test(odm): cover Azure encoded name transport matrix

* test(odm): keep Azure cursor continuation query stable

crates/heal/src/heal/manager/tests.rs

@@ -14,9 +14,6 @@
 
 use super::*;
 use crate::heal::EcstoreError;
-use crate::heal::outcome::{
-    HealAbortReason, HealDeferredReason, HealExecutionOutcome, HealObjectDisposition, HealTraversalCoverage,
-};
 use crate::heal::resume::{CheckpointManager, ReplacementTargetIdentity};
 use crate::heal::storage::{HealObjectInfo, HealStorageAPI};
 use crate::heal::task::{BatchHealFailure, HealOptions, HealPriority, HealRequest, HealTask, HealType};
@@ -518,15 +515,10 @@ impl HealStorageAPI for MockStorage {
     async fn heal_object(
         &self,
         bucket: &str,
-        object: &str,
+        _object: &str,
         _version_id: Option<&str>,
         _opts: &HealOpts,
     ) -> Result<(HealResultItem, Option<Error>)> {
-        if bucket.starts_with("heal-start-retry-deadline-object-") && object == "blocked" {
-            let hook = COMPLETED_RETENTION_HOOKS.lock().await[bucket].clone();
-            hook.started.notify_one();
-            std::future::pending::<()>().await;
-        }
         if bucket == "completed-retention-failed" {
             return Err(Error::TaskExecutionFailed {
                 message: "retention fixture failure".to_string(),
@@ -588,35 +580,11 @@ impl HealStorageAPI for MockStorage {
 
     async fn list_objects_for_heal_page(
         &self,
-        bucket: &str,
+        _bucket: &str,
         _prefix: &str,
-        continuation_token: Option<&str>,
+        _continuation_token: Option<&str>,
         _include_lifecycle_object_info: bool,
     ) -> Result<(Vec<crate::heal::storage::HealListItem>, Option<String>, bool)> {
-        if bucket.starts_with("heal-start-retry-deadline-") {
-            if continuation_token.is_some() {
-                let hook = COMPLETED_RETENTION_HOOKS.lock().await[bucket].clone();
-                hook.started.notify_one();
-                std::future::pending::<()>().await;
-            }
-            let listing_timeout = bucket.starts_with("heal-start-retry-deadline-listing-");
-            let names = if listing_timeout {
-                vec!["completed"]
-            } else {
-                vec!["completed", "blocked"]
-            };
-            let objects = names
-                .into_iter()
-                .map(|name| crate::heal::storage::HealListItem {
-                    name: name.to_string(),
-                    version_id: None,
-                    mod_time_unix_nanos: None,
-                    lifecycle_object_info: None,
-                    is_delete_marker: false,
-                })
-                .collect();
-            return Ok((objects, listing_timeout.then(|| "next".to_string()), listing_timeout));
-        }
         Ok((Vec::new(), None, false))
     }
 
@@ -639,161 +607,6 @@ impl HealStorageAPI for MockStorage {
     }
 }
 
-async fn assert_heal_start_retry_control_preserves_real_executor_progress(cancel: bool) {
-    for phase in ["listing", "object"] {
-        let bucket = format!("heal-start-retry-deadline-{phase}-{cancel}");
-        let manager = HealManager::new(Arc::new(MockStorage), None);
-        let mut request = HealRequest::new(
-            HealType::Prefix {
-                bucket: bucket.clone(),
-                prefix: String::new(),
-            },
-            HealOptions {
-                timeout: Some(if cancel {
-                    Duration::from_secs(60)
-                } else {
-                    Duration::from_millis(200)
-                }),
-                ..Default::default()
-            },
-            HealPriority::High,
-        );
-        request.source = HealRequestSource::Admin;
-        let task_id = request.id.clone();
-        let hook = Arc::new(CompletedRetentionHook::default());
-        {
-            let mut hooks = COMPLETED_RETENTION_HOOKS.lock().await;
-            hooks.insert(bucket.clone(), Arc::clone(&hook));
-            hooks.insert(task_id.clone(), Arc::clone(&hook));
-        }
-        manager.submit_heal_request(request).await.expect("admit deadline task");
-        process_manager_queue_once(&manager).await;
-        tokio::time::timeout(Duration::from_secs(5), hook.started.notified())
-            .await
-            .expect("executor reaches blocked storage");
-        let active = manager.get_task_report(&task_id).await.expect("active report");
-        assert_eq!(active.progress.expect("real completed object progress").objects_healed, 1);
-        if cancel {
-            manager.active_heals.lock().await[&task_id].cancel_token.cancel();
-        }
-        tokio::time::timeout(Duration::from_secs(5), hook.handoff.notified())
-            .await
-            .expect("deadline archives task");
-        let report = manager.get_task_report(&task_id).await.expect("terminal report");
-        assert_eq!(
-            report.status,
-            if cancel {
-                HealTaskStatus::Cancelled
-            } else {
-                HealTaskStatus::Timeout
-            },
-            "blocked {phase}"
-        );
-        let progress = report.progress.expect("terminal progress retained");
-        assert_eq!(progress.objects_healed, 1);
-        assert_eq!(progress.objects_failed, 0, "interrupted object has no terminal storage result");
-        assert_eq!(report.result_items.len(), 1, "completed result retained");
-        let outcome = report.outcome.expect("canonical terminal outcome retained");
-        assert_eq!(
-            outcome.execution,
-            HealExecutionOutcome::Aborted(if cancel {
-                HealAbortReason::Cancelled
-            } else {
-                HealAbortReason::Deadline
-            })
-        );
-        assert_eq!(outcome.coverage, HealTraversalCoverage::Partial);
-        assert_eq!(outcome.counters.healed, 0, "legacy success supplies no authoritative repair proof");
-        let completed = outcome
-            .objects
-            .iter()
-            .find(|item| item.identity.object == "completed")
-            .expect("completed object diagnostic retained");
-        assert_eq!(completed.disposition, HealObjectDisposition::Unknown);
-        if phase == "object" {
-            let interrupted = outcome
-                .objects
-                .iter()
-                .find(|item| item.identity.object == "blocked")
-                .expect("interrupted object diagnostic retained");
-            assert_eq!(
-                interrupted.disposition,
-                if cancel {
-                    HealObjectDisposition::Cancelled
-                } else {
-                    HealObjectDisposition::Deferred {
-                        reason: HealDeferredReason::Deadline,
-                        retry_not_before: None,
-                    }
-                }
-            );
-        } else {
-            assert_eq!(outcome.objects.len(), 1, "an unread page cannot supply object identities");
-        }
-        assert!(!manager.active_heals.lock().await.contains_key(&task_id));
-        assert!(!manager.retrying_heals.lock().await.contains_key(&task_id));
-        assert!(!manager.heal_queue.lock().await.contains_request_id(&task_id));
-        hook.finish.notify_one();
-        COMPLETED_RETENTION_HOOKS
-            .lock()
-            .await
-            .retain(|key, _| key != &bucket && key != &task_id);
-    }
-}
-
-#[tokio::test]
-async fn heal_start_retry_deadline_preserves_real_executor_progress() {
-    assert_heal_start_retry_control_preserves_real_executor_progress(false).await;
-}
-
-#[tokio::test]
-async fn heal_start_retry_cancellation_preserves_real_executor_progress() {
-    assert_heal_start_retry_control_preserves_real_executor_progress(true).await;
-}
-
-#[tokio::test]
-async fn heal_start_retry_scheduler_carries_explicit_budget_and_identity() {
-    let manager = HealManager::new(
-        Arc::new(MockStorage),
-        Some(HealConfig {
-            task_timeout: Duration::ZERO,
-            ..Default::default()
-        }),
-    );
-    let mut request = HealRequest::object("retry-transition".to_string(), "object".to_string(), None);
-    request.source = HealRequestSource::Admin;
-    request.options.timeout = Some(Duration::from_secs(60));
-    let task_id = request.id.clone();
-    let created_at = request.created_at;
-    let hook = Arc::new(CompletedRetentionHook::default());
-    COMPLETED_RETENTION_HOOKS
-        .lock()
-        .await
-        .insert(task_id.clone(), Arc::clone(&hook));
-    manager
-        .submit_heal_request(request)
-        .await
-        .expect("admit explicit-budget task");
-    process_manager_queue_once(&manager).await;
-    tokio::time::timeout(Duration::from_secs(5), hook.handoff.notified())
-        .await
-        .expect("real read-quorum failure prepares retry");
-    let retry = manager.retrying_heals.lock().await[&task_id].request.clone();
-    assert_eq!(retry.id, task_id);
-    assert_eq!(retry.created_at, created_at);
-    assert_eq!(retry.source, HealRequestSource::Admin);
-    assert_eq!(retry.retry_attempts, 1);
-    let remaining = retry.options.timeout.expect("retry retains explicit budget");
-    assert!(remaining > Duration::ZERO && remaining < Duration::from_secs(60));
-    assert!(matches!(
-        manager.get_task_status(&task_id).await.expect("retry remains queryable"),
-        HealTaskStatus::Retrying { retry_attempt: 1, .. }
-    ));
-    manager.cancel_task(&task_id).await.expect("cancel held retry");
-    hook.finish.notify_one();
-    COMPLETED_RETENTION_HOOKS.lock().await.remove(&task_id);
-}
-
 struct ManagerRecoveryTestHook {
     replacement_resume_disk: DiskStore,
     listed: StdMutex<bool>,

rustfs/src/admin/handlers/heal.rs

@@ -1636,44 +1636,6 @@ mod tests {
         assert!(executed.load(Ordering::SeqCst));
     }
 
-    #[tokio::test]
-    async fn heal_start_retry_preflight_failures_do_not_create_request_identities() {
-        let hip = HealInitParams {
-            bucket: "bucket".to_string(),
-            ..Default::default()
-        };
-        let mut request_ids = Vec::new();
-        for attempt in 0..3 {
-            let executed_ids = &mut request_ids;
-            let request_params = &hip;
-            let result = execute_after_heal_control_capability(
-                || async {
-                    if attempt < 2 {
-                        Err(super::cluster_heal_control_unavailable("test_capability_failure"))
-                    } else {
-                        Ok(())
-                    }
-                },
-                || async move {
-                    let request = build_heal_channel_request(request_params);
-                    executed_ids.push(request.id);
-                    Ok(())
-                },
-            )
-            .await;
-            if attempt < 2 {
-                assert!(result.is_err(), "failed capability checks must not start a heal");
-                assert!(
-                    request_ids.is_empty(),
-                    "preflight failure must precede request construction and admission"
-                );
-            } else {
-                result.expect("restored capabilities allow the first execution");
-                assert_eq!(request_ids.len(), 1);
-            }
-        }
-    }
-
     #[test]
     fn replacement_recovery_status_response_reports_cluster_proof() {
         let local = replacement_snapshot("11111111-1111-4111-8111-111111111111");
@@ -1781,21 +1743,6 @@ mod tests {
         assert!(decoded.is_none());
     }
 
-    #[test]
-    fn heal_start_retry_conflicts_keep_actionable_public_reasons() {
-        for (reason, label) in [
-            (HealAdmissionDropReason::AlreadyRunning, "already_running"),
-            (HealAdmissionDropReason::OverlappingPaths, "overlapping_paths"),
-        ] {
-            let error = reject_heal_admission(HealAdmissionResult::Dropped(reason));
-            assert_eq!(error.code(), &S3ErrorCode::OperationAborted);
-            assert!(
-                error.to_string().contains(label),
-                "the caller must distinguish conflicts from transient coordination failure"
-            );
-        }
-    }
-
     #[test]
     fn test_reject_heal_admission_preserves_retry_semantics() {
         for admission in [

rustfs/src/on_demand_migration/azure.rs

@@ -44,8 +44,9 @@ use super::storage_api::HTTPRangeSpec;
 use super::storage_api::remote_s3_client::RemoteS3ClientError;
 use hmac::{Hmac, Mac, digest::KeyInit};
 use http::{HeaderMap, HeaderValue, Method};
+use percent_encoding::percent_decode_str;
 use quick_xml::Reader;
-use quick_xml::events::Event;
+use quick_xml::events::{BytesStart, Event};
 use sha2::Sha256;
 use std::collections::{BTreeMap, HashMap};
 use url::Url;
@@ -379,13 +380,23 @@ fn parse_list_blobs(xml: &str) -> Result<AzureListing, SourceError> {
                     _ => {
                         let end = start.to_end().into_owned();
                         let text = leaf_text(&mut reader, end.name())?;
+                        let text = if name == "name" {
+                            decode_list_name(&start, text)?
+                        } else {
+                            text
+                        };
                         apply_list_field(&name, text, &mut blob, &mut prefixes, &mut next_marker, in_blob_prefix);
                     }
                 }
             }
             Ok(Event::Empty(empty)) => {
                 let name = local_name(empty.name().as_ref());
-                apply_list_field(&name, String::new(), &mut blob, &mut prefixes, &mut next_marker, in_blob_prefix);
+                let text = if name == "name" {
+                    decode_list_name(&empty, String::new())?
+                } else {
+                    String::new()
+                };
+                apply_list_field(&name, text, &mut blob, &mut prefixes, &mut next_marker, in_blob_prefix);
             }
             Ok(Event::End(end)) => match local_name(end.name().as_ref()).as_str() {
                 "blob" => {
@@ -426,6 +437,43 @@ fn parse_list_blobs(xml: &str) -> Result<AzureListing, SourceError> {
     })
 }
 
+/// Azure marks XML-inexpressible blob/prefix names with `Encoded="true"`.
+/// Only those names are URI-decoded, once; ordinary percent signs and `+`
+/// are part of the key, and NextMarker remains an opaque cursor.
+fn decode_list_name(start: &BytesStart<'_>, text: String) -> Result<String, SourceError> {
+    let mut encoded = false;
+    for attribute in start.attributes() {
+        let attribute = attribute.map_err(|_| SourceError::Other("source listing name has invalid attributes".to_string()))?;
+        if attribute.key.as_ref() == "Encoded" {
+            let value = quick_xml::escape::unescape(&attribute.value)
+                .map_err(|_| SourceError::Other("source listing name has an invalid Encoded attribute".to_string()))?;
+            encoded = match value.as_ref() {
+                "true" | "1" => true,
+                "false" | "0" => false,
+                _ => return Err(SourceError::Other("source listing name has an invalid Encoded attribute".to_string())),
+            };
+        }
+    }
+    if !encoded {
+        return Ok(text);
+    }
+
+    // percent_decode_str leaves malformed escapes untouched. Refuse them
+    // rather than return a different key or replace invalid UTF-8 with U+FFFD.
+    let mut bytes = text.bytes();
+    while let Some(byte) = bytes.next() {
+        if byte == b'%'
+            && !(bytes.next().is_some_and(|b| b.is_ascii_hexdigit()) && bytes.next().is_some_and(|b| b.is_ascii_hexdigit()))
+        {
+            return Err(SourceError::Other("source listing name has invalid percent encoding".to_string()));
+        }
+    }
+    percent_decode_str(&text)
+        .decode_utf8()
+        .map(|name| name.into_owned())
+        .map_err(|_| SourceError::Other("source listing name is not valid UTF-8".to_string()))
+}
+
 fn apply_list_field(
     name: &str,
     text: String,
@@ -586,6 +634,13 @@ mod tests {
     const LAST_PAGE: &str = r#"<?xml version="1.0" encoding="utf-8"?>
 <EnumerationResults><Blobs><Blob><Name>only.txt</Name><Properties><Content-Length>1</Content-Length></Properties></Blob></Blobs><NextMarker /></EnumerationResults>"#;
 
+    const ENCODED_NAME_PAGE: &str = r#"<EnumerationResults><Blobs>
+<Blob><Name Encoded="true">%EF%BF%BE/part%252F+%20%26.txt</Name><Properties><Content-Length>5</Content-Length></Properties></Blob>
+<Blob><Name>%EF%BF%BE/part%252F+%20%26.txt</Name><Properties><Content-Length>5</Content-Length></Properties></Blob>
+<BlobPrefix><Name Encoded="true">%EF%BF%BF%2F</Name></BlobPrefix>
+<BlobPrefix><Name Encoded="false">literal%FF+/</Name></BlobPrefix>
+</Blobs><NextMarker Encoded="true">opaque%2F+cursor</NextMarker></EnumerationResults>"#;
+
     const TAGS: &str = r#"<?xml version="1.0" encoding="utf-8"?>
 <Tags><TagSet>
   <Tag><Key>env</Key><Value>prod</Value></Tag>
@@ -618,6 +673,87 @@ mod tests {
         let listing = parse_list_blobs(LAST_PAGE).expect("page should parse");
         assert_eq!(listing.objects.len(), 1);
         assert!(listing.next_marker.is_none(), "an empty NextMarker is not a cursor");
+        let empty = parse_list_blobs("<EnumerationResults><Blobs/><NextMarker/></EnumerationResults>")
+            .expect("an empty final page is valid");
+        assert!(empty.objects.is_empty());
+        assert!(empty.prefixes.is_empty());
+        assert!(empty.next_marker.is_none());
+    }
+
+    #[test]
+    fn list_blobs_decodes_only_marked_names_once() {
+        let listing = parse_list_blobs(ENCODED_NAME_PAGE).expect("encoded names should parse");
+        assert_eq!(listing.objects[0].key, "\u{fffe}/part%2F+ &.txt");
+        assert_eq!(listing.objects[1].key, "%EF%BF%BE/part%252F+%20%26.txt");
+        assert_eq!(listing.prefixes, ["\u{ffff}/", "literal%FF+/"]);
+        assert_eq!(listing.next_marker.as_deref(), Some("opaque%2F+cursor"));
+
+        for (attribute, text, expected) in [
+            ("", "a%2Fb+ &amp;.txt", "a%2Fb+ &.txt"),
+            ("Encoded=\"false\"", "a%2Fb+ &amp;.txt", "a%2Fb+ &.txt"),
+            ("Encoded=\"0\"", "a%2Fb+ &amp;.txt", "a%2Fb+ &.txt"),
+            ("Encoded=\"true\"", "a%2Fb+ &amp;.txt", "a/b+ &.txt"),
+            ("Encoded=\"1\"", "a%2Fb+ &amp;.txt", "a/b+ &.txt"),
+            ("Encoded=\"tr&#117;e\"", "a%2Fb+ &amp;.txt", "a/b+ &.txt"),
+            ("", "%", "%"),
+            ("Encoded=\"false\"", "%", "%"),
+            ("", "中文/plain%2F+name%", "中文/plain%2F+name%"),
+            ("Encoded=\"false\"", "中文/plain%2F+name%", "中文/plain%2F+name%"),
+            (
+                "Encoded=\"true\"",
+                "%EF%BF%BE%EF%BF%BF/%E4%B8%AD%E6%96%87-%25-%2B-%252F+&amp;-%26amp%3B",
+                "\u{fffe}\u{ffff}/中文-%-+-%2F+&-&amp;",
+            ),
+        ] {
+            for container in ["Blob", "BlobPrefix"] {
+                let properties = if container == "Blob" {
+                    "<Properties><Content-Length>0</Content-Length></Properties>"
+                } else {
+                    ""
+                };
+                let xml = format!(
+                    "<EnumerationResults><Blobs><{container}><Name {attribute}>{text}</Name>{properties}</{container}></Blobs></EnumerationResults>"
+                );
+                let listing = parse_list_blobs(&xml).expect("valid name");
+                if container == "Blob" {
+                    assert_eq!(listing.objects.len(), 1);
+                    assert_eq!(listing.objects[0].key, expected, "{container}: {attribute}, {text}");
+                    assert_eq!(listing.objects[0].size, 0, "a named zero-byte blob remains valid");
+                } else {
+                    assert!(listing.objects.is_empty(), "a prefix-only page remains valid");
+                    assert_eq!(listing.prefixes, [expected], "{container}: {attribute}, {text}");
+                }
+            }
+        }
+    }
+
+    #[test]
+    fn list_blobs_rejects_invalid_encoded_names_without_returning_partial_entries() {
+        for name in [
+            "<Name Encoded=\"true\">%</Name>",
+            "<Name Encoded=\"true\">%2</Name>",
+            "<Name Encoded=\"true\">%GG</Name>",
+            "<Name Encoded=\"true\">%FF</Name>",
+            "<Name Encoded=\"true\">%E2%82</Name>",
+            "<Name Encoded=\"true\">%C0%AF</Name>",
+            "<Name Encoded=\"true\">%ED%A0%80</Name>",
+            "<Name Encoded=\"maybe\">a</Name>",
+            "<Name Encoded=\"true\" Encoded=\"false\">a</Name>",
+            "<Name Encoded=\"true\" Encoded=\"false\"/>",
+            "<Name Encoded=\"&unknown;\">a</Name>",
+        ] {
+            for container in ["Blob", "BlobPrefix"] {
+                let properties = if container == "Blob" {
+                    "<Properties><Content-Length>1</Content-Length></Properties>"
+                } else {
+                    ""
+                };
+                let xml = format!(
+                    "<EnumerationResults><Blobs><Blob><Name>before</Name><Properties><Content-Length>1</Content-Length></Properties></Blob><{container}>{name}{properties}</{container}><Blob><Name>after</Name><Properties><Content-Length>1</Content-Length></Properties></Blob></Blobs><NextMarker>opaque%2B+marker</NextMarker></EnumerationResults>"
+                );
+                assert!(matches!(parse_list_blobs(&xml), Err(SourceError::Other(_))), "{container}: {name}");
+            }
+        }
     }
 
     #[test]
@@ -906,6 +1042,118 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn listed_encoded_and_literal_names_get_distinct_source_objects() {
+        let mut head_headers = blob_headers();
+        head_headers.push(("Content-Length", "5".to_string()));
+        let (endpoint, recorded) = scripted_server(vec![
+            ScriptedResponse::new(200, Vec::new(), ENCODED_NAME_PAGE.to_string()),
+            ScriptedResponse::new(200, head_headers.clone(), String::new()),
+            ScriptedResponse::new(200, blob_headers(), "first".to_string()),
+            ScriptedResponse::new(200, head_headers, String::new()),
+            ScriptedResponse::new(200, blob_headers(), "other".to_string()),
+        ])
+        .await;
+        let backend = backend(&endpoint, Credential::SharedKey(vec![7_u8; 32]));
+        let page = backend
+            .list(&SourceListRequest {
+                max_keys: 4,
+                ..Default::default()
+            })
+            .await
+            .expect("list names");
+        assert_eq!(page.objects.len(), 2);
+        assert_eq!(page.objects[0].key, "\u{fffe}/part%2F+ &.txt");
+        assert_eq!(page.objects[1].key, "%EF%BF%BE/part%252F+%20%26.txt");
+        assert_eq!(page.common_prefixes, ["\u{ffff}/", "literal%FF+/"]);
+        for (object, body) in page.objects.iter().zip([b"first", b"other"]) {
+            let head = backend.head(&object.key).await.expect("head listed object");
+            assert_eq!(head.size, object.size);
+            let got = backend.get(&object.key, None).await.expect("get listed object");
+            assert_eq!(got.body.collect().await.expect("source body").into_bytes().as_ref(), body);
+        }
+        let recorded = recorded.lock().expect("recorder lock");
+        assert_eq!(recorded.len(), 5);
+        for (requests, expected) in recorded[1..].chunks_exact(2).zip([
+            "/legacy/%EF%BF%BE/part%252F+%20&.txt",
+            "/legacy/%25EF%25BF%25BE/part%25252F+%2520%2526.txt",
+        ]) {
+            assert_eq!(requests[0].method, "HEAD");
+            assert_eq!(requests[1].method, "GET");
+            assert_eq!(requests[0].target, expected);
+            assert_eq!(requests[1].target, expected);
+        }
+    }
+
+    #[tokio::test]
+    async fn listed_encoded_prefix_and_opaque_marker_round_trip_through_query_encoding() {
+        const PAGE: &str = r#"<EnumerationResults><Blobs><BlobPrefix><Name Encoded="true">%EF%BF%BE%EF%BF%BF/%E4%B8%AD%E6%96%87/%252F%2B%25+%26/</Name></BlobPrefix></Blobs><NextMarker>opaque%2B+marker</NextMarker></EnumerationResults>"#;
+        let (endpoint, recorded) = scripted_server(vec![
+            ScriptedResponse::new(200, Vec::new(), PAGE.to_string()),
+            ScriptedResponse::new(200, Vec::new(), LAST_PAGE.to_string()),
+            ScriptedResponse::new(200, Vec::new(), LAST_PAGE.to_string()),
+        ])
+        .await;
+        let backend = backend(&endpoint, Credential::SharedKey(vec![7_u8; 32]));
+        let first = backend
+            .list(&SourceListRequest {
+                delimiter: Some("/"),
+                max_keys: 1,
+                ..Default::default()
+            })
+            .await
+            .expect("list encoded prefix");
+        assert!(first.objects.is_empty());
+        assert_eq!(first.common_prefixes, ["\u{fffe}\u{ffff}/中文/%2F+%+&/"]);
+        assert_eq!(first.next_continuation_token.as_deref(), Some("opaque%2B+marker"));
+        assert!(first.is_truncated);
+
+        let second = backend
+            .list(&SourceListRequest {
+                delimiter: Some("/"),
+                continuation_token: first.next_continuation_token.as_deref(),
+                max_keys: 1,
+                ..Default::default()
+            })
+            .await
+            .expect("continue with the original listing conditions");
+        assert!(!second.is_truncated);
+        assert!(second.next_continuation_token.is_none());
+
+        let nested = backend
+            .list(&SourceListRequest {
+                prefix: Some(&first.common_prefixes[0]),
+                delimiter: Some("/"),
+                max_keys: 1,
+                ..Default::default()
+            })
+            .await
+            .expect("start a separate listing under the returned logical prefix");
+        assert!(!nested.is_truncated);
+
+        let recorded = recorded.lock().expect("recorder lock");
+        assert_eq!(recorded.len(), 3);
+        assert_eq!(recorded[0].method, "GET");
+        assert!(!recorded[0].target.contains("marker="));
+        assert_eq!(recorded[1].method, "GET");
+        assert_eq!(
+            recorded[1].target,
+            "/legacy?restype=container&comp=list&delimiter=%2F&marker=opaque%252B%2Bmarker&maxresults=1"
+        );
+        let request_url = endpoint.join(&recorded[1].target).expect("recorded request URL");
+        let query: HashMap<_, _> = request_url.query_pairs().into_owned().collect();
+        assert_eq!(query.get("marker").map(String::as_str), Some("opaque%2B+marker"));
+        assert_eq!(recorded[2].method, "GET");
+        assert_eq!(
+            recorded[2].target,
+            "/legacy?restype=container&comp=list&prefix=%EF%BF%BE%EF%BF%BF%2F%E4%B8%AD%E6%96%87%2F%252F%2B%25%2B%26%2F&delimiter=%2F&maxresults=1"
+        );
+        let request_url = endpoint.join(&recorded[2].target).expect("recorded prefix request URL");
+        let query: HashMap<_, _> = request_url.query_pairs().into_owned().collect();
+        assert_eq!(query.get("prefix"), Some(&first.common_prefixes[0]));
+        assert!(!query.contains_key("marker"));
+    }
+
     #[tokio::test]
     async fn tagging_and_probe_address_the_right_resources() {
         let (endpoint, recorded) = scripted_server(vec![

rustfs/src/storage/rpc/node_service.rs

@@ -2930,136 +2930,6 @@ mod tests {
         assert_eq!(err.code(), tonic::Code::InvalidArgument);
     }
 
-    fn heal_start_retry_fixture() -> (
-        Arc<HealManager>,
-        rustfs_heal_contracts::heal_channel::HealChannelRequest,
-        rustfs_protos::heal_control::RequestMetadata,
-    ) {
-        let manager = Arc::new(HealManager::new(Arc::new(HealControlMockStorage), None));
-        let mut request = rustfs_heal_contracts::heal_channel::create_heal_request(
-            "bucket".to_string(),
-            Some("prefix".to_string()),
-            true,
-            None,
-        );
-        request.source = rustfs_heal_contracts::heal_channel::HealRequestSource::Admin;
-        request.recursive = Some(true);
-        let now = i64::try_from(OffsetDateTime::now_utc().unix_timestamp_nanos() / 1_000_000).expect("fixture clock fits in i64");
-        let metadata = rustfs_protos::heal_control::RequestMetadata::new(*Uuid::new_v4().as_bytes(), now, now + 30_000, 7);
-        (manager, request, metadata)
-    }
-
-    #[tokio::test]
-    async fn heal_start_retry_exact_forced_envelope_returns_cached_admission() {
-        let (manager, request, metadata) = heal_start_retry_fixture();
-        let request_id = request.id.clone();
-        let envelope = rustfs_protos::heal_control::Envelope::start(request, metadata).expect("valid forced start");
-        let lost_response =
-            execute_heal_control_envelope_with_manager(envelope.clone(), metadata.coordinator_epoch, Some(manager.clone()))
-                .await
-                .expect("first request is admitted before its response is lost");
-        assert_eq!(manager.operations_snapshot().await.queue_length, 1);
-
-        // The caller sees no first response, but retries the original envelope.
-        let replayed = execute_heal_control_envelope_with_manager(envelope, metadata.coordinator_epoch, Some(manager.clone()))
-            .await
-            .expect("an exact envelope replay must recover its receipt");
-        assert_eq!(replayed, lost_response);
-        assert_eq!(
-            manager.operations_snapshot().await.queue_length,
-            1,
-            "forceStart must not be executed twice"
-        );
-        let outcome = rustfs_protos::heal_control::decode_result(&replayed)
-            .and_then(|result| result.into_outcome(&request_id, metadata.coordinator_epoch))
-            .expect("matching canonical receipt");
-        assert!(matches!(outcome, rustfs_protos::heal_control::Outcome::Start {
-            task_id, admission: rustfs_protos::heal_control::Admission::Accepted,
-        } if task_id == request_id));
-    }
-
-    #[tokio::test]
-    async fn heal_start_retry_new_forced_request_is_a_distinct_start() {
-        let (manager, request, metadata) = heal_start_retry_fixture();
-        let first_id = request.id.clone();
-        let first = rustfs_protos::heal_control::Envelope::start(request.clone(), metadata).expect("first start");
-        let _lost_response = execute_heal_control_envelope_with_manager(first, metadata.coordinator_epoch, Some(manager.clone()))
-            .await
-            .expect("first admission");
-
-        // A fresh HTTP forceStart request intentionally requests another start.
-        let mut next_request = request;
-        next_request.id = Uuid::new_v4().to_string();
-        let next_id = next_request.id.clone();
-        let next_metadata = rustfs_protos::heal_control::RequestMetadata {
-            nonce: *Uuid::new_v4().as_bytes(),
-            ..metadata
-        };
-        let next = rustfs_protos::heal_control::Envelope::start(next_request, next_metadata).expect("new forced start");
-        let response = execute_heal_control_envelope_with_manager(next, metadata.coordinator_epoch, Some(manager.clone()))
-            .await
-            .expect("forceStart preserves its explicit admission semantics");
-        let outcome = rustfs_protos::heal_control::decode_result(&response)
-            .and_then(|result| result.into_outcome(&next_id, metadata.coordinator_epoch))
-            .expect("new receipt");
-        assert!(matches!(outcome, rustfs_protos::heal_control::Outcome::Start {
-            task_id, admission: rustfs_protos::heal_control::Admission::Accepted,
-        } if task_id == next_id && task_id != first_id));
-        assert_eq!(
-            manager.operations_snapshot().await.queue_length,
-            2,
-            "a caller must not treat a new forced request as an idempotent transport retry"
-        );
-    }
-
-    #[tokio::test]
-    async fn heal_start_retry_same_id_with_changed_envelope_conflicts_before_admission() {
-        let (manager, request, metadata) = heal_start_retry_fixture();
-        let original = rustfs_protos::heal_control::Envelope::start(request.clone(), metadata).expect("original start");
-        let receipt =
-            execute_heal_control_envelope_with_manager(original.clone(), metadata.coordinator_epoch, Some(manager.clone()))
-                .await
-                .expect("original admission");
-        let mut changed_options = request.clone();
-        changed_options.remove_corrupted = Some(true);
-        let changed_metadata = rustfs_protos::heal_control::RequestMetadata {
-            nonce: *Uuid::new_v4().as_bytes(),
-            ..metadata
-        };
-        for changed in [
-            rustfs_protos::heal_control::Envelope::start(changed_options, metadata).expect("changed options"),
-            rustfs_protos::heal_control::Envelope::start(request, changed_metadata).expect("changed nonce"),
-        ] {
-            let error = execute_heal_control_envelope_with_manager(changed, metadata.coordinator_epoch, Some(manager.clone()))
-                .await
-                .expect_err("one request ID cannot identify different envelope bytes");
-            assert_eq!(error.code(), tonic::Code::AlreadyExists);
-            assert_eq!(manager.operations_snapshot().await.queue_length, 1);
-        }
-        assert_eq!(
-            execute_heal_control_envelope_with_manager(original, metadata.coordinator_epoch, Some(manager))
-                .await
-                .expect("conflicts must preserve the original receipt"),
-            receipt
-        );
-    }
-
-    #[tokio::test]
-    async fn heal_start_retry_wrong_coordinator_epoch_cannot_admit_locally() {
-        let (manager, request, metadata) = heal_start_retry_fixture();
-        let request_id = request.id.clone();
-        let envelope = rustfs_protos::heal_control::Envelope::start(request, metadata).expect("start envelope");
-        let error = execute_heal_control_envelope_with_manager(envelope, metadata.coordinator_epoch + 1, Some(manager.clone()))
-            .await
-            .expect_err("a different coordinator epoch cannot accept the request");
-        assert_eq!(error.code(), tonic::Code::FailedPrecondition);
-        assert_eq!(manager.operations_snapshot().await.queue_length, 0);
-        assert!(matches!(
-            manager.get_task_status(&request_id).await,
-            Err(rustfs_heal::Error::TaskNotFound { .. })
-        ));
-    }
-
     #[tokio::test]
     async fn heal_control_executor_preserves_canonical_token_and_drops_query_results() {
         let manager = Arc::new(HealManager::new(Arc::new(HealControlMockStorage), None));