fix(s3): preserve delete marker read error headers

crates/ecstore/src/store/object.rs

@@ -4251,8 +4251,10 @@ impl ECStore {
     }
 
     /// Return metadata for DELETE preflight, including an explicitly addressed
-    /// delete marker. Read APIs must keep using `get_object_info`; authorization
-    /// and Object Lock enforcement still belong to the caller and locked delete.
+    /// delete marker. GET/HEAD may also use this metadata-only lookup to enrich
+    /// an already failed read with marker headers, never to serve marker data.
+    /// Normal reads must keep using `get_object_info`; authorization and Object
+    /// Lock enforcement still belong to the caller and locked delete.
     #[instrument(level = "trace", skip_all)]
     pub async fn get_object_info_for_delete(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<ObjectInfo> {
         self.get_object_info_snapshot(bucket, object, opts, true).await

rustfs/src/app/object/delete.rs

@@ -1211,6 +1211,215 @@ mod tests {
     };
     use std::sync::Arc;
 
+    #[test]
+    #[serial_test::serial]
+    fn execute_read_delete_marker_headers_in_single_and_multi_pool() {
+        crate::app::gating_test_env::run_large_stack_test("read-marker-headers", || async {
+            use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions};
+
+            let single_pool = crate::app::gating_test_env::shared_gating_ecstore().await;
+            if current_app_context().is_none() {
+                crate::app::runtime_sources::install_test_app_context(Arc::clone(&single_pool)).await;
+            }
+            let ambient = current_app_context().expect("read API test context");
+            let (_temp_dir, _disk_paths, multi_pool) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
+            for (pool_count, store) in [(1, single_pool), (2, multi_pool)] {
+                let context = Arc::new(AppContext::new(Arc::clone(&store), ambient.iam(), ambient.kms()));
+                let usecase = DefaultObjectUsecase::with_context(Some(context));
+                for suspended in [false, true] {
+                    let bucket = format!("read-marker-headers-{pool_count}-{}", Uuid::new_v4());
+                    store
+                        .make_bucket(
+                            &bucket,
+                            &MakeBucketOptions {
+                                versioning_enabled: true,
+                                ..Default::default()
+                            },
+                        )
+                        .await
+                        .expect("create versioned read fixture");
+                    let key = "history";
+                    let payload = b"historical payload behind the marker";
+                    let original = store
+                        .put_object(
+                            &bucket,
+                            key,
+                            &mut PutObjReader::from_vec(payload.to_vec()),
+                            &ObjectOptions {
+                                versioned: true,
+                                ..Default::default()
+                            },
+                        )
+                        .await
+                        .expect("write historical version");
+                    if suspended {
+                        store
+                            .update_bucket_metadata_config(
+                                &bucket,
+                                crate::app::storage_api::test::bucket::metadata::BUCKET_VERSIONING_CONFIG,
+                                b"<VersioningConfiguration><Status>Suspended</Status></VersioningConfiguration>".to_vec(),
+                            )
+                            .await
+                            .expect("suspend versioning before creating the null marker");
+                    }
+                    // get_opts still reads versioning through the ambient metadata
+                    // facade. Publish the fixture's actual config there while the
+                    // request-bound context keeps object I/O on the selected store.
+                    let metadata = store.get_bucket_metadata(&bucket).await.expect("fixture bucket metadata");
+                    crate::app::storage_api::test::set_bucket_metadata(bucket.clone(), (*metadata).clone())
+                        .await
+                        .expect("publish fixture versioning config");
+                    let read_opts = get_opts(&bucket, key, None, None, &HeaderMap::new())
+                        .await
+                        .expect("read fixture versioning config");
+                    assert_eq!(read_opts.versioned, !suspended);
+                    assert_eq!(read_opts.version_suspended, suspended);
+                    let marker = store
+                        .delete_object(
+                            &bucket,
+                            key,
+                            ObjectOptions {
+                                versioned: !suspended,
+                                version_suspended: suspended,
+                                ..Default::default()
+                            },
+                        )
+                        .await
+                        .expect("create latest delete marker");
+                    let marker_id = delete_response_version_id(marker.version_id, false).expect("marker identity");
+                    assert_eq!(marker_id == "null", suspended);
+
+                    for explicit in [false, true] {
+                        let requested_version = explicit.then(|| marker_id.clone());
+                        let get = GetObjectInput::builder()
+                            .bucket(bucket.clone())
+                            .key(key.to_string())
+                            .version_id(requested_version.clone())
+                            .build()
+                            .expect("GET request");
+                        let head = HeadObjectInput::builder()
+                            .bucket(bucket.clone())
+                            .key(key.to_string())
+                            .version_id(requested_version)
+                            .build()
+                            .expect("HEAD request");
+                        let get_error = usecase
+                            .execute_get_object(build_request(get, Method::GET))
+                            .await
+                            .expect_err("marker GET");
+                        let head_error = usecase
+                            .execute_head_object(build_request(head, Method::HEAD))
+                            .await
+                            .expect_err("marker HEAD");
+                        for (method, error) in [("GET", get_error), ("HEAD", head_error)] {
+                            assert_eq!(
+                                error.code(),
+                                if explicit {
+                                    &S3ErrorCode::MethodNotAllowed
+                                } else {
+                                    &S3ErrorCode::NoSuchKey
+                                }
+                            );
+                            let headers = error
+                                .headers()
+                                .unwrap_or_else(|| panic!("{method} marker response omitted headers"));
+                            assert_eq!(headers.get("x-amz-delete-marker").expect("marker header"), "true");
+                            assert_eq!(headers.get("x-amz-version-id").expect("version header"), marker_id.as_str());
+                            if explicit {
+                                let modified = marker
+                                    .mod_time
+                                    .expect("marker timestamp")
+                                    .format(&RFC1123)
+                                    .expect("HTTP date");
+                                assert_eq!(headers.get("last-modified").expect("marker timestamp header"), modified.as_str());
+                            }
+                            let wire = error.to_http_response().expect("serialize marker error response");
+                            assert_eq!(
+                                wire.status(),
+                                if explicit {
+                                    StatusCode::METHOD_NOT_ALLOWED
+                                } else {
+                                    StatusCode::NOT_FOUND
+                                }
+                            );
+                            assert_eq!(wire.headers()["x-amz-delete-marker"], "true");
+                            assert_eq!(wire.headers()["x-amz-version-id"], marker_id.as_str());
+                            assert_eq!(wire.headers()[http::header::CONTENT_TYPE], "application/xml");
+                        }
+                    }
+
+                    let original_id = original.version_id.expect("historical version id").to_string();
+                    let get = GetObjectInput::builder()
+                        .bucket(bucket.clone())
+                        .key(key.to_string())
+                        .version_id(Some(original_id.clone()))
+                        .build()
+                        .expect("historical GET request");
+                    let response = usecase
+                        .execute_get_object(build_request(get, Method::GET))
+                        .await
+                        .expect("historical GET");
+                    assert!(!response.headers.contains_key("x-amz-delete-marker"));
+                    let mut body = response.output.body.expect("historical body");
+                    let mut actual = Vec::new();
+                    while let Some(chunk) = body.next().await {
+                        actual.extend_from_slice(&chunk.expect("historical data remains readable"));
+                    }
+                    assert_eq!(actual, payload);
+                    let head = HeadObjectInput::builder()
+                        .bucket(bucket.clone())
+                        .key(key.to_string())
+                        .version_id(Some(original_id.clone()))
+                        .build()
+                        .expect("historical HEAD request");
+                    let response = usecase
+                        .execute_head_object(build_request(head, Method::HEAD))
+                        .await
+                        .expect("historical HEAD");
+                    assert_eq!(response.output.content_length, Some(i64::try_from(payload.len()).unwrap()));
+                    assert_eq!(response.output.version_id.as_deref(), Some(original_id.as_str()));
+                    assert!(!response.headers.contains_key("x-amz-delete-marker"));
+
+                    for (absent_key, version) in [("missing", None), (key, Some(Uuid::new_v4().to_string()))] {
+                        let get = GetObjectInput::builder()
+                            .bucket(bucket.clone())
+                            .key(absent_key.to_string())
+                            .version_id(version.clone())
+                            .build()
+                            .expect("missing GET request");
+                        let head = HeadObjectInput::builder()
+                            .bucket(bucket.clone())
+                            .key(absent_key.to_string())
+                            .version_id(version)
+                            .build()
+                            .expect("missing HEAD request");
+                        let get_error = usecase
+                            .execute_get_object(build_request(get, Method::GET))
+                            .await
+                            .expect_err("missing GET");
+                        let head_error = usecase
+                            .execute_head_object(build_request(head, Method::HEAD))
+                            .await
+                            .expect_err("missing HEAD");
+                        for error in [get_error, head_error] {
+                            assert_eq!(error.code().status_code(), Some(StatusCode::NOT_FOUND));
+                            assert!(
+                                error
+                                    .headers()
+                                    .is_none_or(|headers| !headers.contains_key("x-amz-delete-marker"))
+                            );
+                            assert!(
+                                error
+                                    .headers()
+                                    .is_none_or(|headers| !headers.contains_key("x-amz-version-id"))
+                            );
+                        }
+                    }
+                }
+            }
+        });
+    }
+
     #[test]
     #[serial_test::serial]
     fn execute_delete_marker_versions_in_single_and_multi_pool() {

rustfs/src/app/object/get.rs

@@ -4093,6 +4093,7 @@ impl DefaultObjectUsecase {
         let prepared_read = match prepared_read {
             Ok(prepared_read) => prepared_read,
             Err(err) => {
+                let err = enrich_delete_marker_read_error(&store, &bucket, &key, &opts, err).await;
                 lifecycle.finish_err();
                 return Self::complete_get_object_error(helper.version_id(version_id_for_event), err);
             }

rustfs/src/app/object/head.rs

@@ -381,10 +381,17 @@ impl DefaultObjectUsecase {
                     {
                         return Self::finish_on_demand_migration_head(&req, &bucket, helper, result?).await;
                     }
-                    return Err(S3Error::new(S3ErrorCode::NoSuchKey));
+                    return Err(enrich_delete_marker_read_error(
+                        &store,
+                        &bucket,
+                        &key,
+                        &opts,
+                        S3Error::new(S3ErrorCode::NoSuchKey),
+                    )
+                    .await);
                 }
                 // Other errors, such as insufficient permissions, still return the original error
-                return Err(ApiError::from(err).into());
+                return Err(enrich_delete_marker_read_error(&store, &bucket, &key, &opts, ApiError::from(err).into()).await);
             }
         };
         if info.delete_marker {
@@ -396,9 +403,13 @@ impl DefaultObjectUsecase {
                 {
                     return Self::finish_on_demand_migration_head(&req, &bucket, helper, result?).await;
                 }
-                return Err(S3Error::new(S3ErrorCode::NoSuchKey));
+                return Err(with_delete_marker_read_headers(S3Error::new(S3ErrorCode::NoSuchKey), &info, None));
             }
-            return Err(S3Error::new(S3ErrorCode::MethodNotAllowed));
+            return Err(with_delete_marker_read_headers(
+                S3Error::new(S3ErrorCode::MethodNotAllowed),
+                &info,
+                opts.version_id.as_deref(),
+            ));
         }
         if let Some(match_etag) = if_none_match
             && let Some(strong_etag) = match_etag.into_etag()

rustfs/src/app/object/shared.rs

@@ -16,6 +16,7 @@
 
 use super::*;
 use crate::on_demand_migration::{OdmStateError, PolicyConfig, SourceErrorPolicy, SourceHead};
+use s3s::header::{X_AMZ_DELETE_MARKER, X_AMZ_VERSION_ID};
 
 pub(super) const RUSTFS_EXPECTED_CURRENT_VERSION_ID: &str = "x-rustfs-expected-current-version-id";
 
@@ -31,6 +32,87 @@ pub(super) const LOG_COMPONENT_APP: &str = "app";
 
 pub(super) const LOG_SUBSYSTEM_OBJECT: &str = "object";
 
+fn is_delete_marker_read_error(err: &S3Error, version_id: Option<&str>) -> bool {
+    let code = if version_id.is_some() {
+        S3ErrorCode::MethodNotAllowed
+    } else {
+        S3ErrorCode::NoSuchKey
+    };
+    err.code() == &code && err.status_code() == code.status_code()
+}
+
+/// A delete marker is an error response, but its identity is still part of the
+/// S3 read contract. Never attach the identity of a different explicit version
+/// or turn an unrelated failure into a marker response.
+pub(super) fn with_delete_marker_read_headers(mut err: S3Error, info: &ObjectInfo, version_id: Option<&str>) -> S3Error {
+    if !info.delete_marker || !is_delete_marker_read_error(&err, version_id) {
+        return err;
+    }
+    let marker_version = info.version_id.unwrap_or_else(Uuid::nil);
+    if let Some(requested) = version_id {
+        let requested = if requested.eq_ignore_ascii_case(NULL_VERSION_ID) {
+            Ok(Uuid::nil())
+        } else {
+            Uuid::parse_str(requested)
+        };
+        if requested.ok() != Some(marker_version) {
+            return err;
+        }
+    }
+    let version = if marker_version.is_nil() {
+        HeaderValue::from_static(NULL_VERSION_ID)
+    } else {
+        let Ok(version) = HeaderValue::from_str(&marker_version.to_string()) else {
+            return err;
+        };
+        version
+    };
+    let mut headers = err.headers().cloned().unwrap_or_default();
+    // s3s replaces, rather than extends, the serialized error's header map.
+    // Keep its XML content type when adding our custom error headers.
+    headers
+        .entry(http::header::CONTENT_TYPE)
+        .or_insert(HeaderValue::from_static("application/xml"));
+    headers.insert(X_AMZ_DELETE_MARKER, HeaderValue::from_static("true"));
+    headers.insert(X_AMZ_VERSION_ID, version);
+    if version_id.is_some()
+        && let Some(mod_time) = info.mod_time
+        && let Ok(date) = mod_time.to_offset(time::UtcOffset::UTC).format(&RFC1123)
+        && let Ok(value) = HeaderValue::from_str(&date)
+    {
+        headers.insert(http::header::LAST_MODIFIED, value);
+    }
+    err.set_headers(headers);
+    err
+}
+
+/// Recover marker metadata only after a local read and its existing fallbacks
+/// have failed. Successful reads, unversioned misses and other error classes
+/// do not pay for another lookup. A racing PUT/purge or failed metadata lookup
+/// must keep the original failure, never resurrect an object or invent an ID.
+pub(super) async fn enrich_delete_marker_read_error(
+    store: &ECStore,
+    bucket: &str,
+    key: &str,
+    opts: &ObjectOptions,
+    err: S3Error,
+) -> S3Error {
+    if !(opts.versioned || opts.version_suspended || opts.version_id.is_some())
+        || !is_delete_marker_read_error(&err, opts.version_id.as_deref())
+    {
+        return err;
+    }
+    let mut metadata_opts = opts.clone();
+    // The read already chose its error. This lookup supplies identity only;
+    // object-body conditions cannot replace that error or hide its marker.
+    metadata_opts.http_preconditions = None;
+    metadata_opts.part_number = None;
+    match store.get_object_info_for_delete(bucket, key, &metadata_opts).await {
+        Ok(info) => with_delete_marker_read_headers(err, &info, opts.version_id.as_deref()),
+        Err(_) => err,
+    }
+}
+
 pub(super) fn decoded_content_length_from_headers(headers: &HeaderMap) -> S3Result<Option<i64>> {
     let Some(val) = headers.get(AMZ_DECODED_CONTENT_LENGTH) else {
         return Ok(None);
@@ -1019,6 +1101,117 @@ mod tests {
     };
     use std::sync::Arc;
 
+    #[test]
+    fn delete_marker_read_headers_round_trip_uuid_and_null_errors() {
+        let uuid = Uuid::parse_str("9341ae04-d4ce-468c-a4e1-6501d58cd6b7").unwrap();
+        let modified = time::macros::datetime!(2026-09-09 12:30:45 +08:00);
+        for stored_version in [Some(uuid), Some(Uuid::nil()), None] {
+            let expected_version = stored_version
+                .filter(|id| !id.is_nil())
+                .map(|id| id.to_string())
+                .unwrap_or_else(|| "null".to_string());
+            let info = ObjectInfo {
+                delete_marker: true,
+                version_id: stored_version,
+                mod_time: Some(modified),
+                ..Default::default()
+            };
+            let explicit = stored_version.unwrap_or_else(Uuid::nil).to_string().to_uppercase();
+            for requested in [None, Some(explicit.as_str()), Some(expected_version.as_str())] {
+                let code = if requested.is_some() {
+                    S3ErrorCode::MethodNotAllowed
+                } else {
+                    S3ErrorCode::NoSuchKey
+                };
+                let status = code.status_code().unwrap();
+                let error = with_delete_marker_read_headers(S3Error::new(code), &info, requested);
+                let response = error.to_http_response().expect("marker error must serialize");
+                assert_eq!(response.status(), status);
+                assert_eq!(response.headers()[http::header::CONTENT_TYPE], "application/xml");
+                assert_eq!(response.headers()[X_AMZ_DELETE_MARKER], "true");
+                assert_eq!(response.headers()[X_AMZ_VERSION_ID], expected_version);
+                if requested.is_some() {
+                    assert_eq!(response.headers()[http::header::LAST_MODIFIED], "Wed, 09 Sep 2026 04:30:45 GMT");
+                } else {
+                    assert!(!response.headers().contains_key(http::header::LAST_MODIFIED));
+                }
+            }
+        }
+    }
+
+    #[test]
+    fn delete_marker_read_headers_preserve_error_context() {
+        let info = ObjectInfo {
+            delete_marker: true,
+            version_id: Some(Uuid::new_v4()),
+            ..Default::default()
+        };
+        let mut original = S3Error::with_message(S3ErrorCode::NoSuchKey, "original local read failure");
+        original.set_source(Box::new(io::Error::other("original storage cause")));
+        original.set_request_id("request-id");
+        original.set_status_code(StatusCode::NOT_FOUND);
+        let mut headers = HeaderMap::new();
+        headers.insert("x-test-existing", HeaderValue::from_static("preserved"));
+        headers.insert(http::header::CONTENT_TYPE, HeaderValue::from_static("application/custom+xml"));
+        original.set_headers(headers);
+        let error = with_delete_marker_read_headers(original, &info, None);
+        assert_eq!(error.code(), &S3ErrorCode::NoSuchKey);
+        assert_eq!(error.status_code(), Some(StatusCode::NOT_FOUND));
+        assert_eq!(error.message(), Some("original local read failure"));
+        assert_eq!(error.request_id(), Some("request-id"));
+        assert_eq!(error.source().unwrap().to_string(), "original storage cause");
+        assert_eq!(error.headers().unwrap()["x-test-existing"], "preserved");
+        assert_eq!(error.headers().unwrap()[http::header::CONTENT_TYPE], "application/custom+xml");
+        assert_eq!(error.headers().unwrap()[X_AMZ_DELETE_MARKER], "true");
+
+        let explicit = info.version_id.unwrap().to_string();
+        let error = with_delete_marker_read_headers(S3Error::new(S3ErrorCode::MethodNotAllowed), &info, Some(&explicit));
+        assert!(
+            !error.headers().unwrap().contains_key(http::header::LAST_MODIFIED),
+            "missing metadata cannot invent a date"
+        );
+    }
+
+    #[test]
+    fn delete_marker_read_headers_reject_unrelated_failures_and_versions() {
+        let marker_id = Uuid::new_v4();
+        let version = marker_id.to_string();
+        let mut info = ObjectInfo {
+            delete_marker: true,
+            version_id: Some(marker_id),
+            ..Default::default()
+        };
+        for (requested, code) in [
+            (None, S3ErrorCode::AccessDenied),
+            (None, S3ErrorCode::InternalError),
+            (None, S3ErrorCode::PreconditionFailed),
+            (None, S3ErrorCode::NotModified),
+            (None, S3ErrorCode::NoSuchVersion),
+            (None, S3ErrorCode::MethodNotAllowed),
+            (Some(version.as_str()), S3ErrorCode::NoSuchKey),
+            (Some(version.as_str()), S3ErrorCode::AccessDenied),
+        ] {
+            let error = with_delete_marker_read_headers(S3Error::new(code), &info, requested);
+            assert!(error.headers().is_none());
+        }
+        let wrong_version = Uuid::new_v4().to_string();
+        for requested in [wrong_version.as_str(), "null", "", "not-a-version", "bad\r\nheader: injected"] {
+            let error = with_delete_marker_read_headers(S3Error::new(S3ErrorCode::MethodNotAllowed), &info, Some(requested));
+            assert!(error.headers().is_none());
+        }
+        let mut original = S3Error::new(S3ErrorCode::NoSuchKey);
+        original.set_status_code(StatusCode::FORBIDDEN);
+        let error = with_delete_marker_read_headers(original, &info, None);
+        assert_eq!(error.status_code(), Some(StatusCode::FORBIDDEN));
+        assert!(error.headers().is_none());
+
+        info.delete_marker = false;
+        let error = with_delete_marker_read_headers(S3Error::new(S3ErrorCode::NoSuchKey), &info, None);
+        assert!(error.headers().is_none(), "a racing PUT is not a delete marker");
+        let error = with_delete_marker_read_headers(S3Error::new(S3ErrorCode::MethodNotAllowed), &info, Some(&version));
+        assert!(error.headers().is_none());
+    }
+
     #[test]
     fn parse_expires_header_accepts_http_date() {
         let expires = parse_expires_header(Some("Wed, 21 Oct 2015 07:28:00 GMT"))