perf(s3): borrow conditional write headers

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

crates/e2e_test/src/delete_marker_migration_semantics_test.rs

@@ -16,8 +16,193 @@
 mod tests {
     use crate::common::{RustFSTestEnvironment, init_logging};
     use aws_sdk_s3::Client;
+    use aws_sdk_s3::config::http::HttpResponse;
+    use aws_sdk_s3::error::{ProvideErrorMetadata, SdkError};
     use aws_sdk_s3::primitives::ByteStream;
-    use aws_sdk_s3::types::{BucketVersioningStatus, VersioningConfiguration};
+    use aws_sdk_s3::types::{BucketVersioningStatus, CompletedMultipartUpload, CompletedPart, VersioningConfiguration};
+
+    fn assert_not_found<E: ProvideErrorMetadata + std::fmt::Debug>(error: &SdkError<E, HttpResponse>) {
+        assert_eq!(error.raw_response().map(|r| r.status().as_u16()), Some(404), "{error:?}");
+        assert_eq!(
+            error.as_service_error().and_then(ProvideErrorMetadata::code),
+            Some("NoSuchKey"),
+            "{error:?}"
+        );
+    }
+
+    #[tokio::test]
+    async fn conditional_writes_cannot_replace_delete_marker_with_if_match() {
+        init_logging();
+        let mut env = RustFSTestEnvironment::new().await.expect("create test environment");
+        env.start_rustfs_server(vec![]).await.expect("start RustFS");
+        let client = env.create_s3_client();
+        let bucket = "conditional-delete-marker-proof";
+        create_versioned_bucket(&client, bucket).await;
+        client
+            .put_object()
+            .bucket(bucket)
+            .key("source")
+            .body(ByteStream::from_static(b"source"))
+            .send()
+            .await
+            .expect("write copy source");
+
+        for operation in ["put", "copy", "multipart"] {
+            let old = client
+                .put_object()
+                .bucket(bucket)
+                .key(operation)
+                .body(ByteStream::from_static(b"old"))
+                .send()
+                .await
+                .expect("write historical version");
+            let old_etag = old.e_tag().expect("historical ETag");
+            let marker = client
+                .delete_object()
+                .bucket(bucket)
+                .key(operation)
+                .send()
+                .await
+                .expect("create delete marker");
+            let marker_id = marker.version_id().expect("delete marker version");
+
+            let (upload_id, completed) = if operation == "multipart" {
+                let upload = client
+                    .create_multipart_upload()
+                    .bucket(bucket)
+                    .key(operation)
+                    .send()
+                    .await
+                    .expect("initiate multipart upload");
+                let upload_id = upload.upload_id().expect("upload ID").to_owned();
+                let part = client
+                    .upload_part()
+                    .bucket(bucket)
+                    .key(operation)
+                    .upload_id(&upload_id)
+                    .part_number(1)
+                    .body(ByteStream::from_static(b"new"))
+                    .send()
+                    .await
+                    .expect("upload part");
+                let completed = CompletedMultipartUpload::builder()
+                    .parts(
+                        CompletedPart::builder()
+                            .part_number(1)
+                            .set_e_tag(part.e_tag().map(str::to_owned))
+                            .build(),
+                    )
+                    .build();
+                (Some(upload_id), Some(completed))
+            } else {
+                (None, None)
+            };
+
+            for etag in [old_etag, "\"00000000000000000000000000000000\""] {
+                match operation {
+                    "put" => {
+                        let err = client
+                            .put_object()
+                            .bucket(bucket)
+                            .key(operation)
+                            .if_match(etag)
+                            .body(ByteStream::from_static(b"new"))
+                            .send()
+                            .await
+                            .expect_err("If-Match must reject marker");
+                        assert_not_found(&err);
+                    }
+                    "copy" => {
+                        let err = client
+                            .copy_object()
+                            .bucket(bucket)
+                            .key(operation)
+                            .copy_source(format!("{bucket}/source"))
+                            .if_match(etag)
+                            .send()
+                            .await
+                            .expect_err("If-Match copy must reject marker");
+                        assert_not_found(&err);
+                    }
+                    _ => {
+                        let err = client
+                            .complete_multipart_upload()
+                            .bucket(bucket)
+                            .key(operation)
+                            .upload_id(upload_id.as_deref().expect("upload ID"))
+                            .multipart_upload(completed.clone().expect("completed parts"))
+                            .if_match(etag)
+                            .send()
+                            .await
+                            .expect_err("If-Match completion must reject marker");
+                        assert_not_found(&err);
+                    }
+                }
+                let versions = client
+                    .list_object_versions()
+                    .bucket(bucket)
+                    .prefix(operation)
+                    .send()
+                    .await
+                    .expect("list versions after rejected write");
+                assert_eq!(versions.versions().len(), 1, "{operation}: failed write created a version");
+                assert_eq!(versions.delete_markers().len(), 1);
+                assert_eq!(versions.delete_markers()[0].version_id(), Some(marker_id));
+                assert_eq!(versions.delete_markers()[0].is_latest(), Some(true));
+            }
+
+            match operation {
+                "put" => {
+                    client
+                        .put_object()
+                        .bucket(bucket)
+                        .key(operation)
+                        .if_none_match("*")
+                        .body(ByteStream::from_static(b"new"))
+                        .send()
+                        .await
+                        .expect("If-None-Match PUT over marker");
+                }
+                "copy" => {
+                    client
+                        .copy_object()
+                        .bucket(bucket)
+                        .key(operation)
+                        .copy_source(format!("{bucket}/source"))
+                        .if_none_match("*")
+                        .send()
+                        .await
+                        .expect("If-None-Match copy over marker");
+                }
+                _ => {
+                    client
+                        .complete_multipart_upload()
+                        .bucket(bucket)
+                        .key(operation)
+                        .upload_id(upload_id.as_deref().expect("upload ID"))
+                        .multipart_upload(completed.expect("completed parts"))
+                        .if_none_match("*")
+                        .send()
+                        .await
+                        .expect("If-None-Match completion over marker");
+                }
+            }
+            let current = client
+                .get_object()
+                .bucket(bucket)
+                .key(operation)
+                .send()
+                .await
+                .expect("read successful conditional write")
+                .body
+                .collect()
+                .await
+                .expect("read object body")
+                .into_bytes();
+            let expected: &[u8] = if operation == "copy" { b"source" } else { b"new" };
+            assert_eq!(current.as_ref(), expected, "{operation}: unexpected current object");
+        }
+    }
 
     async fn create_versioned_bucket(client: &Client, bucket: &str) {
         client

crates/ecstore/src/set_disk/core/io_primitives.rs

@@ -6737,9 +6737,7 @@ impl SetDisks {
                         object,
                     );
                 }
-                let if_none_match = http_preconditions.if_none_match_value().map(str::to_owned);
-                let if_match = http_preconditions.if_match_value().map(str::to_owned);
-                if should_prevent_write(&oi, if_none_match, if_match)
+                if should_prevent_write(&oi, http_preconditions.if_none_match_value(), http_preconditions.if_match_value())
                     && !crate::data_movement::can_replace_stale_data_movement_target(&oi, opts)
                 {
                     return Some(StorageError::PreconditionFailed);

crates/ecstore/src/set_disk/mod.rs

@@ -120,7 +120,6 @@ use bytesize::ByteSize;
 use futures::future::join_all;
 use http::HeaderMap;
 use md5::{Digest as Md5Digest, Md5};
-use regex::Regex;
 use rustfs_config::MI_B;
 use rustfs_filemeta::metadata_keys;
 use rustfs_filemeta::{
@@ -8036,16 +8035,29 @@ fn parts_after_marker(part_numbers: &[usize], part_number_marker: usize) -> Opti
     Some(&part_numbers[index..])
 }
 
-pub fn canonicalize_etag(etag: &str) -> String {
-    let re = Regex::new("\"*?([^\"]*?)\"*?$").unwrap();
-    re.replace_all(etag, "$1").to_string()
+fn canonicalized_etag(etag: &str) -> Cow<'_, str> {
+    // Preserve the legacy suffix-matching behavior for embedded quotes.
+    let trimmed = etag.trim_end_matches('"');
+    let Some(last_quote) = trimmed.rfind('"') else {
+        return Cow::Borrowed(trimmed);
+    };
+    let prefix = trimmed[..last_quote].trim_end_matches('"');
+    let suffix = &trimmed[last_quote + 1..];
+    if prefix.is_empty() {
+        Cow::Borrowed(suffix)
+    } else {
+        let mut result = String::with_capacity(prefix.len() + suffix.len());
+        result.push_str(prefix);
+        result.push_str(suffix);
+        Cow::Owned(result)
+    }
 }
 
 pub fn e_tag_matches(etag: &str, condition: &str) -> bool {
     if condition.trim() == "*" {
         return true;
     }
-    canonicalize_etag(etag) == canonicalize_etag(condition)
+    canonicalized_etag(etag) == canonicalized_etag(condition)
 }
 
 fn weak_etag_matches(etag: &str, condition: &str) -> bool {
@@ -8060,15 +8072,12 @@ fn weak_etag_matches(etag: &str, condition: &str) -> bool {
             .or_else(|| trimmed.strip_prefix("w/"))
             .unwrap_or(trimmed)
     }
-    canonicalize_etag(strip_weak(etag)) == canonicalize_etag(strip_weak(condition))
+    canonicalized_etag(strip_weak(etag)) == canonicalized_etag(strip_weak(condition))
 }
 
-pub fn should_prevent_write(oi: &ObjectInfo, if_none_match: Option<String>, if_match: Option<String>) -> bool {
-    let if_none_match = if_none_match
-        .as_deref()
-        .map(str::trim)
-        .filter(|condition| !condition.is_empty());
-    let if_match = if_match.as_deref().map(str::trim).filter(|condition| !condition.is_empty());
+pub fn should_prevent_write(oi: &ObjectInfo, if_none_match: Option<&str>, if_match: Option<&str>) -> bool {
+    let if_none_match = if_none_match.map(str::trim).filter(|condition| !condition.is_empty());
+    let if_match = if_match.map(str::trim).filter(|condition| !condition.is_empty());
 
     match &oi.etag {
         Some(etag) => {
@@ -13772,6 +13781,22 @@ mod tests {
         assert!(result2.iter().all(|d| d.is_none()));
     }
 
+    #[test]
+    fn canonicalized_etag_preserves_legacy_regex_semantics() {
+        let regex = regex::Regex::new("\"*?([^\"]*?)\"*?$").expect("valid legacy ETag pattern");
+        let alphabet = ['a', '"', '\n', 'é', '\0'];
+        for length in 0..=6_u32 {
+            for mut number in 0..alphabet.len().pow(length) {
+                let mut input = String::new();
+                for _ in 0..length {
+                    input.push(alphabet[number % alphabet.len()]);
+                    number /= alphabet.len();
+                }
+                assert_eq!(canonicalized_etag(&input), regex.replace_all(&input, "$1"), "input: {input:?}");
+            }
+        }
+    }
+
     #[test]
     fn test_etag_matches() {
         assert!(e_tag_matches("abc", "abc"));
@@ -14347,61 +14372,61 @@ mod tests {
             etag: Some("abc".to_string()),
             ..Default::default()
         };
-        let if_none_match = Some("abc".to_string());
+        let if_none_match = Some("abc");
         let if_match = None;
         assert!(should_prevent_write(&oi, if_none_match, if_match));
 
-        let if_none_match = Some("*".to_string());
+        let if_none_match = Some("*");
         let if_match = None;
         assert!(should_prevent_write(&oi, if_none_match, if_match));
 
         let if_none_match = None;
-        let if_match = Some("def".to_string());
+        let if_match = Some("def");
         assert!(should_prevent_write(&oi, if_none_match, if_match));
 
         let if_none_match = None;
-        let if_match = Some("*".to_string());
+        let if_match = Some("*");
         assert!(!should_prevent_write(&oi, if_none_match, if_match));
 
-        let if_none_match = Some("def".to_string());
+        let if_none_match = Some("def");
         let if_match = None;
         assert!(!should_prevent_write(&oi, if_none_match, if_match));
 
-        let if_none_match = Some("def".to_string());
-        let if_match = Some("*".to_string());
+        let if_none_match = Some("def");
+        let if_match = Some("*");
         assert!(!should_prevent_write(&oi, if_none_match, if_match));
 
-        let if_none_match = Some("def".to_string());
-        let if_match = Some("\"abc\"".to_string());
+        let if_none_match = Some("def");
+        let if_match = Some("\"abc\"");
         assert!(!should_prevent_write(&oi, if_none_match, if_match));
 
-        let if_none_match = Some("*".to_string());
-        let if_match = Some("\"abc\"".to_string());
+        let if_none_match = Some("*");
+        let if_match = Some("\"abc\"");
         assert!(should_prevent_write(&oi, if_none_match, if_match));
 
         let oi = ObjectInfo {
             etag: None,
             ..Default::default()
         };
-        let if_none_match = Some("*".to_string());
-        let if_match = Some("\"abc\"".to_string());
+        let if_none_match = Some("*");
+        let if_match = Some("\"abc\"");
         assert!(should_prevent_write(&oi, if_none_match, if_match));
 
         let if_none_match = None;
         let if_match = None;
         assert!(!should_prevent_write(&oi, if_none_match, if_match));
 
-        let if_none_match = Some(String::new());
-        let if_match = Some(" ".to_string());
+        let if_none_match = Some("");
+        let if_match = Some(" ");
         assert!(!should_prevent_write(&oi, if_none_match, if_match));
 
         let oi = ObjectInfo {
             etag: Some("abc".to_string()),
             ..Default::default()
         };
-        assert!(should_prevent_write(&oi, Some("W/\"abc\"".to_string()), None));
-        assert!(!should_prevent_write(&oi, Some("W/\"def\"".to_string()), None));
-        assert!(should_prevent_write(&oi, None, Some("W/\"abc\"".to_string())));
+        assert!(should_prevent_write(&oi, Some("W/\"abc\""), None));
+        assert!(!should_prevent_write(&oi, Some("W/\"def\""), None));
+        assert!(should_prevent_write(&oi, None, Some("W/\"abc\"")));
     }
 
     #[test]

docs/testing/ci-event-drills.md

@@ -0,0 +1,49 @@
+# Event-class drills and retained evidence
+
+**Use this when:** you need to confirm that a CI event class has been exercised end to end, locate its retained run evidence, or repeat the drill. **Source of truth:** the live `main` ruleset for required status; `.github/workflows/ci.yml` for event triggers; `scripts/ci_gate.py` for per-event selection and expected results; the run URLs below for observed outcomes.
+
+Every event class the planner serves must have at least one reviewable drill: an actual run whose plan, job set and gate result can be re-inspected. A drill proves the wiring for that event, not product quality and not a cost saving. Static fixtures in `scripts/test_ci_plan.py` and `scripts/test_ci_merge_queue.py` cover the negative cases (missing, cancelled, mismatched, combined-candidate conflicts) and are not substitutes for the live drills listed here.
+
+## Event classes and expected selection
+
+| Event | Selection | Expected gate behavior |
+|---|---|---|
+| `pull_request` (ordinary code) | reverse-dependency closure + fixed core inventory; shared/unknown inputs expand to full | `Test and Lint` requires exactly the planned job results |
+| `pull_request` (documentation only) | `docs` mode: no code jobs | Quick Checks and Typos run; expensive jobs are required **skipped** |
+| `pull_request` (isolated E2E) | `e2e` mode: workspace + server build + E2E smoke + S3 lanes | Connect boundary, serial ILM, protocol/rio-v2, real io_uring required **skipped** |
+| `pull_request` (closed) | `Cancel Closed PR Runs` only | in-flight validation for that PR is cancelled; no new validation is claimed |
+| `pull_request` (merged) | entry into `merge_group` | combined candidate is validated as its own SHA/tree with the full matrix |
+| `push` to `main` | full | complete validation; no reuse inferred from tree equality |
+| `push` of a release tag | `build.yml` release matrix | build and publication path (not the PR planner) |
+| `schedule` | registry-selected cadence plan | suites run under their resource locks; failures feed the health monitors |
+| `workflow_dispatch` | full unless an explicit narrow input is passed | manual runs without a plan stay visible as unverified |
+| `workflow_run` | monitor/filter through the responsibility registry | independent health layer reports expected SHA, attempt and evidence completeness |
+
+## Retained evidence index
+
+Run URLs are `https://github.com/rustfs/rustfs/actions/runs/<id>`. Re-open the run and its `Select CI scope` job log to read the planner's recorded mode, reasons and expected job results.
+
+| Class | Drill | Run | Observed |
+|---|---|---|---|
+| Ordinary PR | `test(heal): align cleanup tests with removal-disabled heals` | [41153190859](https://github.com/rustfs/rustfs/actions/runs/41153190859) | full scope (12 real jobs: workspace build/tests, E2E, S3, core contract, io_uring); gate accepted exact planned results |
+| PR closed | `test(heal): align cleanup tests…` closure handler | [41160735383](https://github.com/rustfs/rustfs/actions/runs/41160735383) | only `Cancel Closed PR Runs` executed; every validation job required **skipped** |
+| `push` to `main` | `chore(deps): refresh workspace dependencies (#9080)` | [41160768241](https://github.com/rustfs/rustfs/actions/runs/41160768241) | full matrix on the merge commit; no selection shortcut |
+| Release tag | `1.0.4` | [40198812318](https://github.com/rustfs/rustfs/actions/runs/40198812318) | `Build and Release` release matrix succeeded on the tag ref |
+| `schedule` | `Functional chain health` | [41159991473](https://github.com/rustfs/rustfs/actions/runs/41159991473) | monitor refused to advance freshness on an incomplete chain attempt (failure is the expected answer while the chain is unhealthy) |
+| `workflow_dispatch` | manual full validation | [41152382212](https://github.com/rustfs/rustfs/actions/runs/41152382212) | full suite set ran and the aggregate accepted the complete plan (85 min) |
+| `workflow_run` | `Scheduled Validation Watchdog` | [41161431618](https://github.com/rustfs/rustfs/actions/runs/41161431618) | independent filter/monitor path executed against the registry |
+| `merge_group` | first queued candidate after queue enablement | see below | live evidence attached after the first queue entry completes |
+| Documentation-only PR | this document's own PR | see below | `docs` mode: expensive jobs required **skipped** |
+| Isolated E2E PR | drill branch confined to `crates/e2e_test/src/` | see below | `e2e` mode: Connect/ILM/protocol lanes required **skipped** |
+
+## Repeating a drill
+
+Ordinary, docs and E2E classes: open a pull request containing only that class's paths and inspect `Select CI scope` → the planner's `mode` and `reasons`, then the aggregate's expected-vs-actual comparison. Close the PR without merging to leave no product change; the closure handler itself is the closed-event drill.
+
+Merge-group class: queue a pull request through the enabled `merge_queue` rule (see [verification plans and artifacts](ci-verification-pipeline.md) for the switch record and rollback payload). The `merge_group` run validates the combined candidate's own SHA/tree. Two individually passing changes whose combination fails stay blocked until their combined candidate passes; that negative case is fixed in `scripts/test_ci_merge_queue.py`.
+
+Schedule and `workflow_run` classes cannot be forced on demand without an event; use the retained runs above and the registry's own `python3 scripts/validation_registry.py check` output. Tag class requires a real tag push; use the retained release runs or the preview-tag rehearsal described in [CI gate matrix](ci-gates.md).
+
+## Boundaries
+
+A completed drill is evidence that the event wiring behaves as planned for that candidate. It is not release certification, not a latency or cost result (use [CI timing samples](ci-timing.md) and [resource accounting](ci-resource-accounting.md)), and not a substitute for the required checks' own product coverage. Do not edit this file to claim a drill that has no run URL; attach the run first.