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

@@ -33,8 +33,8 @@ Run URLs are `https://github.com/rustfs/rustfs/actions/runs/<id>`. Re-open the r
 | `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 | PR #9088, this document's own PR | [41162871854](https://github.com/rustfs/rustfs/actions/runs/41162871854) | `docs` mode: `Select CI scope` recorded `mode: docs`; `Test and Lint` accepted in 8 s with every expensive job required **skipped** |
-| Isolated E2E PR | PR #9089, comment-only change under `crates/e2e_test/src/` | [41163203959](https://github.com/rustfs/rustfs/actions/runs/41163203959) | `e2e` mode: Connect boundary, serial ILM, rio-v2/protocol lanes and real io_uring required **skipped**; workspace build, E2E and both S3 lanes retained |
+| 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
 

docs/testing/ci-timing-sample-2026-10-31.md

@@ -1,72 +0,0 @@
-# CI timing sample 2026-10-31 (backlog #2963 CI01 measurement)
-
-Operational measurement of ordinary-code PR feedback and runner occupancy against the CI01 experimental targets (feedback median 15–20 min, P95 ≤ 30 min, paid occupancy median 50–60 runner minutes). Methodology: [ci-timing.md](ci-timing.md) and [ci-resource-accounting.md](ci-resource-accounting.md). Raw snapshots: `/tmp/ci-timing.json` (limit 30 reproduction), `/tmp/ci-timing-200.json` (analysis sample), `/tmp/ci-acc-w{1,3}.json`. Nothing in this report is an acceptance result; the targets are **not met** by the measured sample.
-
-## Sample window and n
-
-| Item | Value |
-|---|---|
-| Collection time | 2026-10-31 ~02:50–03:15 UTC |
-| Window | completed `pull_request` runs of `ci.yml` created in the last 7 days |
-| Runs sampled | 30 (reproduction) and 200 (analysis, script max) |
-| Successful-code runs (classifier) | 1 of 30; **39 of 200** |
-| Other outcomes in 200-run sample | 68 cancelled (34 %), 36 failed, remainder non-code greens/closure handlers |
-| Lane shapes in sample | 24 combined-lane runs (`Workspace Test and Lint`), 15 split-lane runs (`Workspace Lint and Build` + `Workspace tests` + `Core contract`) |
-
-The limit-30 reproduction reports median wall 76.85 min and median job-sum 133.33 runner-min on n=1 (per methodology, an absent/small median is not zero). All medians/P95 below are from the n=39 analysis sample. Two of the 39 are rerun continuations (run_attempt > 1, wall 7.0 and 22.6 min re-executing only failed jobs under the rerun-start convention); they do not move the median.
-
-## Measured feedback and occupancy (ordinary successful code runs, n=39)
-
-| Metric | Median | P90 | P95 | Max |
-|---|---|---|---|---|
-| Wall feedback (created/rerun-start → last job end) | **76.3 min** | 94.5 | **97.0** | 103.2 |
-| Runner-minute job sum per run | **179.5** | 206.5 | **213.6** | 220.4 |
-
-Independent accounting cross-check (`scripts/ci/actions_cost_report.py`, 2026-10-30T00–12Z window, 547 attempts, collection **incomplete** — see gaps): successful-code PR feedback median 88.34 min (n=12), runner_sum median 199.68 min (n=12). A complete half-day window (2026-10-31T00–12Z, 78 attempts) shows total known runner occupancy 1379.72 min with 42 jobs of unknown occupancy; queue wait summed 28.43 min, so queueing is not the dominant cost.
-
-## Per-job critical path breakdown
-
-Execution minutes across the 39 runs (known only):
-
-| Job | n | Median | P95 |
-|---|---|---|---|
-| Workspace Test and Lint (combined lane) | 24 | 63.6 | 73.9 |
-| Connect Feature Boundaries | 38 | 53.5 | 70.4 |
-| Workspace Lint and Build (split lane) | 15 | 43.3 | 56.5 |
-| Workspace tests (split lane) | 15 | 29.2 | 43.0 |
-| Build RustFS Debug Binary | 39 | 18.5 | 24.9 |
-| End-to-End Tests | 39 | 11.4 | 16.7 |
-| S3 Implemented Tests | 39 | 9.8 | 11.1 |
-| io_uring Integration (real) | 39 | 7.7 | 8.8 |
-| Quick Checks | 39 | 7.2 | 9.2 |
-
-Per-run chain timeline (start = attempt start; see `/tmp/per-run-jobs.json`):
-
-- Split lane (15 runs): `Quick Checks` (~7–10) → `Workspace Lint and Build` (ends ~46–67) → `Workspace tests` (ends ~71–103) → shard evidence → aggregator `Test and Lint`/`required-checks`. The lint→tests chain sets the wall in most split runs.
-- Combined lane (24 runs): `Quick Checks` → `Workspace Test and Lint` alone runs 63–86 min and sets the wall.
-- `Connect Feature Boundaries` (median 53.5, P95 70.4) runs parallel to the lint chain and occasionally becomes the tail (e.g. runs 41021787700, 40750122159: connect end within 0.2 min of wall).
-- `Build RustFS Debug Binary` → E2E/Core-contract/S3 chain finishes with 30–60 min of slack in every sampled run.
-- The required-check aggregator `Test and Lint` (`required-checks`, `ci_gate.py verify`) finishes last by construction (~0.2 min).
-
-## Against the CI01 targets
-
-| Target | Measured | Verdict |
-|---|---|---|
-| Feedback median 15–20 min | 76.3 min (n=39) | **not met** (~3.8–5× over) |
-| Feedback P95 ≤ 30 min | 97.0 min | **not met** (~3.2× over) |
-| Occupancy median 50–60 runner-min | 179.5 runner-min | **not met** (~3–3.6× over) |
-
-These are the experimental-target measurements. The full merge-lifecycle cost is separate and was not established here (the accounting PR-allocation buckets were empty in the partial window; see gaps).
-
-## Bottlenecks (honest list)
-
-1. **Workspace lint+test compile/test chain (critical path):** combined `Workspace Test and Lint` median 63.6 min, or split `Workspace Lint and Build` 43.3 + `Workspace tests` 29.2 ≈ 72.5 min after Quick Checks. This is the wall clock.
-2. **Connect Feature Boundaries (parallel near-critical):** median 53.5 min on its own lane with `CARGO_BUILD_JOBS: "2"` limiting build concurrency on the credential-test step; becomes the tail when the lint chain is fast.
-3. **Duplicate compilation across lanes:** `Workspace Lint and Build` builds workspace test binaries; `Build RustFS Debug Binary` separately builds the debug binary + E2E binaries on another runner; Connect builds again. `cache-save-if: 'false'` on the Connect lane means that lane never warms the shared `ci-dev` cache.
-4. Runner queueing is small (28.43 queue-min vs 1379.72 runner-min in the complete half-day window) — not the bottleneck.
-
-## Gaps and caveats
-
-- Accounting windows: 2026-10-30T00–12Z collected 547 attempts but is flagged incomplete; 2026-10-30T12–24Z collection did not finish (timeout) and has no report; 2026-10-31T12–24Z was empty at collection time. Occupancy figures are per-window partial sums and must not be summed as a bill.
-- No CI workflow or selection change was made in this pass; strict gate semantics unchanged. No savings are claimed. Candidate fixes (build-artifact reuse between lanes, enabling cache save on the Connect lane, finer affected-selection granularity in shadow mode per ci-timing.md) are **unmeasured proposals**, not results.
-- Timing data alone does not establish functional coverage or escaped-regression rate; the 15–20 min target remains an experiment that currently misses by a wide margin.

scripts/test/oidc_keycloak_live.sh

@@ -27,7 +27,7 @@ cleanup() {
 }
 trap cleanup EXIT INT TERM
 
-for command in curl docker python3 aws; do
+for command in curl docker python3 awscurl; do
   command -v "${command}" >/dev/null || {
     echo "missing required command: ${command}" >&2
     exit 1
@@ -192,19 +192,12 @@ print("\t".join(values[field] for field in ("AccessKeyId", "SecretAccessKey", "S
 PY
 )
 
-# Sign with the STS session to prove the web-identity credentials are usable.
-# The AWS CLI replaces awscurl here: the functional fleet's system Python is
-# PEP 668 managed, so the live gate must not require pip-installed tools. The
-# security suite already requires the AWS CLI. Strip proxy env explicitly
-# (the runners define it) so this localhost request never leaves the host.
-env -u http_proxy -u https_proxy -u HTTP_PROXY -u HTTPS_PROXY \
-  AWS_ACCESS_KEY_ID="${STS_ACCESS_KEY}" \
-  AWS_SECRET_ACCESS_KEY="${STS_SECRET_KEY}" \
-  AWS_SESSION_TOKEN="${STS_SESSION_TOKEN}" \
-  AWS_EC2_METADATA_DISABLED=true \
-  aws --endpoint-url "${RUSTFS_ORIGIN}" --region us-east-1 \
-  s3api list-buckets --output json >"${WORK_DIR}/list-buckets.json"
-grep -q '"Buckets"' "${WORK_DIR}/list-buckets.json"
+awscurl --fail-with-body --service s3 --region us-east-1 \
+  --access_key "${STS_ACCESS_KEY}" \
+  --secret_key "${STS_SECRET_KEY}" \
+  --security_token "${STS_SESSION_TOKEN}" \
+  "${RUSTFS_ORIGIN}/" >"${WORK_DIR}/list-buckets.xml"
+grep -q '<ListAllMyBucketsResult' "${WORK_DIR}/list-buckets.xml"
 
 TAMPERED_TOKEN="$(python3 - "${GOOD_TOKEN}" <<'PY'
 import sys