fix(odm): reject ambiguous native listing and absence evidence

rustfs/src/on_demand_migration/azure.rs

@@ -393,6 +393,9 @@ fn parse_list_blobs(xml: &str) -> Result<AzureListing, SourceError> {
         match reader.read_event() {
             Ok(Event::Start(start)) => {
                 let name = local_name(start.name().as_ref());
+                if matches!(name.as_str(), "blob" | "blobprefix") && (blob.is_some() || in_blob_prefix) {
+                    return Err(SourceError::Other("source listing entries must not be nested".to_string()));
+                }
                 match name.as_str() {
                     "blob" => {
                         depth += 1;
@@ -478,8 +481,14 @@ fn apply_list_field(
     match name {
         "name" => {
             if in_blob_prefix {
+                if blob_prefix.is_some() {
+                    return Err(SourceError::Other("source listing prefix has duplicate names".to_string()));
+                }
                 *blob_prefix = Some(text);
             } else if let Some(entry) = blob.as_mut() {
+                if entry.name.is_some() {
+                    return Err(SourceError::Other("source listing object has duplicate names".to_string()));
+                }
                 entry.name = Some(text);
             }
         }
@@ -491,6 +500,9 @@ fn apply_list_field(
         }
         "content-length" => {
             if let Some(entry) = blob.as_mut() {
+                if entry.size.is_some() {
+                    return Err(SourceError::Other("source listing object has duplicate sizes".to_string()));
+                }
                 entry.size = Some(
                     text.trim()
                         .parse()

rustfs/src/on_demand_migration/gcs.rs

@@ -280,6 +280,9 @@ struct ListedObject {
 fn parse_objects_list(body: &str) -> Result<SourcePage, SourceError> {
     let listing: ObjectsList =
         serde_json::from_str(body).map_err(|err| SourceError::Other(format!("source listing is not valid JSON: {err}")))?;
+    if listing.prefixes.iter().any(|prefix| prefix.is_empty()) {
+        return Err(SourceError::Other("source listing prefix has no name".to_string()));
+    }
     let next_continuation_token = listing.next_page_token.filter(|token| !token.is_empty());
     let objects = listing
         .items

rustfs/src/on_demand_migration/native_http.rs

@@ -177,7 +177,12 @@ impl NativeHttp {
             Some(code) => format!("source returned HTTP {status} ({code})"),
             None => format!("source returned HTTP {status}"),
         };
-        Err(classify_status(status.as_u16(), code.as_deref(), message))
+        match classify_status(status.as_u16(), code.as_deref(), message.clone()) {
+            // Native object absence needs provider-specific evidence or a
+            // successful bucket probe, never an alias from the S3 classifier.
+            SourceError::NotFound => Err(classify_status(status.as_u16(), None, message)),
+            error => Err(error),
+        }
     }
 }