ci: accept the proving merge-group run as release evidence (#9376)

* ci: accept the proving merge-group run as release evidence

A commit that lands through the merge queue gets a push run of ci.yml in
mode proven: Quick Checks and Test and Lint succeed, and every validation
job, End-to-End Tests (full merge gate) included, is skipped. source_ci
refused such a commit, so no release candidate could be cut from it.

source_ci now accepts that push run when Quick Checks and Test and Lint
succeeded and a successful merge_group run of ci.yml on that branch's
queue has the same head tree and passed all three required jobs itself;
that run is recorded as the candidate's exact-source CI identity. The
match is ci_merge_group_proof.proves, the same predicate the push plan
uses.

Also: a test that a full merge_group run requires every job a full
integration push requires; the sibling queue-branch case
(gh-readonly-queue/mainline/...); select_plan's proof is keyword-only;
the release, CI gate and workflow docs describe proven pushes.

* test(release): pin the proving run's position, attempt and branch

The release test now lists the matching merge-group run among a run of
another tree and a run of another branch's queue, in both orders; gives
it attempt 3 and reads jobs by run and attempt; and accepts a release
branch's own queue run while refusing a main queue run for it.

.config/e2e-full-selection.txt

@@ -1,2 +1,2 @@
-sha256-darwin=0374e6b9b061dd5d622686301386232a4aa1cb22eedc18af70d2a7411bc66f8d
-sha256-linux=f2cee240f7a744ae991971820f48a6c9c28d19f525e8b2345fcb76eeaaf1e382
+sha256-darwin=dde32c85d39a3c2eb1cad6912861883af2b3554fc6d04b86b3aeb2afbcc86c7e
+sha256-linux=155b526f75743796c18c4c5202bbef3a1770b0ea4ba18b32910317a957ad36f2

.config/e2e-smoke-selection.txt

@@ -1 +1 @@
-sha256=21e9a1561ac20382503c38353bbb9b830658534a4f277c254a42d0afc6ebed8b
+sha256=caccbc5bcbfbe1b7548fbd184f1a6ae3aedec7cb84ed862d62dddbe27e75271f

.config/make/tests.mak

@@ -83,6 +83,7 @@ script-tests: ## Run shell script tests
 	$(RUSTFS_PYTHON_BIN) ./scripts/ci_gate.py --self-test
 	$(RUSTFS_PYTHON_BIN) ./scripts/check_connect_release_boundary.py --self-test
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_ci_test_artifact.py
+	$(RUSTFS_PYTHON_BIN) ./scripts/test_ci_merge_group_proof.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_ci_test_shards.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_ci_merge_queue.py
 	$(RUSTFS_PYTHON_BIN) -m unittest discover -s scripts/ci -p 'test_*.py'

.github/workflows/README.md

@@ -53,7 +53,7 @@ A hosted runner is also the only choice for a few things the pods cannot do: `mo
 These are the patterns that have cost real money here without validating anything more:
 
 - An `if: always()` summary or alert job on a self-hosted label inside a `workflow_run`-triggered workflow. The upstream workflow completes many times a day, the real jobs skip, and the summary still starts a pod each time. Gate it on the upstream job having run, and keep it on `ubuntu-latest`.
-- Per-push workflows on `main`. A merge triggers the applicable `ci.yml` matrix and cache warming; `build.yml` development builds are scheduled or manual; with tens of merges a day that is as much self-hosted time as all pull requests together. Anything that does not verify the merged tree itself belongs on a schedule or behind a `paths` filter.
+- Per-push workflows on `main`. A merge triggers cache warming and a `ci.yml` run, which validates the merged tree in full unless a merge-queue run already validated that tree; `build.yml` development builds are scheduled or manual; with tens of merges a day that is as much self-hosted time as all pull requests together. Anything that does not verify the merged tree itself belongs on a schedule or behind a `paths` filter.
 - `apt-get install` of tools the hosted image already ships. It is only needed on the minimal pod images.
 - A throttled `CARGO_BUILD_JOBS` on a 14-CPU pod. The limit exists for link-storm reasons documented in `ci.yml`; measure before raising it, and measure before lowering it further.
 - Tiny jobs on ARC pods. Each one pays roughly a minute of pod start-up before its first step; a compile-free job that takes ten seconds costs more to schedule than to run.

.github/workflows/ci.yml

@@ -88,6 +88,9 @@ jobs:
     if: github.event_name != 'pull_request' || github.event.action != 'closed'
     runs-on: ubuntu-latest
     timeout-minutes: 10
+    permissions:
+      contents: read
+      actions: read
     outputs:
       mode: ${{ steps.scope.outputs.mode }}
       plan: ${{ steps.scope.outputs.plan }}
@@ -107,7 +110,14 @@ jobs:
           CI_BASE_SHA: ${{ github.event.pull_request.base.sha || github.event.merge_group.base_sha || github.event.before }}
           CI_EVENT_ACTION: ${{ github.event.action }}
           CI_REHEARSE_SHARDS: ${{ inputs.rehearse_workspace_shards && 'true' || 'false' }}
+          # A push whose tree a successful merge-group run already passed skips
+          # the validation jobs; the lookup answers null on any doubt.
+          PROOF_TOKEN: ${{ github.event_name == 'push' && github.token || '' }}
         run: |
+          if [[ "$GITHUB_EVENT_NAME" == "push" ]]; then
+            CI_MERGE_GROUP_PROOF="$(GITHUB_TOKEN="$PROOF_TOKEN" python3 -I scripts/ci_merge_group_proof.py)" || CI_MERGE_GROUP_PROOF=null
+            export CI_MERGE_GROUP_PROOF
+          fi
           if [[ "$GITHUB_EVENT_NAME" != "pull_request" ]]; then
             python3 -I scripts/ci_gate.py select
           elif [[ "$CI_BASE_SHA" =~ ^[0-9a-f]{40}$ ]] && git show "$CI_BASE_SHA:scripts/ci_gate.py" > "$RUNNER_TEMP/ci-gate-base.py"; then
@@ -116,7 +126,7 @@ jobs:
             printf '%s\n' 'mode=full' >> "$GITHUB_OUTPUT"
             echo "Base CI policy unavailable; running the full matrix."
           fi
-          if ! grep -q '^plan=' "$GITHUB_OUTPUT" || ! grep -q '^plan_contract=4$' "$GITHUB_OUTPUT"; then
+          if ! grep -q '^plan=' "$GITHUB_OUTPUT" || ! grep -q '^plan_contract=5$' "$GITHUB_OUTPUT"; then
             printf '%s\n' 'mode=full' 'connect=true' 'full_validation=true' 'rehearse_shards=false' 'gateway_smoke=true' 'packages=' 'plan=' >> "$GITHUB_OUTPUT"
             echo "Base policy does not cover this workflow contract; retaining the full matrix."
           fi

crates/audit/tests/performance_test.rs

@@ -34,9 +34,6 @@ async fn test_audit_system_startup_performance() {
 
     println!("Audit system startup took: {elapsed:?}");
 
-    // Should complete within 5 seconds
-    assert!(elapsed < Duration::from_secs(5), "Startup took too long: {elapsed:?}");
-
     // Clean up
     let _ = system.close().await;
 }
@@ -68,9 +65,6 @@ async fn test_concurrent_target_creation() {
 
     println!("Concurrent target creation took: {elapsed:?}");
 
-    // Should complete quickly even with multiple targets
-    assert!(elapsed < Duration::from_secs(10), "Target creation took too long: {elapsed:?}");
-
     // Verify it fails with expected error (server not initialized)
     match result {
         Err(AuditError::StorageNotAvailable(_)) => {
@@ -166,9 +160,6 @@ async fn test_audit_log_dispatch_performance() {
 
     println!("Audit log dispatch took: {elapsed:?}");
 
-    // Should be very fast (sub-millisecond for no targets)
-    assert!(elapsed < Duration::from_millis(100), "Dispatch took too long: {elapsed:?}");
-
     assert!(
         matches!(result, Err(AuditError::NotInitialized(_))),
         "Dispatch on a stopped system should return NotInitialized, got: {result:?}"
@@ -207,11 +198,12 @@ async fn test_system_state_transitions() {
 }
 
 #[test]
-fn test_event_name_mask_performance() {
+fn test_event_name_mask_matches_expansion() {
     use rustfs_targets::EventName;
 
-    // Test that event name mask calculation is efficient
-    let events = vec![
+    // A compound event's mask is exactly the union of its leaves' masks, and a
+    // leaf owns exactly one bit. Timing is printed for information only.
+    let events = [
         EventName::ObjectCreatedPut,
         EventName::ObjectAccessedGet,
         EventName::ObjectRemovedDelete,
@@ -220,47 +212,59 @@ fn test_event_name_mask_performance() {
     ];
 
     let start = Instant::now();
-
-    // Calculate masks for many events
-    for _ in 0..1000 {
-        for event in &events {
-            let _mask = event.mask();
-        }
+    for event in events {
+        let union = event.expand().iter().fold(0u64, |acc, leaf| acc | leaf.mask());
+        assert_eq!(event.mask(), union, "{event:?} mask must equal the union of its expansion");
     }
+    println!("Event mask check took: {:?}", start.elapsed());
 
-    let elapsed = start.elapsed();
-    println!("Event mask calculation (5000 ops) took: {elapsed:?}");
-
-    // Should be very fast
-    assert!(elapsed < Duration::from_millis(100), "Mask calculation too slow: {elapsed:?}");
+    assert_eq!(EventName::ObjectCreatedPut.mask().count_ones(), 1);
+    assert_eq!(EventName::ObjectCreatedAll.mask().count_ones(), 4);
 }
 
 #[test]
 fn test_event_name_expansion_performance() {
     use rustfs_targets::EventName;
+    use std::collections::HashSet;
 
-    // Test that event name expansion is efficient
-    let compound_events = vec![
-        EventName::ObjectCreatedAll,
-        EventName::ObjectAccessedAll,
-        EventName::ObjectRemovedAll,
-        EventName::Everything,
+    // Expansion is a pure lookup. Wall-clock bounds flake on shared runners, so
+    // this asserts what the time was a proxy for: each compound event yields a
+    // fixed-size set of distinct leaf events and nothing else (no duplicates,
+    // no compound left to expand again). Timing is printed for information only.
+    let cases = [
+        (EventName::ObjectCreatedAll, 4),
+        (EventName::ObjectAccessedAll, 5),
+        (EventName::ObjectRemovedAll, 3),
     ];
 
     let start = Instant::now();
-
-    // Expand events many times
-    for _ in 0..1000 {
-        for event in &compound_events {
-            let _expanded = event.expand();
+    for (event, expected_len) in cases {
+        let expanded = event.expand();
+        assert_eq!(expanded.len(), expected_len, "{event:?} expansion size");
+        assert_eq!(
+            expanded.iter().collect::<HashSet<_>>().len(),
+            expanded.len(),
+            "{event:?} expansion has duplicates"
+        );
+        assert_eq!(expanded, event.expand(), "{event:?} expansion must be deterministic");
+        for leaf in &expanded {
+            assert_eq!(leaf.expand(), vec![*leaf], "{event:?} expanded to non-leaf {leaf:?}");
         }
     }
 
-    let elapsed = start.elapsed();
-    println!("Event expansion (4000 ops) took: {elapsed:?}");
-
-    // Should be very fast
-    assert!(elapsed < Duration::from_millis(100), "Expansion too slow: {elapsed:?}");
+    let everything = EventName::Everything.expand();
+    assert!(everything.len() > 12, "Everything must cover at least every compound's leaves");
+    assert_eq!(
+        everything.iter().collect::<HashSet<_>>().len(),
+        everything.len(),
+        "Everything has duplicates"
+    );
+    for (event, _) in cases {
+        for leaf in event.expand() {
+            assert!(everything.contains(&leaf), "Everything is missing {leaf:?}");
+        }
+    }
+    println!("Event expansion check took: {:?}", start.elapsed());
 }
 
 #[tokio::test]
@@ -278,46 +282,22 @@ async fn test_registry_operations_performance() {
 
     let elapsed = start.elapsed();
     println!("Registry operations (2000 ops) took: {elapsed:?}");
-
-    // Should be very fast for empty registry
-    assert!(elapsed < Duration::from_millis(100), "Registry ops too slow: {elapsed:?}");
 }
 
-// Performance requirements validation
+// The former 3k EPS / latency assertions measured the CI host, not the code
+// (the loop only formats strings and reads the clock), so they are replaced by
+// a count of what the loop produced. Throughput is printed for information.
 #[test]
 fn test_performance_requirements() {
-    // According to requirements: ≥ 3k EPS/node; P99 < 30ms (default)
-
-    // These are synthetic tests since we can't actually achieve 3k EPS
-    // without real server storage and network targets, but we can validate
-    // that our core algorithms are efficient enough
+    use std::collections::HashSet;
 
     let start = Instant::now();
-
-    // Simulate processing 3000 events worth of operations
-    for i in 0..3000 {
-        // Simulate event name parsing and processing
-        let _event_id = format!("s3:ObjectCreated:Put_{i}");
-        let _timestamp = jiff::Timestamp::now().to_string();
-
-        // Simulate basic audit entry creation overhead
-        let _entry_size = 512; // bytes
-        let _processing_time = std::time::Duration::from_nanos(100); // simulated
-    }
-
+    let ids: HashSet<String> = (0..3000).map(|i| format!("s3:ObjectCreated:Put_{i}")).collect();
+    let timestamps = (0..3000).filter(|_| !jiff::Timestamp::now().to_string().is_empty()).count();
     let elapsed = start.elapsed();
-    let eps = 3000.0 / elapsed.as_secs_f64();
 
-    println!("Simulated 3000 events in {elapsed:?} ({eps:.0} EPS)");
+    println!("Simulated 3000 events in {elapsed:?}");
 
-    // Our core processing should easily handle 3k EPS worth of CPU overhead
-    // The actual EPS limit will be determined by network I/O to targets
-    assert!(eps > 10000.0, "Core processing too slow for 3k EPS target: {eps} EPS");
-
-    // P99 latency requirement: < 30ms
-    // For core processing, we should be much faster than this
-    let avg_latency = elapsed / 3000;
-    println!("Average processing latency: {avg_latency:?}");
-
-    assert!(avg_latency < Duration::from_millis(1), "Processing latency too high: {avg_latency:?}");
+    assert_eq!(ids.len(), 3000, "every simulated event id must be distinct");
+    assert_eq!(timestamps, 3000, "every simulated event must get a timestamp");
 }

crates/e2e_test/src/sts_query_compat_test.rs

@@ -395,6 +395,56 @@ impl Drop for OpaMock {
     }
 }
 
+#[tokio::test]
+async fn test_sts_session_policy_id_limits_object_access() -> TestResult {
+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server(vec![]).await?;
+    let admin = env.create_s3_client();
+    let bucket = "session-policy-id";
+    admin.create_bucket().bucket(bucket).send().await?;
+    for key in ["allowed", "denied"] {
+        admin
+            .put_object()
+            .bucket(bucket)
+            .key(key)
+            .body(Bytes::from_static(b"data").into())
+            .send()
+            .await?;
+    }
+
+    let policy = serde_json::json!({
+        "Id": "read-one-object",
+        "Version": "2012-10-17",
+        "Statement": [{"Effect": "Allow", "Action": "s3:GetObject", "Resource": format!("arn:aws:s3:::{bucket}/allowed")}]
+    });
+    let response = sts_client(&env.url, &env.access_key, &env.secret_key, None)
+        .assume_role()
+        .role_arn("arn:aws:iam::123456789012:role/test")
+        .role_session_name("session-policy-id")
+        .policy(policy.to_string())
+        .send()
+        .await?;
+    let credentials = response.credentials().ok_or("AssumeRole returned no credentials")?;
+    let session = aws_sdk_s3::Client::from_conf(build_test_s3_config(
+        &env.url,
+        credentials.access_key_id(),
+        credentials.secret_access_key(),
+        Some(credentials.session_token()),
+        "e2e-session-policy-id",
+    ));
+    let allowed = session.get_object().bucket(bucket).key("allowed").send().await?;
+    assert_eq!(allowed.body.collect().await?.into_bytes().as_ref(), b"data");
+    let error = session
+        .get_object()
+        .bucket(bucket)
+        .key("denied")
+        .send()
+        .await
+        .expect_err("object outside session policy");
+    assert_eq!(error.as_service_error().and_then(ProvideErrorMetadata::code), Some("AccessDenied"));
+    Ok(())
+}
+
 #[tokio::test]
 async fn test_sts_query_responses_are_aws_sdk_compatible() -> TestResult {
     init_logging();

crates/ecstore/src/disk/local.rs

@@ -559,18 +559,264 @@ fn remove_dir_all_if_exists(path: &Path) -> std::io::Result<()> {
     }
 }
 
-fn snapshot_part_transaction_file(src: &Path, backup: &Path, absent: &Path) -> std::io::Result<bool> {
+#[expect(clippy::too_many_arguments)]
+fn snapshot_part_transaction_file(
+    src: &Path,
+    src_parent_fd: Option<&std::fs::File>,
+    backup: &Path,
+    txn_dir_fd: Option<&std::fs::File>,
+    backup_name: &str,
+    absent: &Path,
+    absent_name: &str,
+    link_absent_from: Option<(&str, &Path)>,
+) -> std::io::Result<bool> {
+    #[cfg(unix)]
+    if let (Some(p_fd), Some(d_fd), Some(src_name)) = (src_parent_fd, txn_dir_fd, src.file_name().and_then(|n| n.to_str())) {
+        use rustix::fs::{AtFlags, Mode, OFlags, linkat, openat, statat};
+        return match statat(p_fd, src_name, AtFlags::SYMLINK_NOFOLLOW).map_err(std::io::Error::from) {
+            Ok(st) if (st.st_mode & libc::S_IFMT) == libc::S_IFREG => linkat(p_fd, src_name, d_fd, backup_name, AtFlags::empty())
+                .map(|()| true)
+                .map_err(std::io::Error::from),
+            Ok(st) if (st.st_mode & libc::S_IFMT) == libc::S_IFLNK => {
+                Err(std::io::Error::new(ErrorKind::InvalidInput, "multipart transaction source is a symlink"))
+            }
+            Ok(_) => Err(std::io::Error::new(ErrorKind::InvalidData, "multipart transaction source is not a file")),
+            Err(err) if err.kind() == ErrorKind::NotFound => {
+                if let Some((from_name, _)) = link_absent_from
+                    && linkat(d_fd, from_name, d_fd, absent_name, AtFlags::empty()).is_ok()
+                {
+                    return Ok(false);
+                }
+                openat(
+                    d_fd,
+                    absent_name,
+                    OFlags::WRONLY | OFlags::CREATE | OFlags::TRUNC | OFlags::CLOEXEC,
+                    Mode::from_raw_mode(0o666),
+                )
+                .map(|_| false)
+                .map_err(std::io::Error::from)
+            }
+            Err(err) => Err(err),
+        };
+    }
+    #[cfg(not(unix))]
+    let _ = (src_parent_fd, txn_dir_fd, backup_name, absent_name);
     match std::fs::symlink_metadata(src) {
         Ok(metadata) if metadata.is_file() => std::fs::hard_link(src, backup).map(|()| true),
         Ok(metadata) if metadata.file_type().is_symlink() => {
             Err(std::io::Error::new(ErrorKind::InvalidInput, "multipart transaction source is a symlink"))
         }
         Ok(_) => Err(std::io::Error::new(ErrorKind::InvalidData, "multipart transaction source is not a file")),
-        Err(err) if err.kind() == ErrorKind::NotFound => std::fs::write(absent, []).map(|()| false),
+        Err(err) if err.kind() == ErrorKind::NotFound => {
+            if let Some((_, from_path)) = link_absent_from
+                && std::fs::hard_link(from_path, absent).is_ok()
+            {
+                return Ok(false);
+            }
+            std::fs::write(absent, []).map(|()| false)
+        }
         Err(err) => Err(err),
     }
 }
 
+#[cfg(unix)]
+fn open_dir_fd(path: &Path) -> std::io::Result<std::fs::File> {
+    use rustix::fs::{Mode, OFlags, open};
+    open(path, OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC, Mode::empty())
+        .map(std::fs::File::from)
+        .map_err(std::io::Error::from)
+}
+
+#[cfg(unix)]
+fn mkdir_and_open_part_txn_dir(
+    parent_fd: Option<&std::fs::File>,
+    dir_name: &str,
+    dir_path: &Path,
+) -> std::io::Result<Option<Arc<std::fs::File>>> {
+    if let Some(p_fd) = parent_fd {
+        use rustix::fs::{Mode, OFlags, mkdirat, openat};
+        mkdirat(p_fd, dir_name, Mode::from_raw_mode(0o777)).map_err(std::io::Error::from)?;
+        let fd = openat(p_fd, dir_name, OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC, Mode::empty())
+            .ok()
+            .map(std::fs::File::from)
+            .map(Arc::new);
+        return Ok(fd);
+    }
+    std::fs::create_dir(dir_path)?;
+    Ok(open_dir_fd(dir_path).ok().map(Arc::new))
+}
+
+#[cfg(not(unix))]
+fn mkdir_and_open_part_txn_dir(
+    _parent_fd: Option<&std::fs::File>,
+    _dir_name: &str,
+    dir_path: &Path,
+) -> std::io::Result<Option<Arc<std::fs::File>>> {
+    std::fs::create_dir(dir_path)?;
+    Ok(None)
+}
+
+#[cfg(unix)]
+fn link_in_same_dir_fd(
+    dir_fd: Option<&std::fs::File>,
+    src_name: &str,
+    src_fallback: &Path,
+    dst_name: &str,
+    dst_fallback: &Path,
+) -> std::io::Result<()> {
+    if let Some(d_fd) = dir_fd {
+        return rustix::fs::linkat(d_fd, src_name, d_fd, dst_name, rustix::fs::AtFlags::empty()).map_err(std::io::Error::from);
+    }
+    std::fs::hard_link(src_fallback, dst_fallback)
+}
+
+#[cfg(not(unix))]
+fn link_in_same_dir_fd(
+    _dir_fd: Option<&std::fs::File>,
+    _src_name: &str,
+    src_fallback: &Path,
+    _dst_name: &str,
+    dst_fallback: &Path,
+) -> std::io::Result<()> {
+    std::fs::hard_link(src_fallback, dst_fallback)
+}
+
+#[cfg(unix)]
+fn create_new_in_dir_fd(dir_fd: Option<&std::fs::File>, name: &str, fallback_path: &Path) -> std::io::Result<std::fs::File> {
+    if let Some(d_fd) = dir_fd {
+        use rustix::fs::{Mode, OFlags, openat};
+        let fd = openat(
+            d_fd,
+            name,
+            OFlags::WRONLY | OFlags::CREATE | OFlags::EXCL | OFlags::CLOEXEC,
+            Mode::from_raw_mode(0o666),
+        )
+        .map_err(std::io::Error::from)?;
+        return Ok(std::fs::File::from(fd));
+    }
+    std::fs::OpenOptions::new().create_new(true).write(true).open(fallback_path)
+}
+
+#[cfg(not(unix))]
+fn create_new_in_dir_fd(_dir_fd: Option<&std::fs::File>, _name: &str, fallback_path: &Path) -> std::io::Result<std::fs::File> {
+    std::fs::OpenOptions::new().create_new(true).write(true).open(fallback_path)
+}
+
+#[cfg(unix)]
+fn rename_in_same_dir_fd(
+    parent_fd: Option<&std::fs::File>,
+    src_name: Option<&str>,
+    src_fallback: &Path,
+    dst_name: Option<&str>,
+    dst_fallback: &Path,
+) -> std::io::Result<()> {
+    if let (Some(p_fd), Some(s_name), Some(d_name)) = (parent_fd, src_name, dst_name) {
+        return rustix::fs::renameat(p_fd, s_name, p_fd, d_name).map_err(std::io::Error::from);
+    }
+    std::fs::rename(src_fallback, dst_fallback)
+}
+
+#[cfg(not(unix))]
+fn rename_in_same_dir_fd(
+    _parent_fd: Option<&std::fs::File>,
+    _src_name: Option<&str>,
+    src_fallback: &Path,
+    _dst_name: Option<&str>,
+    dst_fallback: &Path,
+) -> std::io::Result<()> {
+    std::fs::rename(src_fallback, dst_fallback)
+}
+
+#[cfg(unix)]
+fn part_txn_files_ready(
+    txn_dir_fd: Option<&std::fs::File>,
+    publish_meta_path: &Path,
+    transaction_meta_path: &Path,
+    expected_len: u64,
+) -> bool {
+    if let Some(d_fd) = txn_dir_fd {
+        use rustix::fs::{AtFlags, statat};
+        let pub_ok = statat(d_fd, PART_TRANSACTION_PUBLISH_META, AtFlags::SYMLINK_NOFOLLOW)
+            .is_ok_and(|st| (st.st_mode & libc::S_IFMT) == libc::S_IFREG);
+        let new_ok = statat(d_fd, PART_TRANSACTION_NEW_META, AtFlags::SYMLINK_NOFOLLOW)
+            .is_ok_and(|st| (st.st_mode & libc::S_IFMT) == libc::S_IFREG && st.st_size as u64 == expected_len);
+        return pub_ok && new_ok;
+    }
+    publish_meta_path.is_file() && std::fs::metadata(transaction_meta_path).is_ok_and(|m| m.is_file() && m.len() == expected_len)
+}
+
+#[cfg(not(unix))]
+fn part_txn_files_ready(
+    _txn_dir_fd: Option<&std::fs::File>,
+    publish_meta_path: &Path,
+    transaction_meta_path: &Path,
+    expected_len: u64,
+) -> bool {
+    publish_meta_path.is_file() && std::fs::metadata(transaction_meta_path).is_ok_and(|m| m.is_file() && m.len() == expected_len)
+}
+
+#[cfg(unix)]
+fn check_part_txn_dir_in_parent_fd(
+    parent_fd: Option<&std::fs::File>,
+    dir_name: Option<&str>,
+    fallback_path: &Path,
+) -> std::io::Result<Result<bool>> {
+    if let (Some(p_fd), Some(name)) = (parent_fd, dir_name) {
+        return match rustix::fs::statat(p_fd, name, rustix::fs::AtFlags::SYMLINK_NOFOLLOW).map_err(std::io::Error::from) {
+            Ok(st) if (st.st_mode & libc::S_IFMT) == libc::S_IFDIR => Ok(Ok(true)),
+            Ok(_) => Ok(Err(DiskError::FileCorrupt)),
+            Err(err) if err.kind() == ErrorKind::NotFound => Ok(Ok(false)),
+            Err(err) => Ok(Err(to_file_error(err).into())),
+        };
+    }
+    match std::fs::symlink_metadata(fallback_path) {
+        Ok(metadata) if metadata.is_dir() => Ok(Ok(true)),
+        Ok(_) => Ok(Err(DiskError::FileCorrupt)),
+        Err(err) if err.kind() == ErrorKind::NotFound => Ok(Ok(false)),
+        Err(err) => Ok(Err(to_file_error(err).into())),
+    }
+}
+
+#[cfg(not(unix))]
+fn check_part_txn_dir_in_parent_fd(
+    _parent_fd: Option<&std::fs::File>,
+    _dir_name: Option<&str>,
+    fallback_path: &Path,
+) -> std::io::Result<Result<bool>> {
+    match std::fs::symlink_metadata(fallback_path) {
+        Ok(metadata) if metadata.is_dir() => Ok(Ok(true)),
+        Ok(_) => Ok(Err(DiskError::FileCorrupt)),
+        Err(err) if err.kind() == ErrorKind::NotFound => Ok(Ok(false)),
+        Err(err) => Ok(Err(to_file_error(err).into())),
+    }
+}
+
+#[cfg(unix)]
+fn unlink_in_dir_fd(dir_fd: Option<&std::fs::File>, name: &str, fallback_path: &Path) -> std::io::Result<()> {
+    if let Some(d_fd) = dir_fd {
+        return rustix::fs::unlinkat(d_fd, name, rustix::fs::AtFlags::empty()).map_err(std::io::Error::from);
+    }
+    std::fs::remove_file(fallback_path)
+}
+
+#[cfg(not(unix))]
+fn unlink_in_dir_fd(_dir_fd: Option<&std::fs::File>, _name: &str, fallback_path: &Path) -> std::io::Result<()> {
+    std::fs::remove_file(fallback_path)
+}
+
+#[cfg(unix)]
+fn rmdir_in_dir_fd(dir_fd: Option<&std::fs::File>, name: &str, fallback_path: &Path) -> std::io::Result<()> {
+    if let Some(d_fd) = dir_fd {
+        return rustix::fs::unlinkat(d_fd, name, rustix::fs::AtFlags::REMOVEDIR).map_err(std::io::Error::from);
+    }
+    std::fs::remove_dir(fallback_path)
+}
+
+#[cfg(not(unix))]
+fn rmdir_in_dir_fd(_dir_fd: Option<&std::fs::File>, _name: &str, fallback_path: &Path) -> std::io::Result<()> {
+    std::fs::remove_dir(fallback_path)
+}
+
 struct PreparedPartTxnReceipt {
     src_volume_dir: PathBuf,
     dst_volume_dir: PathBuf,
@@ -581,6 +827,8 @@ struct PreparedPartTxnReceipt {
     meta: Bytes,
     had_old_data: bool,
     had_old_meta: bool,
+    txn_dir_fd: Option<Arc<std::fs::File>>,
+    txn_parent_fd: Option<Arc<std::fs::File>>,
 }
 
 struct PublishedPartTxnReceipt {
@@ -589,6 +837,8 @@ struct PublishedPartTxnReceipt {
     transaction_path: PathBuf,
     had_old_data: bool,
     had_old_meta: bool,
+    txn_dir_fd: Option<Arc<std::fs::File>>,
+    txn_parent_fd: Option<Arc<std::fs::File>>,
 }
 
 const PART_TXN_RECEIPTS_PER_SHARD: usize = 64;
@@ -615,6 +865,13 @@ fn take_published_part_txn_receipt(transaction_path: &Path) -> Option<PublishedP
     PUBLISHED_PART_TXN_RECEIPTS.remove(transaction_path)
 }
 
+fn clear_part_txn_receipts_under(io_root: &Path, root: &Path) {
+    for prefix in [io_root, root] {
+        PREPARED_PART_TXN_RECEIPTS.retain_outside_prefix(prefix);
+        PUBLISHED_PART_TXN_RECEIPTS.retain_outside_prefix(prefix);
+    }
+}
+
 fn restore_part_transaction_file(current: &Path, backup: &Path, absent: &Path, restore: &Path) -> std::io::Result<()> {
     os::record_native_publication(current);
     match std::fs::symlink_metadata(backup) {
@@ -7829,6 +8086,9 @@ impl Drop for SnapshotLeaseDeleteClaim {
 
 impl Drop for LocalDisk {
     fn drop(&mut self) {
+        let io_root = self.io_root();
+        clear_part_txn_receipts_under(io_root, &self.root);
+        delete_recovery::clear_delete_receipts_under(io_root, &self.root);
         if let Some(exit_signal) = self.exit_signal.take() {
             let _ = exit_signal.send(());
         }
@@ -14147,6 +14407,9 @@ impl DiskAPI for LocalDisk {
         if let Some(history) = self.publication_history.as_ref() {
             history.invalidate();
         }
+        let io_root = self.io_root();
+        clear_part_txn_receipts_under(io_root, &self.root);
+        delete_recovery::clear_delete_receipts_under(io_root, &self.root);
         // This disk instance is being retired (e.g. replaced by renew_disk on
         // reconnect). Drop its cached descriptors so a replacement instance's
         // invalidations are never defeated by this one continuing to serve stale
@@ -14765,175 +15028,196 @@ impl DiskAPI for LocalDisk {
         }
         os::record_native_publication(&transaction_path);
         let lease = os::acquire_namespace_mutation_lease_with_owner(&transaction_path, source_owner).await;
-        let res = os::run_blocking_namespace_operation_with_sync(lease, sync, move || -> std::io::Result<Result<()>> {
-            if let Some(ref admission) = admission
-                && let Err(err) = admission.validate()
-            {
-                return Ok(Err(to_file_error(err).into()));
-            }
-            let source = match std::fs::symlink_metadata(&src_file_path) {
-                Ok(source) => source,
-                Err(err) => return Ok(Err(to_file_error(err).into())),
-            };
-            if !source.is_file() {
-                return Ok(Err(DiskError::FileAccessDenied));
-            }
-            match std::fs::symlink_metadata(&transaction_path) {
-                Err(err) if err.kind() == ErrorKind::NotFound => {}
-                Ok(m) if m.file_type().is_symlink() => return Ok(Err(DiskError::InvalidPath)),
-                Ok(_) => return Ok(Err(DiskError::FileAccessDenied)),
-                Err(err) => return Ok(Err(to_file_error(err).into())),
-            }
-            let mut created_transaction_parent = false;
-            if let Some((integrity, source, destination, expected_len)) = integrity_prep {
-                let stat = match std::fs::symlink_metadata(&source) {
-                    Ok(stat) => stat,
-                    Err(err) => return Ok(Err(to_file_error(err).into())),
-                };
-                if !stat.is_file() || stat.len() != expected_len {
-                    return Ok(Err(DiskError::FileCorrupt));
-                }
-                let presynced_proof = take_presynced_tmp_file(&source);
-                let verify_and_sync = (|| -> std::io::Result<bool> {
-                    let mut file = os::open_file_for_sync_std(&source)?;
-                    let mut header = [0u8; rustfs_filemeta::shard_integrity::INDEX_HEADER_SIZE];
-                    std::io::Read::read_exact(&mut file, &mut header)?;
-                    if header != integrity.index_header() {
-                        return Ok(false);
-                    }
-                    if durability.syncs_data_shards() && !presynced_proof {
-                        file.sync_all()?;
-                    }
-                    Ok(true)
-                })();
-                match verify_and_sync {
-                    Ok(true) => {}
-                    Ok(false) => return Ok(Err(DiskError::FileCorrupt)),
-                    Err(err) => return Ok(Err(to_file_error(err).into())),
-                }
-                #[cfg(all(any(test, feature = "test-util"), not(windows)))]
-                let has_hooks = os::prepared_publication_test_hooks::has_any_hooks();
-                #[cfg(not(all(any(test, feature = "test-util"), not(windows))))]
-                let has_hooks = false;
-                let precondition = has_hooks
-                    .then(|| admission.as_ref().map(|check| check.precondition()))
-                    .flatten();
-                // Publish the immutable generation before preparing the part switch.
-                // Rollback retains the old generation; an unreferenced new index is
-                // reclaimed with the upload directory if preparation is interrupted.
-                // Defer the parent directory fsync to the transaction_parent fsync
-                // below when both live in the same parent directory.
-                let same_parent = destination.parent() == transaction_path.parent();
-                if let Err(err) = os::rename_all_with_precondition_and_parent_sync_std(
-                    &source,
-                    &destination,
-                    &dst_volume_dir,
-                    &publication_root,
-                    precondition.as_deref(),
-                    durability.syncs_commit_metadata() && !same_parent,
-                ) {
+        let res = os::run_blocking_namespace_operation_with_sync(
+            lease,
+            sync,
+            move || -> std::io::Result<Result<(PathBuf, PreparedPartTxnReceipt)>> {
+                if let Some(ref admission) = admission
+                    && let Err(err) = admission.validate()
+                {
                     return Ok(Err(to_file_error(err).into()));
                 }
-                created_transaction_parent = same_parent;
-            }
-
-            let Some(transaction_parent) = transaction_path.parent() else {
-                return Ok(Err(DiskError::InvalidPath));
-            };
-            if !created_transaction_parent && let Err(err) = std::fs::create_dir_all(transaction_parent) {
-                return Ok(Err(to_file_error(err).into()));
-            }
-            let staging_path = transaction_parent.join(format!(".part-txn-{}", Uuid::new_v4()));
-            if let Err(err) = std::fs::create_dir(&staging_path) {
-                return Ok(Err(to_file_error(err).into()));
-            }
-
-            let prepare_result = (|| -> std::io::Result<(bool, bool)> {
-                let old_data_absent = staging_path.join(PART_TRANSACTION_OLD_DATA_ABSENT);
-                let old_meta_absent = staging_path.join(PART_TRANSACTION_OLD_META_ABSENT);
-                let had_old_data = snapshot_part_transaction_file(
-                    &dst_file_path,
-                    &staging_path.join(PART_TRANSACTION_OLD_DATA),
-                    &old_data_absent,
-                )?;
-                let had_old_meta = match std::fs::symlink_metadata(&dst_meta_path) {
-                    Ok(metadata) if metadata.is_file() => {
-                        std::fs::hard_link(&dst_meta_path, staging_path.join(PART_TRANSACTION_OLD_META))?;
-                        true
+                let source = match std::fs::symlink_metadata(&src_file_path) {
+                    Ok(source) => source,
+                    Err(err) => return Ok(Err(to_file_error(err).into())),
+                };
+                if !source.is_file() {
+                    return Ok(Err(DiskError::FileAccessDenied));
+                }
+                match std::fs::symlink_metadata(&transaction_path) {
+                    Err(err) if err.kind() == ErrorKind::NotFound => {}
+                    Ok(m) if m.file_type().is_symlink() => return Ok(Err(DiskError::InvalidPath)),
+                    Ok(_) => return Ok(Err(DiskError::FileAccessDenied)),
+                    Err(err) => return Ok(Err(to_file_error(err).into())),
+                }
+                let mut created_transaction_parent = false;
+                if let Some((integrity, source, destination, expected_len)) = integrity_prep {
+                    let stat = match std::fs::symlink_metadata(&source) {
+                        Ok(stat) => stat,
+                        Err(err) => return Ok(Err(to_file_error(err).into())),
+                    };
+                    if !stat.is_file() || stat.len() != expected_len {
+                        return Ok(Err(DiskError::FileCorrupt));
                     }
-                    Ok(metadata) if metadata.file_type().is_symlink() => {
-                        return Err(std::io::Error::new(ErrorKind::InvalidInput, "multipart transaction source is a symlink"));
-                    }
-                    Ok(_) => {
-                        return Err(std::io::Error::new(ErrorKind::InvalidData, "multipart transaction source is not a file"));
-                    }
-                    Err(err) if err.kind() == ErrorKind::NotFound => {
-                        if !had_old_data && std::fs::hard_link(&old_data_absent, &old_meta_absent).is_ok() {
-                            false
-                        } else {
-                            std::fs::write(&old_meta_absent, [])?;
-                            false
+                    let presynced_proof = take_presynced_tmp_file(&source);
+                    let verify_and_sync = (|| -> std::io::Result<bool> {
+                        let mut file = os::open_file_for_sync_std(&source)?;
+                        let mut header = [0u8; rustfs_filemeta::shard_integrity::INDEX_HEADER_SIZE];
+                        std::io::Read::read_exact(&mut file, &mut header)?;
+                        if header != integrity.index_header() {
+                            return Ok(false);
                         }
+                        if durability.syncs_data_shards() && !presynced_proof {
+                            file.sync_all()?;
+                        }
+                        Ok(true)
+                    })();
+                    match verify_and_sync {
+                        Ok(true) => {}
+                        Ok(false) => return Ok(Err(DiskError::FileCorrupt)),
+                        Err(err) => return Ok(Err(to_file_error(err).into())),
                     }
-                    Err(err) => return Err(err),
+                    #[cfg(all(any(test, feature = "test-util"), not(windows)))]
+                    let has_hooks = os::prepared_publication_test_hooks::has_any_hooks();
+                    #[cfg(not(all(any(test, feature = "test-util"), not(windows))))]
+                    let has_hooks = false;
+                    let precondition = has_hooks
+                        .then(|| admission.as_ref().map(|check| check.precondition()))
+                        .flatten();
+                    // Publish the immutable generation before preparing the part switch.
+                    // Rollback retains the old generation; an unreferenced new index is
+                    // reclaimed with the upload directory if preparation is interrupted.
+                    // Defer the parent directory fsync to the transaction_parent fsync
+                    // below when both live in the same parent directory.
+                    let same_parent = destination.parent() == transaction_path.parent();
+                    if let Err(err) = os::rename_all_with_precondition_and_parent_sync_std(
+                        &source,
+                        &destination,
+                        &dst_volume_dir,
+                        &publication_root,
+                        precondition.as_deref(),
+                        durability.syncs_commit_metadata() && !same_parent,
+                    ) {
+                        return Ok(Err(to_file_error(err).into()));
+                    }
+                    created_transaction_parent = same_parent;
+                }
+
+                let Some(transaction_parent) = transaction_path.parent() else {
+                    return Ok(Err(DiskError::InvalidPath));
                 };
-
-                let new_meta_path = staging_path.join(PART_TRANSACTION_NEW_META);
-                let mut new_meta = std::fs::OpenOptions::new()
-                    .create_new(true)
-                    .write(true)
-                    .open(&new_meta_path)?;
-                std::io::Write::write_all(&mut new_meta, &meta)?;
-                if durability.syncs_commit_metadata() {
-                    new_meta.sync_data()?;
-                }
-                // Windows rejects renaming a directory while one of its children is
-                // still open, even when the child handle shares delete access.
-                drop(new_meta);
-                let _ = std::fs::hard_link(&new_meta_path, staging_path.join(PART_TRANSACTION_PUBLISH_META));
-                let sync_overwrite_backup = durability.syncs_commit_metadata() && (had_old_data || had_old_meta);
-                if sync_overwrite_backup {
-                    os::fsync_dir_std(&staging_path)?;
-                }
-                std::fs::rename(&staging_path, &transaction_path)?;
-                if sync_overwrite_backup {
-                    os::fsync_dir_std(transaction_parent)?;
-                }
-                Ok((had_old_data, had_old_meta))
-            })();
-
-            let (had_old_data, had_old_meta) = match prepare_result {
-                Ok(flags) => flags,
-                Err(err) => {
-                    let _ = remove_dir_all_if_exists(&staging_path);
+                if !created_transaction_parent && let Err(err) = std::fs::create_dir_all(transaction_parent) {
                     return Ok(Err(to_file_error(err).into()));
                 }
-            };
-            put_prepared_part_txn_receipt(
-                &transaction_path,
-                PreparedPartTxnReceipt {
-                    src_volume_dir,
-                    dst_volume_dir,
-                    src_file_path,
-                    dst_file_path,
-                    dst_meta_path,
-                    transaction_path: transaction_path.clone(),
-                    meta,
-                    had_old_data,
-                    had_old_meta,
-                },
-            );
+                #[cfg(unix)]
+                let txn_parent_fd = open_dir_fd(transaction_parent).ok().map(Arc::new);
+                #[cfg(not(unix))]
+                let txn_parent_fd: Option<Arc<std::fs::File>> = None;
+                let staging_name = format!(".part-txn-{}", Uuid::new_v4());
+                let staging_path = transaction_parent.join(&staging_name);
+                let txn_dir_fd = match mkdir_and_open_part_txn_dir(txn_parent_fd.as_deref(), &staging_name, &staging_path) {
+                    Ok(fd) => fd,
+                    Err(err) => return Ok(Err(to_file_error(err).into())),
+                };
 
-            Ok(Ok(()))
-        })
+                let prepare_result = (|| -> std::io::Result<(bool, bool)> {
+                    let old_data_absent = staging_path.join(PART_TRANSACTION_OLD_DATA_ABSENT);
+                    let old_meta_absent = staging_path.join(PART_TRANSACTION_OLD_META_ABSENT);
+                    let had_old_data = snapshot_part_transaction_file(
+                        &dst_file_path,
+                        txn_parent_fd.as_deref(),
+                        &staging_path.join(PART_TRANSACTION_OLD_DATA),
+                        txn_dir_fd.as_deref(),
+                        PART_TRANSACTION_OLD_DATA,
+                        &old_data_absent,
+                        PART_TRANSACTION_OLD_DATA_ABSENT,
+                        None,
+                    )?;
+                    let had_old_meta = snapshot_part_transaction_file(
+                        &dst_meta_path,
+                        txn_parent_fd.as_deref(),
+                        &staging_path.join(PART_TRANSACTION_OLD_META),
+                        txn_dir_fd.as_deref(),
+                        PART_TRANSACTION_OLD_META,
+                        &old_meta_absent,
+                        PART_TRANSACTION_OLD_META_ABSENT,
+                        (!had_old_data).then_some((PART_TRANSACTION_OLD_DATA_ABSENT, old_data_absent.as_path())),
+                    )?;
+
+                    let new_meta_path = staging_path.join(PART_TRANSACTION_NEW_META);
+                    let mut new_meta = create_new_in_dir_fd(txn_dir_fd.as_deref(), PART_TRANSACTION_NEW_META, &new_meta_path)?;
+                    std::io::Write::write_all(&mut new_meta, &meta)?;
+                    if durability.syncs_commit_metadata() {
+                        new_meta.sync_data()?;
+                    }
+                    // Windows rejects renaming a directory while one of its children is
+                    // still open, even when the child handle shares delete access.
+                    drop(new_meta);
+                    let _ = link_in_same_dir_fd(
+                        txn_dir_fd.as_deref(),
+                        PART_TRANSACTION_NEW_META,
+                        &new_meta_path,
+                        PART_TRANSACTION_PUBLISH_META,
+                        &staging_path.join(PART_TRANSACTION_PUBLISH_META),
+                    );
+                    let sync_overwrite_backup = durability.syncs_commit_metadata() && (had_old_data || had_old_meta);
+                    if sync_overwrite_backup {
+                        os::fsync_directory_handle_std(&staging_path, txn_dir_fd.as_deref())?;
+                    }
+                    rename_in_same_dir_fd(
+                        txn_parent_fd.as_deref(),
+                        Some(&staging_name),
+                        &staging_path,
+                        transaction_path.file_name().and_then(|n| n.to_str()),
+                        &transaction_path,
+                    )?;
+                    if sync_overwrite_backup {
+                        os::fsync_directory_handle_std(transaction_parent, txn_parent_fd.as_deref())?;
+                    }
+                    Ok((had_old_data, had_old_meta))
+                })();
+
+                let (had_old_data, had_old_meta) = match prepare_result {
+                    Ok(flags) => flags,
+                    Err(err) => {
+                        drop(txn_dir_fd);
+                        drop(txn_parent_fd);
+                        let _ = remove_dir_all_if_exists(&staging_path);
+                        return Ok(Err(to_file_error(err).into()));
+                    }
+                };
+                Ok(Ok((
+                    transaction_path.clone(),
+                    PreparedPartTxnReceipt {
+                        src_volume_dir,
+                        dst_volume_dir,
+                        src_file_path,
+                        dst_file_path,
+                        dst_meta_path,
+                        transaction_path,
+                        meta,
+                        had_old_data,
+                        had_old_meta,
+                        txn_dir_fd,
+                        txn_parent_fd,
+                    },
+                )))
+            },
+        )
         .await
         .map_err(to_file_error)?;
-        if res.is_err()
-            && let Some(ref source) = admission_source
-        {
-            multipart_source_state::invalidate_active_multipart_source_head(source);
+        match res {
+            Ok((txn_path, entry)) => {
+                put_prepared_part_txn_receipt(&txn_path, entry);
+                Ok(())
+            }
+            Err(err) => {
+                if let Some(ref source) = admission_source {
+                    multipart_source_state::invalidate_active_multipart_source_head(source);
+                }
+                Err(err)
+            }
         }
-        res
     }
 
     #[tracing::instrument(level = "trace", skip_all)]
@@ -14942,29 +15226,41 @@ impl DiskAPI for LocalDisk {
         let admission = self.multipart_source_admission(volume, path, true, false)?;
         let admission_source = admission.as_ref().map(|a| a.source().to_path_buf());
         let txn_rel_path = crate::disk::part_transaction_path(path);
-        let cached_published = self
-            .io_get_object_open_path(volume, &txn_rel_path)
-            .ok()
-            .and_then(|txn_open| take_published_part_txn_receipt(&txn_open));
-        let (current_data_path, current_meta_path, transaction_path, skip_obsolete_read_on_commit, cached_old_flags) =
-            match cached_published {
-                Some(entry) => (
-                    entry.current_data_path,
-                    entry.current_meta_path,
-                    entry.transaction_path,
-                    !entry.had_old_meta,
-                    Some((entry.had_old_data, entry.had_old_meta)),
-                ),
-                None => {
-                    let (_, current_data_path) = self.io_get_bucket_and_object_path(volume, path)?;
-                    let current_meta_path = self.io_get_object_path(volume, &format!("{path}.meta"))?;
-                    let transaction_path = self.io_get_object_path(volume, &txn_rel_path)?;
-                    for candidate in [&current_data_path, &current_meta_path, &transaction_path] {
-                        check_path_length(candidate.to_string_lossy().as_ref())?;
-                    }
-                    (current_data_path, current_meta_path, transaction_path, false, None)
+        let txn_open_path = self.io_get_object_open_path(volume, &txn_rel_path).ok();
+        let cached_published = txn_open_path.as_deref().and_then(take_published_part_txn_receipt);
+        if cached_published.is_none()
+            && let Some(ref txn_open) = txn_open_path
+        {
+            let _ = take_prepared_part_txn_receipt(txn_open);
+        }
+        let (
+            current_data_path,
+            current_meta_path,
+            transaction_path,
+            skip_obsolete_read_on_commit,
+            cached_old_flags,
+            mut txn_dir_fd,
+            mut txn_parent_fd,
+        ) = match cached_published {
+            Some(entry) => (
+                entry.current_data_path,
+                entry.current_meta_path,
+                entry.transaction_path,
+                !entry.had_old_meta,
+                Some((entry.had_old_data, entry.had_old_meta)),
+                entry.txn_dir_fd,
+                entry.txn_parent_fd,
+            ),
+            None => {
+                let (_, current_data_path) = self.io_get_bucket_and_object_path(volume, path)?;
+                let current_meta_path = self.io_get_object_path(volume, &format!("{path}.meta"))?;
+                let transaction_path = self.io_get_object_path(volume, &txn_rel_path)?;
+                for candidate in [&current_data_path, &current_meta_path, &transaction_path] {
+                    check_path_length(candidate.to_string_lossy().as_ref())?;
                 }
-            };
+                (current_data_path, current_meta_path, transaction_path, false, None, None, None)
+            }
+        };
         let durability = effective_durability(volume);
         let sync = durability.syncs_commit_metadata();
         let part_number = current_data_path
@@ -14976,11 +15272,14 @@ impl DiskAPI for LocalDisk {
 
         os::record_native_publication(&current_data_path);
         let res = os::run_blocking_namespace_operation_with_sync(lease, sync, move || -> std::io::Result<Result<bool>> {
-            match std::fs::symlink_metadata(&transaction_path) {
-                Ok(metadata) if metadata.is_dir() => {}
-                Ok(_) => return Ok(Err(DiskError::FileCorrupt)),
-                Err(err) if err.kind() == ErrorKind::NotFound => return Ok(Ok(false)),
-                Err(err) => return Ok(Err(to_file_error(err).into())),
+            match check_part_txn_dir_in_parent_fd(
+                txn_parent_fd.as_deref(),
+                transaction_path.file_name().and_then(|n| n.to_str()),
+                &transaction_path,
+            )? {
+                Ok(true) => {}
+                Ok(false) => return Ok(Ok(false)),
+                Err(err) => return Ok(Err(err)),
             }
 
             if let Some(admission) = admission
@@ -14993,7 +15292,7 @@ impl DiskAPI for LocalDisk {
                     return Ok(Err(to_file_error(err).into()));
                 }
                 if durability.syncs_commit_metadata()
-                    && let Err(err) = os::fsync_dir_std(&transaction_path)
+                    && let Err(err) = os::fsync_directory_handle_std(&transaction_path, txn_dir_fd.as_deref())
                 {
                     return Ok(Err(to_file_error(err).into()));
                 }
@@ -15015,7 +15314,7 @@ impl DiskAPI for LocalDisk {
                 }
                 if durability.syncs_commit_metadata()
                     && let Some(parent) = current_data_path.parent()
-                    && let Err(err) = os::fsync_dir_std(parent)
+                    && let Err(err) = os::fsync_directory_handle_std(parent, txn_parent_fd.as_deref())
                 {
                     return Ok(Err(to_file_error(err).into()));
                 }
@@ -15049,18 +15348,25 @@ impl DiskAPI for LocalDisk {
                     != Some(&integrity)
             {
                 let obsolete_path = parent.join(integrity.file_name());
-                match std::fs::remove_file(&obsolete_path) {
+                match unlink_in_dir_fd(txn_parent_fd.as_deref(), &integrity.file_name(), &obsolete_path) {
                     Ok(()) => {}
                     Err(error) if error.kind() == ErrorKind::NotFound => {}
                     Err(error) => return Ok(Err(to_file_error(error).into())),
                 }
             }
-            let cleanup_path = parent.join(format!(".part-txn-settled-{}", Uuid::new_v4()));
-            if let Err(err) = std::fs::rename(&transaction_path, &cleanup_path) {
+            let cleanup_name = format!(".part-txn-settled-{}", Uuid::new_v4());
+            let cleanup_path = parent.join(&cleanup_name);
+            if let Err(err) = rename_in_same_dir_fd(
+                txn_parent_fd.as_deref(),
+                transaction_path.file_name().and_then(|n| n.to_str()),
+                &transaction_path,
+                Some(&cleanup_name),
+                &cleanup_path,
+            ) {
                 return Ok(Err(to_file_error(err).into()));
             }
             if durability.syncs_commit_metadata()
-                && let Err(err) = os::fsync_dir_std(parent)
+                && let Err(err) = os::fsync_directory_handle_std(parent, txn_parent_fd.as_deref())
             {
                 return Ok(Err(to_file_error(err).into()));
             }
@@ -15076,15 +15382,22 @@ impl DiskAPI for LocalDisk {
                     PART_TRANSACTION_OLD_META_ABSENT
                 };
                 for name in [PART_TRANSACTION_NEW_META, old_data_name, old_meta_name] {
-                    let _ = std::fs::remove_file(cleanup_path.join(name));
+                    let _ = unlink_in_dir_fd(txn_dir_fd.as_deref(), name, &cleanup_path.join(name));
                 }
-                if std::fs::remove_dir(&cleanup_path).is_err()
+                drop(txn_dir_fd.take());
+                let rmdir_res = rmdir_in_dir_fd(txn_parent_fd.as_deref(), &cleanup_name, &cleanup_path);
+                drop(txn_parent_fd.take());
+                if rmdir_res.is_err()
                     && let Err(err) = remove_dir_all_if_exists(&cleanup_path)
                 {
                     return Ok(Err(to_file_error(err).into()));
                 }
-            } else if let Err(err) = remove_dir_all_if_exists(&cleanup_path) {
-                return Ok(Err(to_file_error(err).into()));
+            } else {
+                drop(txn_dir_fd.take());
+                drop(txn_parent_fd.take());
+                if let Err(err) = remove_dir_all_if_exists(&cleanup_path) {
+                    return Ok(Err(to_file_error(err).into()));
+                }
             }
             Ok(Ok(true))
         })
@@ -15156,6 +15469,8 @@ impl DiskAPI for LocalDisk {
             dst_meta_path,
             transaction_path,
             preverified_meta_and_old_flags,
+            txn_dir_fd,
+            txn_parent_fd,
         ) = match cached_prepared {
             Some(entry) => (
                 entry.src_volume_dir,
@@ -15165,6 +15480,8 @@ impl DiskAPI for LocalDisk {
                 Some(entry.dst_meta_path),
                 Some(entry.transaction_path),
                 Some((entry.had_old_data, entry.had_old_meta)),
+                entry.txn_dir_fd,
+                entry.txn_parent_fd,
             ),
             None => {
                 let (src_volume_dir, src_file_path) = self.io_get_bucket_and_object_path(src_volume, src_path)?;
@@ -15199,6 +15516,8 @@ impl DiskAPI for LocalDisk {
                     dst_meta_path,
                     transaction_path,
                     None,
+                    None,
+                    None,
                 )
             }
         };
@@ -15214,8 +15533,10 @@ impl DiskAPI for LocalDisk {
         let meta_for_legacy = meta.clone();
         let lease = os::acquire_namespace_mutation_lease_with_owner(&dst_file_path, source_owner.clone()).await;
 
-        let published_transaction_meta =
-            os::run_blocking_namespace_operation_with_sync(lease, sync, move || -> std::io::Result<bool> {
+        let published_transaction_meta = os::run_blocking_namespace_operation_with_sync(
+            lease,
+            sync,
+            move || -> std::io::Result<(bool, Option<(PathBuf, PublishedPartTxnReceipt)>)> {
                 if let Some(check) = &admission {
                     check.validate().map_err(to_file_error)?;
                 }
@@ -15262,8 +15583,12 @@ impl DiskAPI for LocalDisk {
                     let publish_meta_path = transaction_path.join(PART_TRANSACTION_PUBLISH_META);
                     if !cfg!(test)
                         && preverified_meta_and_old_flags.is_some()
-                        && publish_meta_path.is_file()
-                        && std::fs::metadata(&transaction_meta_path).is_ok_and(|m| m.is_file() && m.len() == meta.len() as u64)
+                        && part_txn_files_ready(
+                            txn_dir_fd.as_deref(),
+                            &publish_meta_path,
+                            &transaction_meta_path,
+                            meta.len() as u64,
+                        )
                     {
                         Some(publish_meta_path)
                     } else {
@@ -15366,29 +15691,39 @@ impl DiskAPI for LocalDisk {
                             }
                         }
                     }
-                    if let (Some((had_old_data, had_old_meta)), Some(txn_path), Some(meta_path)) =
-                        (preverified_meta_and_old_flags, transaction_path.as_ref(), dst_meta_path.as_path().into())
+                    let published_entry = if let (Some((had_old_data, had_old_meta)), Some(txn_path), Some(meta_path)) =
+                        (preverified_meta_and_old_flags, transaction_path, dst_meta_path.as_path().into())
                     {
-                        put_published_part_txn_receipt(
-                            txn_path,
+                        Some((
+                            txn_path.clone(),
                             PublishedPartTxnReceipt {
                                 current_data_path: dst_file_path,
                                 current_meta_path: meta_path.to_path_buf(),
-                                transaction_path: txn_path.clone(),
+                                transaction_path: txn_path,
                                 had_old_data,
                                 had_old_meta,
+                                txn_dir_fd,
+                                txn_parent_fd,
                             },
-                        );
-                    }
-                    Ok(true)
+                        ))
+                    } else {
+                        None
+                    };
+                    Ok((true, published_entry))
                 } else {
-                    Ok(false)
+                    Ok((false, None))
                 }
-            })
-            .await
-            .map_err(DiskError::from);
+            },
+        )
+        .await
+        .map_err(DiskError::from);
         let published_transaction_meta = match published_transaction_meta {
-            Ok(published) => published,
+            Ok((published, maybe_entry)) => {
+                if let Some((txn_path, entry)) = maybe_entry {
+                    put_published_part_txn_receipt(&txn_path, entry);
+                }
+                published
+            }
             Err(err) => {
                 if let Some(ref source) = admission_source {
                     multipart_source_state::invalidate_active_multipart_source_head(source);
@@ -38604,4 +38939,74 @@ mod test {
         assert_eq!(&on_disk[..expected_hash.as_ref().len()], expected_hash.as_ref());
         assert_eq!(&on_disk[expected_hash.as_ref().len()..], payload.as_ref());
     }
+
+    #[tokio::test]
+    async fn part_txn_files_ready_and_receipt_cache_eviction_on_close_and_drop() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let endpoint = Endpoint::try_from(dir.path().to_str().expect("utf8 path")).expect("endpoint");
+        let disk = LocalDisk::new(&endpoint, false).await.expect("local disk");
+        ensure_test_volume(&disk, "tmp").await;
+        ensure_test_volume(&disk, "bucket").await;
+
+        let payload = Bytes::from_static(b"part payload");
+        let meta = Bytes::from(
+            ObjectPartInfo {
+                number: 1,
+                size: payload.len(),
+                ..Default::default()
+            }
+            .marshal_msg()
+            .expect("part metadata"),
+        );
+        disk.write_all("tmp", "upload/part.1", payload.clone())
+            .await
+            .expect("stage part");
+        disk.prepare_part_transaction("tmp", "upload/part.1", "bucket", "object/part.1", meta.clone())
+            .await
+            .expect("prepare part transaction");
+
+        let txn_rel = crate::disk::part_transaction_path("object/part.1");
+        let txn_open = disk
+            .io_get_object_open_path("bucket", &txn_rel)
+            .expect("transaction open path");
+        let publish_meta = txn_open.join(PART_TRANSACTION_PUBLISH_META);
+        let new_meta = txn_open.join(PART_TRANSACTION_NEW_META);
+        let entry_fd = PREPARED_PART_TXN_RECEIPTS
+            .with_get(&txn_open, |entry| entry.txn_dir_fd.clone())
+            .expect("prepared receipt should be cached");
+        assert!(part_txn_files_ready(entry_fd.as_deref(), &publish_meta, &new_meta, meta.len() as u64,));
+        assert!(!part_txn_files_ready(
+            entry_fd.as_deref(),
+            &publish_meta,
+            &new_meta,
+            (meta.len() + 1) as u64,
+        ));
+        drop(entry_fd);
+
+        disk.close().await.expect("close disk");
+        assert!(
+            take_prepared_part_txn_receipt(&txn_open).is_none(),
+            "LocalDisk::close must purge cached part transaction receipts"
+        );
+
+        disk.write_all("tmp", "upload2/part.2", payload)
+            .await
+            .expect("stage second part");
+        disk.prepare_part_transaction("tmp", "upload2/part.2", "bucket", "object/part.2", meta)
+            .await
+            .expect("prepare second part transaction");
+        let txn2_rel = crate::disk::part_transaction_path("object/part.2");
+        let txn2_open = disk
+            .io_get_object_open_path("bucket", &txn2_rel)
+            .expect("second transaction open path");
+        assert!(
+            PREPARED_PART_TXN_RECEIPTS.with_get(&txn2_open, |_| ()).is_some(),
+            "second prepared receipt should be cached before drop"
+        );
+        drop(disk);
+        assert!(
+            take_prepared_part_txn_receipt(&txn2_open).is_none(),
+            "Drop for LocalDisk must purge cached part transaction receipts"
+        );
+    }
 }

crates/ecstore/src/disk/local/delete_recovery.rs

@@ -358,14 +358,26 @@ pub(in crate::disk) fn generation(mut canonical: FileInfo) -> Result<crate::disk
             _ => {}
         }
     }
+    struct Sha256Writer(Sha256);
+    impl std::io::Write for Sha256Writer {
+        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+            self.0.update(buf);
+            Ok(buf.len())
+        }
+        fn flush(&mut self) -> std::io::Result<()> {
+            Ok(())
+        }
+    }
     let mut json_value = serde_json::to_value(&canonical)?;
     sort_json_in_place(&mut json_value);
+    let mut hasher = Sha256Writer(Sha256::new());
+    serde_json::to_writer(&mut hasher, &json_value)?;
     Ok(crate::disk::DeleteRecoveryVersionIdentity {
         version_id: canonical.version_id,
         mod_time: canonical.mod_time,
         data_dir: canonical.data_dir,
         deleted: canonical.deleted,
-        metadata_digest: digest(&serde_json::to_vec(&json_value)?),
+        metadata_digest: hasher.0.finalize().into(),
     })
 }
 
@@ -1174,6 +1186,7 @@ struct PreparedDeleteReceiptEntry {
     sole_transaction: bool,
     had_old_data_dir: bool,
     rollback_dir_fd: Option<Arc<std::fs::File>>,
+    file_path_dir: Option<Arc<std::fs::File>>,
 }
 
 #[derive(Clone)]
@@ -1185,6 +1198,189 @@ struct PublishedDeleteReceiptEntry {
     journal_raw: Vec<u8>,
     had_old_data_dir: bool,
     rollback_dir_fd: Option<Arc<std::fs::File>>,
+    file_path_dir: Option<Arc<std::fs::File>>,
+}
+
+#[cfg(unix)]
+fn read_child_regular_bytes(
+    parent_fd: Option<&std::fs::File>,
+    child_name: &str,
+    fallback_path: &Path,
+) -> std::io::Result<Vec<u8>> {
+    if let Some(dir) = parent_fd {
+        use rustix::fs::{Mode, OFlags, openat};
+        let fd = openat(dir, child_name, OFlags::RDONLY | OFlags::CLOEXEC, Mode::empty()).map_err(std::io::Error::from)?;
+        let mut file = std::fs::File::from(fd);
+        let cap = file
+            .metadata()
+            .ok()
+            .and_then(|m| usize::try_from(m.len()).ok())
+            .unwrap_or(0)
+            .min(64 * 1024);
+        let mut buf = Vec::with_capacity(cap);
+        std::io::Read::read_to_end(&mut file, &mut buf)?;
+        return Ok(buf);
+    }
+    std::fs::read(fallback_path)
+}
+
+#[cfg(not(unix))]
+fn read_child_regular_bytes(
+    _parent_fd: Option<&std::fs::File>,
+    _child_name: &str,
+    fallback_path: &Path,
+) -> std::io::Result<Vec<u8>> {
+    std::fs::read(fallback_path)
+}
+
+#[cfg(unix)]
+fn create_and_write_child_file(
+    parent_fd: Option<&std::fs::File>,
+    child_name: &str,
+    fallback_path: &Path,
+    bytes: &[u8],
+) -> std::io::Result<std::fs::File> {
+    let mut file = if let Some(dir) = parent_fd {
+        use rustix::fs::{Mode, OFlags, openat};
+        let fd = openat(
+            dir,
+            child_name,
+            OFlags::WRONLY | OFlags::CREATE | OFlags::TRUNC | OFlags::CLOEXEC,
+            Mode::from_raw_mode(0o666),
+        )
+        .map_err(std::io::Error::from)?;
+        std::fs::File::from(fd)
+    } else {
+        std::fs::File::create(fallback_path)?
+    };
+    std::io::Write::write_all(&mut file, bytes)?;
+    Ok(file)
+}
+
+#[cfg(not(unix))]
+fn create_and_write_child_file(
+    _parent_fd: Option<&std::fs::File>,
+    _child_name: &str,
+    fallback_path: &Path,
+    bytes: &[u8],
+) -> std::io::Result<std::fs::File> {
+    let mut file = std::fs::File::create(fallback_path)?;
+    std::io::Write::write_all(&mut file, bytes)?;
+    Ok(file)
+}
+
+#[cfg(unix)]
+fn link_child_files(
+    src_dir_fd: Option<&std::fs::File>,
+    src_name: &str,
+    src_fallback: &Path,
+    dst_dir_fd: Option<&std::fs::File>,
+    dst_name: &str,
+    dst_fallback: &Path,
+) -> std::io::Result<()> {
+    if let (Some(s_fd), Some(d_fd)) = (src_dir_fd, dst_dir_fd) {
+        return rustix::fs::linkat(s_fd, src_name, d_fd, dst_name, rustix::fs::AtFlags::empty()).map_err(std::io::Error::from);
+    }
+    std::fs::hard_link(src_fallback, dst_fallback)
+}
+
+#[cfg(not(unix))]
+fn link_child_files(
+    _src_dir_fd: Option<&std::fs::File>,
+    _src_name: &str,
+    src_fallback: &Path,
+    _dst_dir_fd: Option<&std::fs::File>,
+    _dst_name: &str,
+    dst_fallback: &Path,
+) -> std::io::Result<()> {
+    std::fs::hard_link(src_fallback, dst_fallback)
+}
+
+#[cfg(unix)]
+fn link_from_dir_fd(
+    src_dir_fd: Option<&std::fs::File>,
+    src_name: &str,
+    src_fallback: &Path,
+    dst_path: &Path,
+) -> std::io::Result<()> {
+    if let Some(s_fd) = src_dir_fd {
+        return rustix::fs::linkat(s_fd, src_name, rustix::fs::CWD, dst_path, rustix::fs::AtFlags::empty())
+            .map_err(std::io::Error::from);
+    }
+    std::fs::hard_link(src_fallback, dst_path)
+}
+
+#[cfg(not(unix))]
+fn link_from_dir_fd(
+    _src_dir_fd: Option<&std::fs::File>,
+    _src_name: &str,
+    src_fallback: &Path,
+    dst_path: &Path,
+) -> std::io::Result<()> {
+    std::fs::hard_link(src_fallback, dst_path)
+}
+
+#[cfg(unix)]
+fn rename_into_dir_fd(
+    src_path: &Path,
+    dst_dir_fd: Option<&std::fs::File>,
+    dst_name: &str,
+    dst_fallback: &Path,
+) -> std::io::Result<()> {
+    if let Some(d_fd) = dst_dir_fd {
+        return rustix::fs::renameat(rustix::fs::CWD, src_path, d_fd, dst_name).map_err(std::io::Error::from);
+    }
+    std::fs::rename(src_path, dst_fallback)
+}
+
+#[cfg(not(unix))]
+fn rename_into_dir_fd(
+    src_path: &Path,
+    _dst_dir_fd: Option<&std::fs::File>,
+    _dst_name: &str,
+    dst_fallback: &Path,
+) -> std::io::Result<()> {
+    std::fs::rename(src_path, dst_fallback)
+}
+
+#[cfg(unix)]
+fn child_regular_file_exists(dir_fd: Option<&std::fs::File>, child_name: &str, fallback_path: &Path) -> bool {
+    if let Some(d_fd) = dir_fd {
+        return rustix::fs::statat(d_fd, child_name, rustix::fs::AtFlags::SYMLINK_NOFOLLOW)
+            .is_ok_and(|st| (st.st_mode & libc::S_IFMT) == libc::S_IFREG);
+    }
+    fallback_path.is_file()
+}
+
+#[cfg(not(unix))]
+fn child_regular_file_exists(_dir_fd: Option<&std::fs::File>, _child_name: &str, fallback_path: &Path) -> bool {
+    fallback_path.is_file()
+}
+
+#[cfg(unix)]
+fn unlink_child_file(dir_fd: Option<&std::fs::File>, child_name: &str, fallback_path: &Path) -> std::io::Result<()> {
+    if let Some(d_fd) = dir_fd {
+        return rustix::fs::unlinkat(d_fd, child_name, rustix::fs::AtFlags::empty()).map_err(std::io::Error::from);
+    }
+    std::fs::remove_file(fallback_path)
+}
+
+#[cfg(not(unix))]
+fn unlink_child_file(_dir_fd: Option<&std::fs::File>, _child_name: &str, fallback_path: &Path) -> std::io::Result<()> {
+    std::fs::remove_file(fallback_path)
+}
+
+#[cfg(unix)]
+fn rmdir_child_dir(dir_fd: Option<&std::fs::File>, child_name: &str, fallback_path: &Path) -> std::io::Result<()> {
+    if let Some(d_fd) = dir_fd {
+        return rustix::fs::unlinkat(d_fd, child_name, rustix::fs::AtFlags::REMOVEDIR).map_err(std::io::Error::from);
+    }
+    std::fs::remove_dir(fallback_path)
+}
+
+#[cfg(not(unix))]
+fn rmdir_child_dir(_dir_fd: Option<&std::fs::File>, _child_name: &str, fallback_path: &Path) -> std::io::Result<()> {
+    std::fs::remove_dir(fallback_path)
 }
 
 const PREPARED_DELETE_RECEIPTS_PER_SHARD: usize = 64;
@@ -1211,6 +1407,21 @@ fn take_published_delete_receipt(rollback: &Path) -> Option<PublishedDeleteRecei
     PUBLISHED_DELETE_RECEIPTS.remove(rollback)
 }
 
+pub(super) fn clear_delete_receipts_under(io_root: &Path, root: &Path) {
+    for prefix in [io_root, root] {
+        PREPARED_DELETE_RECEIPTS.retain_outside_prefix(prefix);
+        PUBLISHED_DELETE_RECEIPTS.retain_outside_prefix(prefix);
+    }
+}
+
+#[cfg(unix)]
+fn open_dir_fd(path: &Path) -> std::io::Result<std::fs::File> {
+    use rustix::fs::{Mode, OFlags, open};
+    open(path, OFlags::RDONLY | OFlags::DIRECTORY | OFlags::CLOEXEC, Mode::empty())
+        .map(std::fs::File::from)
+        .map_err(std::io::Error::from)
+}
+
 fn format_commit_receipt_bytes(journal_raw: &[u8], finished: bool) -> Vec<u8> {
     let suffix: &[u8] = if finished {
         b",\"finished\":true}"
@@ -2542,11 +2753,22 @@ impl LocalDisk {
         let object_str = object.to_owned();
 
         let recovery_scope = recovery_operation_scope();
+        type FusedStage1Outcome = Option<(Option<Uuid>, PathBuf, PreparedDeleteReceiptEntry)>;
         let res = os::run_blocking_namespace_operation_with_sync(
             lease,
             sync,
-            move || -> std::io::Result<Result<Option<Option<Uuid>>>> {
-                let buf = match std::fs::read(&xl_path) {
+            move || -> std::io::Result<Result<FusedStage1Outcome>> {
+                #[cfg(unix)]
+                let file_path_dir = match open_dir_fd(&file_path) {
+                    Ok(dir) => Some(Arc::new(dir)),
+                    Err(err) if matches!(err.kind(), ErrorKind::NotFound | ErrorKind::NotADirectory) => {
+                        return Ok(Ok(None));
+                    }
+                    Err(_) => None,
+                };
+                #[cfg(not(unix))]
+                let file_path_dir: Option<Arc<std::fs::File>> = None;
+                let buf = match read_child_regular_bytes(file_path_dir.as_deref(), STORAGE_FORMAT_FILE, &xl_path) {
                     Ok(buf) => buf,
                     Err(err) if err.kind() == ErrorKind::NotFound => return Ok(Ok(None)),
                     Err(err) => return Err(err),
@@ -2704,10 +2926,23 @@ impl LocalDisk {
                     Err(err) => return Err(err),
                 };
                 let staged_dir = staged.path().to_path_buf();
+                #[cfg(unix)]
+                let rollback_dir_fd = open_dir_fd(&staged_dir).ok().map(Arc::new);
+                #[cfg(not(unix))]
+                let rollback_dir_fd: Option<Arc<std::fs::File>> = None;
                 let backup_path = staged_dir.join(STORAGE_FORMAT_FILE_BACKUP);
-                if std::fs::hard_link(&xl_path, &backup_path).is_err() {
-                    let mut file = std::fs::File::create(&backup_path)?;
-                    std::io::Write::write_all(&mut file, &buf)?;
+                if link_child_files(
+                    file_path_dir.as_deref(),
+                    STORAGE_FORMAT_FILE,
+                    &xl_path,
+                    rollback_dir_fd.as_deref(),
+                    STORAGE_FORMAT_FILE_BACKUP,
+                    &backup_path,
+                )
+                .is_err()
+                {
+                    let file =
+                        create_and_write_child_file(rollback_dir_fd.as_deref(), STORAGE_FORMAT_FILE_BACKUP, &backup_path, &buf)?;
                     if sync && !buf.is_empty() {
                         file.sync_data()?;
                     }
@@ -2717,8 +2952,7 @@ impl LocalDisk {
                     let Some(bytes) = bytes else {
                         continue;
                     };
-                    let mut file = std::fs::File::create(staged_dir.join(name))?;
-                    std::io::Write::write_all(&mut file, bytes)?;
+                    let file = create_and_write_child_file(rollback_dir_fd.as_deref(), name, &staged_dir.join(name), bytes)?;
                     if sync && !bytes.is_empty() {
                         written_files.push(file);
                     }
@@ -2726,21 +2960,20 @@ impl LocalDisk {
                 for file in written_files {
                     file.sync_data()?;
                 }
-                #[cfg(unix)]
-                let rollback_dir_fd = sync.then(|| std::fs::File::open(&staged_dir).ok().map(Arc::new)).flatten();
-                #[cfg(not(unix))]
-                let rollback_dir_fd: Option<Arc<std::fs::File>> = None;
                 if sync {
                     os::fsync_directory_handle_std(&staged_dir, rollback_dir_fd.as_deref())?;
                 }
+                let tx_name = transaction.to_string();
                 if !destination_missing {
                     match std::fs::symlink_metadata(&destination) {
-                        Ok(metadata) if metadata.is_dir() => match std::fs::remove_dir(&destination) {
-                            Ok(()) => {}
-                            Err(err) if err.kind() == ErrorKind::NotFound => {}
-                            Err(err) if is_dir_not_empty_error(&err) => return Ok(Err(DiskError::OutdatedXLMeta)),
-                            Err(err) => return Err(err),
-                        },
+                        Ok(metadata) if metadata.is_dir() => {
+                            match rmdir_child_dir(file_path_dir.as_deref(), &tx_name, &destination) {
+                                Ok(()) => {}
+                                Err(err) if err.kind() == ErrorKind::NotFound => {}
+                                Err(err) if is_dir_not_empty_error(&err) => return Ok(Err(DiskError::OutdatedXLMeta)),
+                                Err(err) => return Err(err),
+                            }
+                        }
                         Ok(_) => return Ok(Err(DiskError::OutdatedXLMeta)),
                         Err(err) if err.kind() == ErrorKind::NotFound => {}
                         Err(err) => return Err(err),
@@ -2748,18 +2981,14 @@ impl LocalDisk {
                 }
                 let staged_dir = staged.keep();
                 os::record_native_publication(&destination);
-                if let Err(err) = std::fs::rename(&staged_dir, &destination) {
+                if let Err(err) = rename_into_dir_fd(&staged_dir, file_path_dir.as_deref(), &tx_name, &destination) {
                     let _ = std::fs::remove_dir_all(&staged_dir);
                     return Err(err);
                 }
                 let old_data_path = old_dir
                     .map(|uuid| file_path.join(uuid.to_string()))
                     .filter(|p| old_data_dir_state.unwrap_or_else(|| p.is_dir()));
-                #[cfg(unix)]
-                let file_path_dir = sync.then(|| std::fs::File::open(&file_path).ok()).flatten();
-                #[cfg(not(unix))]
-                let file_path_dir: Option<std::fs::File> = None;
-                if sync && let Err(err) = os::fsync_directory_handle_std(&file_path, file_path_dir.as_ref()) {
+                if sync && let Err(err) = os::fsync_directory_handle_std(&file_path, file_path_dir.as_deref()) {
                     let _ = std::fs::remove_dir_all(&destination);
                     return Err(err);
                 }
@@ -2769,7 +2998,7 @@ impl LocalDisk {
                 if let Some(old_path) = old_data_path {
                     let reserved_marker = old_path.join(format!("{RESERVED_DELETE_DATA_DIR_MARKER_PREFIX}{transaction}"));
                     #[cfg(unix)]
-                    let old_dir_file = sync.then(|| std::fs::File::open(&old_path).ok()).flatten();
+                    let old_dir_file = sync.then(|| open_dir_fd(&old_path).ok()).flatten();
                     #[cfg(not(unix))]
                     let old_dir_file: Option<std::fs::File> = None;
                     let reserve_res = (|| -> std::io::Result<()> {
@@ -2792,16 +3021,16 @@ impl LocalDisk {
                     let target_in_rollback = destination.join(TARGET);
                     let linked_xl = staging_root.join(format!("delete-xl-{transaction}"));
                     os::record_native_publication(&xl_path);
-                    let linked = match std::fs::hard_link(&target_in_rollback, &linked_xl) {
+                    let linked = match link_from_dir_fd(rollback_dir_fd.as_deref(), TARGET, &target_in_rollback, &linked_xl) {
                         Ok(()) => true,
                         Err(err) if err.kind() == ErrorKind::AlreadyExists => {
                             let _ = std::fs::remove_file(&linked_xl);
-                            std::fs::hard_link(&target_in_rollback, &linked_xl).is_ok()
+                            link_from_dir_fd(rollback_dir_fd.as_deref(), TARGET, &target_in_rollback, &linked_xl).is_ok()
                         }
                         Err(_) => false,
                     };
                     let write_res = if linked {
-                        let rename_res = std::fs::rename(&linked_xl, &xl_path);
+                        let rename_res = rename_into_dir_fd(&linked_xl, file_path_dir.as_deref(), STORAGE_FORMAT_FILE, &xl_path);
                         if rename_res.is_err() {
                             let _ = std::fs::remove_file(&linked_xl);
                         }
@@ -2813,7 +3042,7 @@ impl LocalDisk {
                             if sync && !target_bytes.is_empty() {
                                 staged_xl.as_file().sync_data()?;
                             }
-                            std::fs::rename(staged_xl.path(), &xl_path)
+                            rename_into_dir_fd(staged_xl.path(), file_path_dir.as_deref(), STORAGE_FORMAT_FILE, &xl_path)
                         })()
                     };
                     if let Err(err) = write_res {
@@ -2823,7 +3052,7 @@ impl LocalDisk {
                         let _ = std::fs::remove_dir_all(&destination);
                         return Err(err);
                     }
-                } else if let Err(err) = std::fs::remove_file(&xl_path)
+                } else if let Err(err) = unlink_child_file(file_path_dir.as_deref(), STORAGE_FORMAT_FILE, &xl_path)
                     && err.kind() != ErrorKind::NotFound
                 {
                     if let Some((_, reserved_marker, _)) = &reserved_data_path {
@@ -2833,7 +3062,7 @@ impl LocalDisk {
                     return Err(err);
                 }
                 if sync {
-                    os::fsync_directory_handle_std(&file_path, file_path_dir.as_ref())?;
+                    os::fsync_directory_handle_std(&file_path, file_path_dir.as_deref())?;
                 }
                 let had_old_data_dir = reserved_data_path.is_some();
                 if let Some((old_path, reserved_marker, old_dir_file)) = reserved_data_path {
@@ -2845,7 +3074,8 @@ impl LocalDisk {
                         Err(err) => return Err(err),
                     }
                 }
-                put_prepared_delete_receipt(
+                Ok(Ok(Some((
+                    old_dir.filter(|_| had_old_data_dir),
                     destination,
                     PreparedDeleteReceiptEntry {
                         object_path,
@@ -2857,17 +3087,18 @@ impl LocalDisk {
                         sole_transaction,
                         had_old_data_dir,
                         rollback_dir_fd,
+                        file_path_dir,
                     },
-                );
-                Ok(Ok(Some(old_dir.filter(|_| had_old_data_dir))))
+                ))))
             },
         )
         .await
         .map_err(to_file_error)??;
 
-        let Some(old_dir) = res else {
+        let Some((old_dir, destination, receipt_entry)) = res else {
             return Ok(None);
         };
+        put_prepared_delete_receipt(destination, receipt_entry);
         if let Some(uuid) = old_dir {
             self.io_backend
                 .invalidate_cached_fds_under(volume, &format!("{object}/{uuid}"));
@@ -2918,6 +3149,7 @@ impl LocalDisk {
                     entry.sole_transaction,
                     entry.had_old_data_dir,
                     entry.rollback_dir_fd,
+                    entry.file_path_dir,
                 )),
             ),
             None => {
@@ -2945,12 +3177,14 @@ impl LocalDisk {
         let volume_str = volume.to_owned();
         let object_str = object.to_owned();
 
-        os::run_blocking_namespace_operation_with_sync(
+        type PublishDeleteFastReturn = (DeleteRecoveryJournal, bool, bool, Option<(PathBuf, PublishedDeleteReceiptEntry)>);
+        let res = os::run_blocking_namespace_operation_with_sync(
             lease,
             sync,
-            move || -> std::io::Result<Result<Option<(DeleteRecoveryJournal, bool, bool)>>> {
-                if let Some((journal_raw, journal, sole_transaction, had_old_data_dir, rollback_dir_fd)) = cached_receipt
-                    && rollback.join(PREPARED).is_file()
+            move || -> std::io::Result<Result<Option<PublishDeleteFastReturn>>> {
+                if let Some((journal_raw, journal, sole_transaction, had_old_data_dir, rollback_dir_fd, file_path_dir)) =
+                    cached_receipt
+                    && child_regular_file_exists(rollback_dir_fd.as_deref(), PREPARED, &rollback.join(PREPARED))
                 {
                     let tmp_receipt = staging_root.join(format!("receipt-{transaction}.tmp"));
                     let receipt_bytes = format_commit_receipt_bytes(&journal_raw, false);
@@ -2961,7 +3195,12 @@ impl LocalDisk {
                             file.sync_data()?;
                         }
                         os::record_native_publication(&receipt_path);
-                        std::fs::rename(&tmp_receipt, &receipt_path)?;
+                        rename_into_dir_fd(
+                            &tmp_receipt,
+                            rollback_dir_fd.as_deref(),
+                            DELETE_QUORUM_COMMITTED_FILE,
+                            &receipt_path,
+                        )?;
                         if sync {
                             os::fsync_directory_handle_std(&rollback, rollback_dir_fd.as_deref())?;
                         }
@@ -2971,8 +3210,8 @@ impl LocalDisk {
                         let _ = std::fs::remove_file(&tmp_receipt);
                         return Err(err);
                     }
-                    if let Some(bucket_path) = bucket_path {
-                        put_published_delete_receipt(
+                    let published_entry = bucket_path.map(|bucket_path| {
+                        (
                             rollback,
                             PublishedDeleteReceiptEntry {
                                 object_path,
@@ -2982,10 +3221,11 @@ impl LocalDisk {
                                 journal_raw,
                                 had_old_data_dir,
                                 rollback_dir_fd,
+                                file_path_dir,
                             },
-                        );
-                    }
-                    return Ok(Ok(Some((journal, sole_transaction, true))));
+                        )
+                    });
+                    return Ok(Ok(Some((journal, sole_transaction, true, published_entry))));
                 }
                 let prepared = match regular_bytes(&rollback.join(PREPARED)) {
                     Ok(Some(bytes)) => bytes,
@@ -3064,11 +3304,19 @@ impl LocalDisk {
                 if sync {
                     os::fsync_dir_std(&rollback)?;
                 }
-                Ok(Ok(Some((journal, sole_transaction, target_satisfied))))
+                Ok(Ok(Some((journal, sole_transaction, target_satisfied, None))))
             },
         )
         .await
-        .map_err(to_file_error)?
+        .map_err(to_file_error)?;
+
+        let Some((journal, sole_transaction, target_satisfied, published_entry)) = res? else {
+            return Ok(None);
+        };
+        if let Some((rollback, entry)) = published_entry {
+            put_published_delete_receipt(rollback, entry);
+        }
+        Ok(Some((journal, sole_transaction, target_satisfied)))
     }
 
     #[cfg(test)]
@@ -3118,34 +3366,49 @@ impl LocalDisk {
             None
         };
         let cached_published = rollback_open_path.as_deref().and_then(take_published_delete_receipt);
-        let (object_path, bucket_path, staging_root, root, cached_journal_raw, cached_had_old_data_dir, mut rollback_dir_fd) =
-            match cached_published {
-                Some(entry) => (
-                    entry.object_path,
-                    entry.bucket_path,
-                    entry.staging_root,
-                    entry.file_path,
-                    Some(entry.journal_raw),
-                    Some(entry.had_old_data_dir),
-                    entry.rollback_dir_fd,
-                ),
-                None => {
-                    self.reject_multipart_delete_recovery_mutation(volume, object)?;
-                    if !safe_path(object, false) {
-                        return Err(DiskError::InvalidPath);
-                    }
-                    let obj_path = self.get_object_path(volume, object)?;
-                    (
-                        obj_path,
-                        self.io_get_bucket_path(volume)?,
-                        self.io_get_bucket_path(RUSTFS_META_TMP_BUCKET)?,
-                        self.io_get_object_path(volume, object)?,
-                        None,
-                        None,
-                        None,
-                    )
+        if cached_published.is_none()
+            && let Some(ref tx_open) = rollback_open_path
+        {
+            let _ = take_prepared_delete_receipt(tx_open);
+        }
+        let (
+            object_path,
+            bucket_path,
+            staging_root,
+            root,
+            cached_journal_raw,
+            cached_had_old_data_dir,
+            mut rollback_dir_fd,
+            mut file_path_dir,
+        ) = match cached_published {
+            Some(entry) => (
+                entry.object_path,
+                entry.bucket_path,
+                entry.staging_root,
+                entry.file_path,
+                Some(entry.journal_raw),
+                Some(entry.had_old_data_dir),
+                entry.rollback_dir_fd,
+                entry.file_path_dir,
+            ),
+            None => {
+                self.reject_multipart_delete_recovery_mutation(volume, object)?;
+                if !safe_path(object, false) {
+                    return Err(DiskError::InvalidPath);
                 }
-            };
+                let obj_path = self.get_object_path(volume, object)?;
+                (
+                    obj_path,
+                    self.io_get_bucket_path(volume)?,
+                    self.io_get_bucket_path(RUSTFS_META_TMP_BUCKET)?,
+                    self.io_get_object_path(volume, object)?,
+                    None,
+                    None,
+                    None,
+                    None,
+                )
+            }
+        };
         let metadata_owner = os::acquire_metadata_mutation_lease(&object_path, namespace_owner).await;
         let volume_guard = os::disk_volume_mutation_lock(&self.root, volume).read_owned().await;
         let owner: Option<Arc<dyn Send + Sync>> = Some(Arc::new((metadata_owner, volume_guard)));
@@ -3272,7 +3535,7 @@ impl LocalDisk {
                 }
             }
             if had_physical_data_dirs && sync && root.is_dir() {
-                os::fsync_dir_std(&root)?;
+                os::fsync_directory_handle_std(&root, file_path_dir.as_deref())?;
             }
             if fast_cleanup {
                 let backup_or_marker = if has_original {
@@ -3283,8 +3546,10 @@ impl LocalDisk {
                     None
                 };
                 if let Some(name) = backup_or_marker
-                    && let Err(err) = ignore_not_found(std::fs::remove_file(rollback.join(name)))
+                    && let Err(err) = ignore_not_found(unlink_child_file(rollback_dir_fd.as_deref(), name, &rollback.join(name)))
                 {
+                    drop(rollback_dir_fd.take());
+                    drop(file_path_dir.take());
                     if has_target || had_physical_data_dirs {
                         let _ = persist_receipt(true);
                     }
@@ -3294,7 +3559,9 @@ impl LocalDisk {
                     let path = rollback.join(name);
                     #[cfg(all(any(test, feature = "test-util"), not(windows)))]
                     os::prepared_publication_test_hooks::run(os::prepared_publication_test_hooks::Stage::Remove, &path);
-                    if let Err(err) = ignore_not_found(std::fs::remove_file(path)) {
+                    if let Err(err) = ignore_not_found(unlink_child_file(rollback_dir_fd.as_deref(), name, &path)) {
+                        drop(rollback_dir_fd.take());
+                        drop(file_path_dir.take());
                         if has_target || had_physical_data_dirs {
                             let _ = persist_receipt(true);
                         }
@@ -3303,7 +3570,10 @@ impl LocalDisk {
                 }
             } else {
                 for name in [STORAGE_FORMAT_FILE_BACKUP, DELETE_MARKER_ROLLBACK_FILE] {
-                    if let Err(err) = ignore_not_found(std::fs::remove_file(rollback.join(name))) {
+                    if let Err(err) = ignore_not_found(unlink_child_file(rollback_dir_fd.as_deref(), name, &rollback.join(name)))
+                    {
+                        drop(rollback_dir_fd.take());
+                        drop(file_path_dir.take());
                         if has_target || had_physical_data_dirs {
                             let _ = persist_receipt(true);
                         }
@@ -3314,7 +3584,9 @@ impl LocalDisk {
                     let path = rollback.join(name);
                     #[cfg(all(any(test, feature = "test-util"), not(windows)))]
                     os::prepared_publication_test_hooks::run(os::prepared_publication_test_hooks::Stage::Remove, &path);
-                    if let Err(err) = ignore_not_found(std::fs::remove_file(path)) {
+                    if let Err(err) = ignore_not_found(unlink_child_file(rollback_dir_fd.as_deref(), name, &path)) {
+                        drop(rollback_dir_fd.take());
+                        drop(file_path_dir.take());
                         if has_target || had_physical_data_dirs {
                             let _ = persist_receipt(true);
                         }
@@ -3322,27 +3594,39 @@ impl LocalDisk {
                     }
                 }
             }
-            let rollback_sync = rollback_dir_fd.take();
-            if sync && let Err(err) = os::fsync_directory_handle_std(&rollback, rollback_sync.as_deref()) {
-                drop(rollback_sync);
+            if sync && let Err(err) = os::fsync_directory_handle_std(&rollback, rollback_dir_fd.as_deref()) {
+                drop(rollback_dir_fd.take());
+                drop(file_path_dir.take());
                 if has_target || had_physical_data_dirs {
                     let _ = persist_receipt(true);
                 }
                 return Err(err);
             }
-            drop(rollback_sync);
             #[cfg(all(any(test, feature = "test-util"), not(windows)))]
             os::prepared_publication_test_hooks::run(os::prepared_publication_test_hooks::Stage::Remove, &receipt_path);
-            if let Err(err) = ignore_not_found(std::fs::remove_file(&receipt_path)) {
+            let receipt_unlink_res =
+                ignore_not_found(unlink_child_file(rollback_dir_fd.as_deref(), DELETE_QUORUM_COMMITTED_FILE, &receipt_path));
+            drop(rollback_dir_fd.take());
+            if let Err(err) = receipt_unlink_res {
+                drop(file_path_dir.take());
                 if has_target || had_physical_data_dirs {
                     let _ = persist_receipt(true);
                 }
                 return Err(err);
             }
-            for dir_path in &prune_paths {
+            let tx_name = transaction.to_string();
+            for (idx, dir_path) in prune_paths.iter().enumerate() {
                 #[cfg(all(any(test, feature = "test-util"), not(windows)))]
                 os::prepared_publication_test_hooks::run(os::prepared_publication_test_hooks::Stage::Remove, dir_path);
-                if let Err(err) = std::fs::remove_dir(dir_path) {
+                let rmdir_res = if idx == 0 && *dir_path == rollback {
+                    let r = rmdir_child_dir(file_path_dir.as_deref(), &tx_name, dir_path);
+                    drop(file_path_dir.take());
+                    r
+                } else {
+                    drop(file_path_dir.take());
+                    std::fs::remove_dir(dir_path)
+                };
+                if let Err(err) = rmdir_res {
                     if is_dir_not_empty_error(&err) {
                         break;
                     }
@@ -5834,6 +6118,7 @@ mod tests {
                 drop(disk);
                 let endpoint = Endpoint::try_from(temp.path().to_str().expect("UTF-8 disk path")).expect("endpoint");
                 let disk = LocalDisk::new(&endpoint, false).await.expect("reopen after cleanup");
+                let root = disk.io_get_object_path("recovery", "object").expect("reopened object path");
                 let observed = disk
                     .inspect_delete_transaction("recovery", "object", transaction)
                     .await

crates/protos/src/lib.rs

@@ -4007,9 +4007,20 @@ mod tests {
 
         let _test_guard = TLS02_TEST_LOCK.lock().await;
         let previous = runtime_sources::outbound_tls_state().await;
-        let restore_generation = previous.generation.0.checked_add(3).expect("test generation capacity");
+        let cases = [
+            ("DNS", "127.0.0.1:0", "localhost", "localhost", true),
+            ("IPv4", "127.0.0.1:0", "127.0.0.1", "127.0.0.1", true),
+            ("IPv6", "[::1]:0", "[::1]", "::1", true),
+            ("IPv6 name mismatch", "[::1]:0", "[::1]", "localhost", false),
+        ];
+        let restore_offset = u64::try_from(cases.len() + 1).expect("bounded TLS case count");
+        let restore_generation = previous
+            .generation
+            .0
+            .checked_add(restore_offset)
+            .expect("test generation capacity");
         let mut observations = Vec::new();
-        for (offset, certificate_name) in [(1, "::1"), (2, "localhost")] {
+        for (index, (case, listen_addr, dial_host, certificate_name, expected)) in cases.into_iter().enumerate() {
             let certificate =
                 rcgen::generate_simple_self_signed(vec![certificate_name.to_owned()]).expect("ephemeral TLS test certificate");
             let certificate_pem = certificate.cert.pem();
@@ -4017,10 +4028,11 @@ mod tests {
                 &certificate_pem,
                 certificate.signing_key.serialize_pem(),
             ));
-            let listener = tokio::net::TcpListener::bind("[::1]:0")
+            let listener = tokio::net::TcpListener::bind(listen_addr)
                 .await
-                .expect("IPv6 loopback TLS listener");
-            let addr = format!("https://{}", listener.local_addr().expect("IPv6 test server address"));
+                .expect("loopback TLS listener");
+            let port = listener.local_addr().expect("TLS test server address").port();
+            let addr = format!("https://{dial_host}:{port}/data");
             let accepted = Arc::new(AtomicUsize::new(0));
             let counter = Arc::clone(&accepted);
             let incoming = TcpListenerStream::new(listener).map(move |stream| {
@@ -4042,20 +4054,30 @@ mod tests {
                 root_ca_pem: certificate_pem.into_bytes(),
                 mtls_identity: None,
             };
+            let offset = u64::try_from(index + 1).expect("bounded generation offset");
             publish_global_outbound_tls_state(TlsGeneration(previous.generation.0 + offset), &material).await;
             let mut results = Vec::new();
             for key in [addr.clone(), bulk_cache_key(&addr, 0)] {
-                results.push(
-                    build_channel(&addr, &key)
-                        .await
-                        .map(|_| ())
-                        .map_err(|error| error.to_string()),
-                );
+                let (connected, reached_rpc) = match build_channel(&addr, &key).await {
+                    Ok(channel) => {
+                        let mut client = NodeServiceClient::new(channel);
+                        let response = tokio::time::timeout(
+                            Duration::from_secs(3),
+                            client.ping(proto_gen::node_service::PingRequest::default()),
+                        )
+                        .await;
+                        // The empty router answers only after TLS and HTTP/2 reach the server.
+                        let reached_rpc = matches!(response, Ok(Err(status)) if status.code() == tonic::Code::Unimplemented);
+                        (Ok(()), reached_rpc)
+                    }
+                    Err(error) => (Err(error.to_string()), false),
+                };
+                results.push((connected, reached_rpc));
                 evict_cached_channel(&key, ConnectionEvictionLogLevel::Debug).await;
             }
             server.abort();
             let _ = server.await;
-            observations.push((results, accepted.load(Ordering::SeqCst)));
+            observations.push((case, expected, results, accepted.load(Ordering::SeqCst)));
         }
         let restore = OutboundTlsMaterial {
             root_ca_pem: previous.root_ca_pem.unwrap_or_default(),
@@ -4063,19 +4085,31 @@ mod tests {
         };
         publish_global_outbound_tls_state(TlsGeneration(restore_generation), &restore).await;
 
-        assert!(
-            observations[0].0.iter().all(Result::is_ok),
-            "IPv6 control and bulk channels must verify the URI host: {:?}",
-            observations[0].0
-        );
-        assert!(
-            observations[1].0.iter().all(Result::is_err),
-            "a trusted certificate for localhost must be rejected for ::1"
-        );
-        assert!(
-            observations.iter().all(|(_, accepted)| *accepted == 2),
-            "both certificate cases must exercise actual control and bulk TLS connections"
+        let restored = runtime_sources::outbound_tls_state().await;
+        assert_eq!(restored.root_ca_pem.as_deref().unwrap_or_default(), restore.root_ca_pem.as_slice());
+        assert_eq!(
+            restored
+                .mtls_identity
+                .as_ref()
+                .map(|identity| (&identity.cert_pem, &identity.key_pem)),
+            restore
+                .mtls_identity
+                .as_ref()
+                .map(|identity| (&identity.cert_pem, &identity.key_pem)),
         );
+        assert!(restored.generation.0 > previous.generation.0);
+
+        for (case, expected, results, accepted) in observations {
+            for (connected, reached_rpc) in results {
+                assert_eq!(
+                    connected.is_ok(),
+                    expected,
+                    "{case} must verify the URI certificate identity: {connected:?}"
+                );
+                assert_eq!(reached_rpc, expected, "{case} must reach the RPC service only after successful TLS");
+            }
+            assert_eq!(accepted, 2, "{case} must exercise actual control and bulk connections");
+        }
     }
 
     #[tokio::test]

crates/s3-client/src/transition_api.rs

@@ -60,7 +60,9 @@ use rustfs_utils::{
         AMZ_CHECKSUM_SHA256,
     },
     net::get_endpoint_url,
-    retry::{DEFAULT_RETRY_CAP, DEFAULT_RETRY_UNIT, MAX_JITTER, MAX_RETRY, RetryTimer},
+    retry::{
+        DEFAULT_RETRY_CAP, DEFAULT_RETRY_UNIT, MAX_JITTER, MAX_RETRY, RetryTimer, is_http_status_retryable, is_s3code_retryable,
+    },
 };
 use rustls_pki_types::PrivateKeyDer;
 use rustls_pki_types::pem::PemObject;
@@ -119,6 +121,33 @@ fn transition_transport_error(err: hyper_util::client::legacy::Error) -> std::io
     std::io::Error::other(err)
 }
 
+fn is_retryable_transport_error(error: &std::io::Error) -> bool {
+    // Hyper wraps connection and body errors; the outer io::Error is often Other.
+    let mut current: Option<&(dyn StdError + 'static)> = Some(error);
+    while let Some(error) = current {
+        if error.downcast_ref::<std::io::Error>().is_some_and(|error| {
+            matches!(
+                error.kind(),
+                std::io::ErrorKind::Interrupted
+                    | std::io::ErrorKind::WouldBlock
+                    | std::io::ErrorKind::TimedOut
+                    | std::io::ErrorKind::ConnectionAborted
+                    | std::io::ErrorKind::ConnectionRefused
+                    | std::io::ErrorKind::ConnectionReset
+                    | std::io::ErrorKind::NotConnected
+                    | std::io::ErrorKind::UnexpectedEof
+            )
+        }) || error
+            .downcast_ref::<hyper::Error>()
+            .is_some_and(|error| error.is_incomplete_message() || error.is_closed())
+        {
+            return true;
+        }
+        current = error.source();
+    }
+    false
+}
+
 async fn next_response_body_data<B>(
     mut body: Pin<&mut B>,
     idle_timeout: Option<StdDuration>,
@@ -727,6 +756,18 @@ impl TransitionClient {
         Ok(resp)
     }
 
+    /// Executes a remote request, retrying transient failures up to `max_retries` total attempts.
+    ///
+    /// Only GET, HEAD, and PUT of a fixed multipart upload part with a buffered body
+    /// are replayed. Object readers and other mutations get one attempt: replaying
+    /// ordinary PUT or DELETE can create additional versions or delete markers.
+    /// Request construction failures and errors reading a successful response body
+    /// are not retried. Dropping this future stops further attempts and any pending
+    /// backoff; a failed or canceled request may still have reached the remote server.
+    ///
+    /// # Errors
+    /// Returns a request construction error, a permanent remote error, or the last
+    /// transient error when the attempt budget is exhausted.
     pub async fn execute_method(
         &self,
         method: http::Method,
@@ -738,24 +779,27 @@ impl TransitionClient {
             return Err(std::io::Error::other(s));
         }
 
-        let retryable: bool;
-        //let mut body_seeker: BufferReader;
-        let mut req_retry = self.max_retries;
-        let mut resp: Response<Incoming>;
-
-        //if metadata.content_body != nil {
-        //body_seeker = BufferReader::new(metadata.content_body.read_all().await?);
-        retryable = true;
-        if !retryable {
-            req_retry = 1;
-        }
-        //}
+        let retryable = matches!(&metadata.content_body, ReaderImpl::Body(_))
+            && (method == http::Method::GET
+                || method == http::Method::HEAD
+                || (method == http::Method::PUT
+                    && metadata.query_values.get("uploadId").is_some_and(|value| !value.is_empty())
+                    && metadata.query_values.get("partNumber").is_some_and(|value| !value.is_empty())));
+        let req_retry = if retryable { self.max_retries.max(1) } else { 1 };
+        let mut last_error = None;
 
         let mut retry_timer = RetryTimer::new(req_retry, DEFAULT_RETRY_UNIT, DEFAULT_RETRY_CAP, MAX_JITTER, self.random);
         while retry_timer.next().await.is_some() {
             let req = self.new_request(&method, metadata).await?;
 
-            resp = self.doit(req).await?;
+            let resp = match self.doit(req).await {
+                Ok(resp) => resp,
+                Err(error) if retryable && is_retryable_transport_error(&error) => {
+                    last_error = Some(error);
+                    continue;
+                }
+                Err(error) => return Err(error),
+            };
 
             for http_status in SUCCESS_STATUS {
                 if http_status == resp.status() {
@@ -766,11 +810,17 @@ impl TransitionClient {
             let resp_status = resp.status();
             let h = resp.headers().clone();
 
-            let body_vec = self
-                .collect_response_body(resp.into_body(), MAX_S3_ERROR_RESPONSE_SIZE)
-                .await?;
+            let body_vec = match self.collect_response_body(resp.into_body(), MAX_S3_ERROR_RESPONSE_SIZE).await {
+                Ok(body) => body,
+                Err(error) if retryable && is_retryable_transport_error(&error) => {
+                    last_error = Some(error);
+                    continue;
+                }
+                Err(error) => return Err(error),
+            };
             let parsed_error =
                 http_resp_to_error_response(resp_status, &h, body_vec, &metadata.bucket_name, &metadata.object_name);
+            let retryable_response = is_http_status_retryable(&resp_status) || is_s3code_retryable(parsed_error.code.as_str());
             let routing_region = parsed_error.region;
             let code = match parsed_error.code {
                 code if code.is_custom() => RemoteErrorCode::RESPONSE_INTERRUPTED,
@@ -785,6 +835,11 @@ impl TransitionClient {
                 ..Default::default()
             };
 
+            if retryable && retryable_response {
+                last_error = Some(std::io::Error::other(err_response));
+                continue;
+            }
+
             if self.region == "" {
                 return match err_response.code {
                     RemoteErrorCode::AUTHORIZATION_HEADER_MALFORMED | RemoteErrorCode::INVALID_ARGUMENT /*RemoteErrorCode::INVALID_REGION*/ => {
@@ -819,7 +874,7 @@ impl TransitionClient {
             return Err(std::io::Error::other(err_response));
         }
 
-        Err(std::io::Error::other("remote tier request did not produce a response"))
+        Err(last_error.unwrap_or_else(|| std::io::Error::other("remote tier request did not produce a response")))
     }
 
     pub async fn collect_response_body<B>(&self, body: B, limit: usize) -> Result<Vec<u8>, std::io::Error>

crates/s3-client/tests/transition_retries.rs

@@ -0,0 +1,320 @@
+// Copyright 2026 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use bytes::Bytes;
+use http::{Method, StatusCode};
+use rustfs_s3_client::{
+    api_error_response::to_error_response,
+    credentials::{Credentials, SignatureType, Static, Value},
+    remote_error_code::RemoteErrorCode,
+    transition_api::{
+        BucketLookupType, MAX_S3_ERROR_RESPONSE_SIZE, ObjectReader, Options, ReaderImpl, RequestMetadata, TransitionClient,
+        TransitionClientTimeouts,
+    },
+};
+use std::time::Duration;
+use tokio::{
+    io::{AsyncReadExt, AsyncWriteExt},
+    net::TcpListener,
+};
+
+#[derive(Clone, Copy)]
+enum Reply {
+    Http(StatusCode, &'static str),
+    Disconnect,
+    Truncated,
+    StallHeaders,
+    MalformedHttp,
+    Oversized,
+}
+
+fn metadata(body: ReaderImpl, content_length: i64) -> RequestMetadata {
+    RequestMetadata {
+        bucket_name: "bucket".into(),
+        object_name: "object".into(),
+        bucket_location: "us-east-1".into(),
+        content_body: body,
+        content_length,
+        pre_sign_url: false,
+        query_values: Default::default(),
+        custom_header: Default::default(),
+        extra_pre_sign_header: None,
+        expires: 0,
+        content_md5_base64: String::new(),
+        content_sha256_hex: String::new(),
+        stream_sha256: false,
+        trailer: Default::default(),
+    }
+}
+
+fn empty_metadata() -> RequestMetadata {
+    metadata(ReaderImpl::Body(Bytes::new()), 0)
+}
+
+async fn run_case(
+    method: Method,
+    metadata: &mut RequestMetadata,
+    attempts: i64,
+    replies: Vec<Reply>,
+) -> (Result<StatusCode, std::io::Error>, Vec<Vec<u8>>) {
+    let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind retry fixture");
+    let endpoint = listener.local_addr().expect("fixture address").to_string();
+    let (stop, mut stopped) = tokio::sync::oneshot::channel();
+    let server = tokio::spawn(async move {
+        let mut requests = Vec::new();
+        let mut stalled_sockets = Vec::new();
+        loop {
+            let (mut socket, _) = tokio::select! {
+                _ = &mut stopped => break,
+                accepted = listener.accept() => accepted.expect("accept retry request"),
+            };
+            let mut request = Vec::new();
+            let header_end = loop {
+                let mut buffer = [0; 1024];
+                let size = socket.read(&mut buffer).await.expect("read request headers");
+                assert_ne!(size, 0, "client closed before sending headers");
+                request.extend_from_slice(&buffer[..size]);
+                if let Some(end) = request.windows(4).position(|part| part == b"\r\n\r\n") {
+                    break end + 4;
+                }
+            };
+            let headers = String::from_utf8_lossy(&request[..header_end]).to_ascii_lowercase();
+            assert!(headers.contains("authorization: aws4-hmac-sha256"), "each attempt must be signed");
+            let length = headers
+                .lines()
+                .find_map(|line| line.strip_prefix("content-length: "))
+                .map(|value| value.parse::<usize>().expect("numeric content length"))
+                .unwrap_or(0);
+            while request.len() < header_end + length {
+                let mut buffer = [0; 1024];
+                let size = socket.read(&mut buffer).await.expect("read request body");
+                assert_ne!(size, 0, "client closed before sending the complete body");
+                request.extend_from_slice(&buffer[..size]);
+            }
+            let reply = replies[requests.len().min(replies.len() - 1)];
+            requests.push(request);
+            let (status, body, extra_length) = match reply {
+                Reply::Disconnect => continue,
+                Reply::StallHeaders => {
+                    stalled_sockets.push(socket);
+                    continue;
+                }
+                Reply::MalformedHttp => {
+                    socket
+                        .write_all(b"invalid HTTP\r\n\r\n")
+                        .await
+                        .expect("write malformed response");
+                    continue;
+                }
+                Reply::Truncated => (StatusCode::SERVICE_UNAVAILABLE, "<Error>".to_string(), 100),
+                Reply::Oversized => (StatusCode::SERVICE_UNAVAILABLE, "x".repeat(MAX_S3_ERROR_RESPONSE_SIZE + 1), 0),
+                Reply::Http(status, body) => (status, body.to_string(), 0),
+            };
+            let is_head = requests.last().expect("recorded request").starts_with(b"HEAD ");
+            let response = format!(
+                "HTTP/1.1 {status}\r\nContent-Length: {}\r\nConnection: close\r\nx-amz-request-id: fixture\r\n\r\n{}",
+                body.len() + extra_length,
+                if is_head { "" } else { &body },
+            );
+            socket.write_all(response.as_bytes()).await.expect("write fixture response");
+        }
+        requests
+    });
+    let client = TransitionClient::new_with_timeouts(
+        &endpoint,
+        Options {
+            creds: Credentials::new(Static(Value {
+                access_key_id: "access-key".into(),
+                secret_access_key: "secret-key".into(),
+                signer_type: SignatureType::SignatureV4,
+                ..Default::default()
+            })),
+            region: "us-east-1".into(),
+            bucket_lookup: BucketLookupType::BucketLookupPath,
+            max_retries: attempts,
+            ..Default::default()
+        },
+        "s3",
+        TransitionClientTimeouts::new(Duration::from_secs(1), Duration::from_secs(1), Duration::from_secs(1)),
+    )
+    .await
+    .expect("build real transition client");
+    let result = tokio::time::timeout(Duration::from_secs(5), client.execute_method(method, metadata)).await;
+    let _ = stop.send(());
+    let requests = tokio::time::timeout(Duration::from_secs(2), server)
+        .await
+        .expect("fixture must stop")
+        .expect("fixture must join");
+    (result.expect("retry loop must terminate").map(|response| response.status()), requests)
+}
+
+#[tokio::test]
+async fn retries_transient_http_statuses_then_succeeds() {
+    for method in [Method::GET, Method::HEAD] {
+        for status in [
+            StatusCode::SERVICE_UNAVAILABLE,
+            StatusCode::REQUEST_TIMEOUT,
+            StatusCode::TOO_MANY_REQUESTS,
+            StatusCode::INTERNAL_SERVER_ERROR,
+            StatusCode::BAD_GATEWAY,
+            StatusCode::GATEWAY_TIMEOUT,
+        ] {
+            let (result, requests) = run_case(
+                method.clone(),
+                &mut empty_metadata(),
+                2,
+                vec![Reply::Http(status, ""), Reply::Http(StatusCode::OK, "")],
+            )
+            .await;
+            assert_eq!(result.expect("transient status should recover"), StatusCode::OK);
+            assert_eq!(requests.len(), 2, "{method} {status} must issue a second request");
+        }
+    }
+}
+
+#[tokio::test]
+async fn retries_s3_error_codes_even_with_a_non_retryable_http_status() {
+    for body in [
+        "<Error><Code>SlowDown</Code></Error>",
+        "<Error><Code>Throttling</Code></Error>",
+        "<Error><Code>ThrottlingException</Code></Error>",
+        "<Error><Code>RequestTimeout</Code></Error>",
+    ] {
+        let (result, requests) = run_case(
+            Method::GET,
+            &mut empty_metadata(),
+            2,
+            vec![Reply::Http(StatusCode::BAD_REQUEST, body), Reply::Http(StatusCode::OK, "")],
+        )
+        .await;
+        assert_eq!(result.expect("transient S3 error should recover"), StatusCode::OK);
+        assert_eq!(requests.len(), 2, "{body} must trigger a retry");
+    }
+}
+
+#[tokio::test]
+async fn retries_disconnected_transport_and_interrupted_error_body() {
+    for first_reply in [Reply::Disconnect, Reply::Truncated, Reply::StallHeaders] {
+        let (result, requests) =
+            run_case(Method::GET, &mut empty_metadata(), 2, vec![first_reply, Reply::Http(StatusCode::OK, "")]).await;
+        assert_eq!(result.expect("interrupted request should recover"), StatusCode::OK);
+        assert_eq!(requests.len(), 2);
+    }
+}
+
+#[tokio::test]
+async fn retry_exhaustion_preserves_the_transport_timeout() {
+    let (result, requests) = run_case(Method::GET, &mut empty_metadata(), 2, vec![Reply::StallHeaders]).await;
+    assert_eq!(result.expect_err("both attempts timed out").kind(), std::io::ErrorKind::TimedOut);
+    assert_eq!(requests.len(), 2);
+}
+
+#[tokio::test]
+async fn malformed_and_oversized_responses_are_not_retried() {
+    for reply in [Reply::MalformedHttp, Reply::Oversized] {
+        let (result, requests) = run_case(Method::GET, &mut empty_metadata(), 3, vec![reply]).await;
+        let error = result.expect_err("invalid responses must fail");
+        if matches!(reply, Reply::Oversized) {
+            assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
+        }
+        assert_eq!(requests.len(), 1, "permanent protocol errors must not be replayed");
+    }
+}
+
+#[tokio::test]
+async fn invalid_request_headers_fail_before_dispatch() {
+    let mut metadata = empty_metadata();
+    metadata.content_md5_base64 = "invalid\nheader".into();
+    let (result, requests) = run_case(Method::GET, &mut metadata, 3, vec![Reply::Http(StatusCode::OK, "")]).await;
+    assert_eq!(result.expect_err("invalid header must fail").kind(), std::io::ErrorKind::InvalidInput);
+    assert!(requests.is_empty(), "request construction failures must not reach the server");
+}
+
+#[tokio::test]
+async fn retry_exhaustion_returns_the_last_s3_error() {
+    let (result, requests) = run_case(
+        Method::GET,
+        &mut empty_metadata(),
+        3,
+        vec![
+            Reply::Http(StatusCode::BAD_GATEWAY, ""),
+            Reply::Http(StatusCode::SERVICE_UNAVAILABLE, "<Error><Code>SlowDown</Code></Error>"),
+        ],
+    )
+    .await;
+    let error = to_error_response(&result.expect_err("all attempts failed"));
+    assert_eq!(requests.len(), 3, "max_retries counts total attempts");
+    assert_eq!(error.status_code, StatusCode::SERVICE_UNAVAILABLE);
+    assert_eq!(error.code, RemoteErrorCode::from_wire("SlowDown"));
+}
+
+#[tokio::test]
+async fn permanent_errors_and_single_attempts_are_not_retried() {
+    for (status, body, attempts) in [
+        (StatusCode::FORBIDDEN, "<Error><Code>AccessDenied</Code></Error>", 3),
+        (StatusCode::NOT_FOUND, "<Error><Code>NoSuchKey</Code></Error>", 3),
+        (StatusCode::BAD_REQUEST, "<Error><Code>InvalidArgument</Code></Error>", 3),
+        (StatusCode::SERVICE_UNAVAILABLE, "<Error><Code>SlowDown</Code></Error>", 1),
+        (StatusCode::OK, "", 3),
+    ] {
+        let (result, requests) = run_case(Method::GET, &mut empty_metadata(), attempts, vec![Reply::Http(status, body)]).await;
+        assert_eq!(result.is_ok(), status.is_success());
+        assert_eq!(requests.len(), 1, "{status} with {attempts} attempts");
+    }
+}
+
+#[tokio::test]
+async fn buffered_upload_parts_replay_identical_body_bytes() {
+    let mut metadata = metadata(ReaderImpl::Body(Bytes::from_static(b"part bytes")), 10);
+    metadata.query_values.insert("uploadId".into(), "upload-1".into());
+    metadata.query_values.insert("partNumber".into(), "1".into());
+    let (result, requests) = run_case(
+        Method::PUT,
+        &mut metadata,
+        2,
+        vec![
+            Reply::Http(StatusCode::SERVICE_UNAVAILABLE, ""),
+            Reply::Http(StatusCode::OK, ""),
+        ],
+    )
+    .await;
+    assert_eq!(result.expect("buffered upload part should recover"), StatusCode::OK);
+    assert_eq!(requests.len(), 2);
+    for request in requests {
+        assert!(request.ends_with(b"\r\n\r\npart bytes"), "every attempt must send the full body");
+    }
+}
+
+#[tokio::test]
+async fn non_replayable_bodies_and_object_mutations_are_not_retried() {
+    for method in [Method::GET, Method::PUT, Method::POST, Method::DELETE] {
+        let mut metadata = if method == Method::GET {
+            metadata(ReaderImpl::ObjectBody(ObjectReader::new(std::io::Cursor::new(b"stream"))), 6)
+        } else {
+            empty_metadata()
+        };
+        let (result, requests) =
+            run_case(method.clone(), &mut metadata, 3, vec![Reply::Http(StatusCode::SERVICE_UNAVAILABLE, "")]).await;
+        assert!(result.is_err(), "{method} must return the first failure");
+        assert_eq!(requests.len(), 1, "{method} must not be replayed");
+    }
+    let mut metadata = metadata(ReaderImpl::ObjectBody(ObjectReader::new(std::io::Cursor::new(b"stream"))), 6);
+    metadata.query_values.insert("uploadId".into(), "upload-1".into());
+    metadata.query_values.insert("partNumber".into(), "1".into());
+    let (result, requests) =
+        run_case(Method::PUT, &mut metadata, 3, vec![Reply::Http(StatusCode::SERVICE_UNAVAILABLE, "")]).await;
+    assert!(result.is_err(), "streaming upload parts must return the first failure");
+    assert_eq!(requests.len(), 1, "a consumed object reader must not be replayed");
+    assert!(requests[0].ends_with(b"\r\n\r\nstream"));
+}

crates/utils/src/retry.rs

@@ -65,10 +65,10 @@ impl Stream for RetryTimer {
 
         let jitter = self.jitter.clamp(NO_JITTER, MAX_JITTER);
         let attempt = self.max_retry - self.rem;
-        let mut sleep = self.base_sleep * (1 << attempt);
-        if sleep > self.max_sleep {
-            sleep = self.max_sleep;
-        }
+        let mut sleep = u32::try_from(attempt)
+            .ok()
+            .and_then(|shift| 1_u32.checked_shl(shift))
+            .map_or(self.max_sleep, |factor| self.base_sleep.saturating_mul(factor).min(self.max_sleep));
         if (jitter - NO_JITTER).abs() > 1e-9 {
             let sleep_ms = sleep.as_millis();
             let reduction = ((sleep_ms as f64) * (self.random as f64 * jitter / 100_f64)).round() as u128;
@@ -255,6 +255,15 @@ mod tests {
         assert_eq!(retry_timer.next().await, None);
     }
 
+    #[tokio::test]
+    async fn retry_timer_caps_backoff_for_large_attempt_budgets() {
+        let retry_timer = RetryTimer::new(40, Duration::from_millis(1), Duration::from_millis(2), NO_JITTER, 0);
+        let retries = timeout(Duration::from_secs(5), retry_timer.collect::<Vec<_>>())
+            .await
+            .expect("a large retry budget should finish without overflowing the backoff");
+        assert_eq!(retries.len(), 40);
+    }
+
     #[test]
     fn is_s3code_in_message_retryable_matches_each_retryable_code() {
         for code in [

docs/operations/release-pipeline.md

@@ -14,7 +14,7 @@ The builder workflows (`build.yml`, `package.yml`, `docker.yml`, `helm-package.y
 
 ## Freeze and build
 
-Select a reviewed source commit whose Cargo version equals the intended final version. Finish exact-source CI before dispatching the candidate. The candidate workflow itself must run from upstream main. Unrelated main commits do not require rebuilding an accepted source.
+Select a reviewed source commit whose Cargo version equals the intended final version. Finish exact-source CI before dispatching the candidate: a successful `push` or `workflow_dispatch` run of `ci.yml` for that commit on `main` or the release branch whose `Quick Checks`, `Test and Lint` and `End-to-End Tests (full merge gate)` all succeeded. A commit that landed through the merge queue has a push run whose validation jobs were skipped as proven; `Quick Checks` and `Test and Lint` must still succeed there, and the evidence recorded is the successful `merge_group` run of that branch's queue whose head commit has the same tree, which must itself have passed all three jobs. `scripts/ci_merge_group_proof.py` defines that match for both the push plan and the candidate. The candidate workflow itself must run from upstream main. Unrelated main commits do not require rebuilding an accepted source.
 
 ```bash
 gh workflow run release-candidate.yml --ref main \

docs/testing/ci-gates.md

@@ -39,6 +39,8 @@ The aggregate requires the validation lanes already selected by `ci.yml`; this c
 
 Superseded PR attempts are cancelled. A running main validation finishes, with only the newest pending main run retained, so frequent merges cannot continuously cancel the full baseline. Having a `merge_group` trigger does not itself require use of the merge queue.
 
+A push whose tree a successful `merge_group` run of `ci.yml` already validated does not validate it again. A merge-queue landing fast-forwards the branch to the group's commit, so `Select CI scope` looks that run up (`scripts/ci_merge_group_proof.py`, with the job's read-only Actions token) and plans mode `proven`: `Select CI scope`, `Typos` and `Quick Checks` run, every validation job is skipped, and `Test and Lint` passes only if each of them was skipped. The plan and the aggregate's summary name the proving run; a job skipped for any other reason keeps its own reason. A push that did not come through the queue, such as a direct administrator merge, has no such run, and any doubt about the lookup (no token, an API or network error, a rate limit, no matching successful run, a different tree) plans the full validation.
+
 Protocol matrix jobs finish independently when a sibling fails. Each executed protocol test step preserves its log and available JUnit under a separate, attempt-specific artifact; a skipped test step cannot upload a cached report.
 
 The four site-replication state-writer concurrency proofs reserve the available nextest slots in both local and CI profiles. This isolates their durable IO from unrelated test processes while retaining each proof's internal two-writer race, production five-second lock-acquisition limit, assertions, and zero retries.
@@ -70,7 +72,7 @@ The four site-replication state-writer concurrency proofs reserve the available
 | PR touching `paths` in `oidc-keycloak.yml` | `OIDC Keycloak live gate` | `oidc-keycloak.yml` `oidc-keycloak-live` | Report-only | `cargo build --locked -p rustfs --bin rustfs`, then `bash scripts/test/oidc_keycloak_live.sh ./target/debug/rustfs` |
 | PR touching `paths` in `targets-integration.yml` | `PostgreSQL, MySQL, AMQP, and NATS` | `targets-integration.yml` `targets-live` | Report-only | start the containers as in the job, export the `RUSTFS_TEST_*` DSNs, then the job's `cargo test --locked -p rustfs-targets --test <name> -- --ignored --test-threads=1` commands |
 | PR, documentation-only selection | `Quick Checks`, `Typos`, `Test and Lint` | `ci.yml` `quick-checks`, `typos`, `required-checks` | Required directly or via aggregate | Quick Checks commands; `python3 scripts/ci_gate.py --self-test` |
-| `merge_group`; push to `main` or `gateway/integration` | `End-to-End Tests (full merge gate)` | `ci.yml` `e2e-full` | Via aggregate on these events | `python3 scripts/e2e_binary.py run --features e2e-test-hooks -- cargo nextest run --profile e2e-full -p e2e_test` (CI adds `--binary "$RUSTFS_E2E_STARTUP_CAS_BINARY"` for the downloaded binary and sidecar) |
+| `merge_group`; push to `main` or `gateway/integration` that no successful `merge_group` run of the same tree proved | `End-to-End Tests (full merge gate)` | `ci.yml` `e2e-full` | Via aggregate on these events; skipped on a proven push, whose evidence is the proving run | `python3 scripts/e2e_binary.py run --features e2e-test-hooks -- cargo nextest run --profile e2e-full -p e2e_test` (CI adds `--binary "$RUSTFS_E2E_STARTUP_CAS_BINARY"` for the downloaded binary and sidecar) |
 
 e2e filters live in `.config/nextest.toml`; extend a profile instead of adding a second selector. Before a profile runs, `scripts/check_test_wiring.py` compares its listing to the committed digest in `.config/e2e-<profile>-selection.txt`, so a silent test drop fails closed.
 
@@ -86,7 +88,7 @@ Scheduled lanes never block a PR. Their workflow-local gate fails the run. The v
 
 | Workflow (cadence) | Jobs | Verdict and artifacts | Registry monitor | Reproduce |
 |---|---|---|---|---|
-| `ci.yml` (nightly) | full matrix, including the schedule/dispatch-only rio-v2 jobs `build-rustfs-debug-binary-rio-v2` and `e2e-tests-rio-v2` | strict aggregate; the full E2E lane runs on dispatch, merge groups, and main pushes | yes | dispatch `ci.yml` |
+| `ci.yml` (nightly) | full matrix, including the schedule/dispatch-only rio-v2 jobs `build-rustfs-debug-binary-rio-v2` and `e2e-tests-rio-v2` | strict aggregate; the full E2E lane runs on dispatch, merge groups, and main pushes that no merge group proved | yes | dispatch `ci.yml` |
 | `cache-warm.yml` (every six hours, plus main pushes that change `Cargo.lock`, `rust-toolchain.toml` or the setup action) | `warm-ci-dev`, `warm-ci-feat-rio`, `warm-ci-feat-proto`, `warm-ci-uring` | no verdict; sole writer of the `rust-cache` keys that `ci.yml` restores | yes | dispatch `cache-warm.yml` |
 | `build.yml` (nightly) | `build-rustfs` over the six-target platform matrix in `prepare-platform-matrix` (four Linux, macOS aarch64, Windows x86_64) | build/package integrity | yes | dispatch `build.yml` with an exact platform set |
 | `e2e-replication-nightly.yml` (nightly) | `repl-nightly`, `cluster-nightly`, `protocols-nightly` | three independent gates; JUnit, membership listing, server logs | yes | `python3 scripts/e2e_binary.py build --bins`, then `python3 scripts/e2e_binary.py run -- cargo nextest run --profile e2e-repl-nightly -p e2e_test`; `--features e2e-test-hooks` on both steps with `--profile e2e-nightly`; `--features ftps,webdav,sftp` on both steps with `-j 1 --profile e2e-protocols` |

rustfs/src/admin/service/session_policy.rs

@@ -13,14 +13,33 @@
 // limitations under the License.
 
 use rustfs_iam::sys::SESSION_POLICY_NAME;
-use rustfs_policy::policy::Policy;
+use rustfs_policy::policy::{ID, Policy, Statement};
 use rustfs_s3_types::{S3Error, S3ErrorCode, S3Result, s3_error};
+use serde::Deserialize;
 use serde_json::Value;
 use std::collections::HashMap;
 
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+struct SessionPolicyInput {
+    #[serde(default, rename = "Id", alias = "ID")]
+    id: ID,
+    #[serde(default, rename = "Version")]
+    version: String,
+    #[serde(default, rename = "Statement")]
+    statements: Vec<Statement>,
+}
+
+fn parse_session_policy(input: &str) -> rustfs_policy::error::Result<Policy> {
+    let SessionPolicyInput { id, version, statements } = serde_json::from_str(input)?;
+    let policy = Policy { id, version, statements };
+    policy.validate()?;
+    Ok(policy)
+}
+
 pub(crate) fn populate_session_policy(claims: &mut HashMap<String, Value>, policy: &str) -> S3Result<()> {
     if !policy.is_empty() {
-        let session_policy = Policy::parse_config(policy.as_bytes()).map_err(|e| {
+        let session_policy = parse_session_policy(policy).map_err(|e| {
             let error_msg = format!(
                 "Failed to parse session policy: {}. Please check that the policy is valid JSON format with standard brackets [] for arrays.",
                 e
@@ -59,27 +78,50 @@ mod tests {
     }
 
     #[test]
-    fn valid_policy_is_encoded_in_the_session_claim() {
-        let mut claims = HashMap::new();
-        let policy = r#"{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["s3:GetObject"],"Resource":["arn:aws:s3:::bucket/*"]}]}"#;
+    fn session_policy_claim_preserves_the_iam_policy_format() {
+        for id_key in [Some("Id"), Some("ID"), None] {
+            let mut input = serde_json::json!({
+                "Version": "2012-10-17",
+                "Statement": [{"Effect": "Allow", "Action": ["s3:GetObject"], "Resource": ["arn:aws:s3:::bucket/*"]}]
+            });
+            if let Some(key) = id_key {
+                input[key] = "read-objects".into();
+            }
+            if id_key == Some("Id") {
+                assert!(Policy::parse_config(input.to_string().as_bytes()).is_err());
+            }
+            let mut claims = HashMap::new();
+            populate_session_policy(&mut claims, &input.to_string()).expect("valid session policy");
 
-        populate_session_policy(&mut claims, policy).expect("valid policy should be accepted");
-
-        let encoded = claims
-            .get(SESSION_POLICY_NAME)
-            .and_then(Value::as_str)
-            .expect("session policy claim should be present");
-        let decoded = base64_simd::URL_SAFE_NO_PAD
-            .decode_to_vec(encoded.as_bytes())
-            .expect("session policy claim should be base64url encoded");
-        let decoded_policy: Value = serde_json::from_slice(&decoded).expect("session policy claim should contain JSON");
-        assert_eq!(decoded_policy["Version"], "2012-10-17");
+            let encoded = claims[SESSION_POLICY_NAME].as_str().expect("encoded session policy");
+            let decoded = base64_simd::URL_SAFE_NO_PAD
+                .decode_to_vec(encoded.as_bytes())
+                .expect("base64url claim");
+            let json: Value = serde_json::from_slice(&decoded).expect("JSON claim");
+            assert!(json.get("Id").is_none());
+            assert_eq!(json.get("ID").and_then(Value::as_str), id_key.map(|_| "read-objects"));
+            let policy = Policy::parse_config(&decoded).expect("claim must be readable by the IAM policy parser");
+            assert_eq!(policy.version, "2012-10-17");
+            assert_eq!(policy.statements.len(), 1);
+        }
     }
 
     #[test]
     fn invalid_policy_is_rejected_as_invalid_request() {
-        let mut claims = HashMap::new();
-        let error = populate_session_policy(&mut claims, "not-json").expect_err("invalid policy should fail");
-        assert_eq!(error.code(), &S3ErrorCode::InvalidRequest);
+        for input in [
+            "not-json",
+            r#"{"Id":"a","ID":"b","Version":"2012-10-17"}"#,
+            r#"{"Id":"a","Id":"b","Version":"2012-10-17"}"#,
+            r#"{"id":"a","Version":"2012-10-17"}"#,
+            r#"{"Id":42,"Version":"2012-10-17"}"#,
+            r#"{"Id":"a"}"#,
+            r#"{"Id":"a","Version":"invalid"}"#,
+            r#"{"Id":"a","Version":"2012-10-17","Statement":[{"Effect":"Allow"}]}"#,
+        ] {
+            let mut claims = HashMap::new();
+            let error = populate_session_policy(&mut claims, input).expect_err("invalid session policy");
+            assert_eq!(error.code(), &S3ErrorCode::InvalidRequest, "{input}");
+            assert!(claims.is_empty());
+        }
     }
 }

scripts/ci_gate.py

@@ -33,7 +33,7 @@ NIGHTLY_JOBS = ("connect-feature-boundaries", "fips-provider", "uring-integratio
 # against this revision's expected job contract. The workflow refuses a plan
 # that does not carry the current marker and falls back to the full matrix, so
 # a base planner predating the contract cannot authorize a narrower selection.
-PLAN_CONTRACT = "4"
+PLAN_CONTRACT = "5"
 # These jobs still exercise the server and its black-box harness when only E2E
 # Rust sources change. Production code, manifests and shared test configuration
 # always select the full matrix. No production package depends on e2e_test.
@@ -311,7 +311,8 @@ def closure(seeds: set[str], graph: dict[str, set[str]], reverse: bool = False)
 def expected_results(mode: str, event: str, ref: str, connect: bool = True, action: str = "", rehearse_shards: bool = False) -> dict[str, str]:
     if event not in EVENTS:
         raise ValueError(f"unsupported CI event: {event!r}")
-    if mode not in ("docs", "e2e", "full") or (mode != "full" and event != "pull_request"):
+    # `proven`: a push whose tree a successful merge-group run already validated.
+    if mode not in ("docs", "e2e", "full", "proven") or (mode != "full" and event != ("push" if mode == "proven" else "pull_request")):
         raise ValueError(f"invalid CI selection: {mode!r} for {event!r}")
     if type(rehearse_shards) is not bool or (rehearse_shards and event != "workflow_dispatch"):
         raise ValueError("shard rehearsal requires an explicit manual event")
@@ -376,7 +377,8 @@ def configurations(jobs: dict[str, str]) -> list[dict]:
              "default_features": True, "features": features} for identity, job, profile, features in values if jobs[job] == "success"]
 
 
-def select_plan(event: str, base: str, head: str, ref: str, root: Path, action: str = "", rehearse_shards: bool = False) -> dict:
+def select_plan(event: str, base: str, head: str, ref: str, root: Path, action: str = "", rehearse_shards: bool = False,
+                *, proof: dict | None = None) -> dict:
     if event not in EVENTS or not SHA.fullmatch(head):
         raise ValueError("unsupported event or missing candidate SHA")
     tree = git(root, "rev-parse", f"{head}^{{tree}}").strip()
@@ -473,6 +475,10 @@ def select_plan(event: str, base: str, head: str, ref: str, root: Path, action:
     effective_packages = all_packages if shadow and mode != "docs" else sorted(selected)
     effective_connect = mode == "full" if shadow else connect
     effective_full = mode == "full" if shadow else full
+    proven = event == "push" and mode == "full" and isinstance(proof, dict) and proof.get("tree") == tree
+    if proven:
+        mode = "proven"
+        reasons.append(f"merge_group run {proof['url']} already validated tree {tree}")
     jobs = expected_results(mode, event, ref, effective_connect, action, rehearse_shards)
     plan = {"schema_version": 1, "event": event, "action": action, "ref": ref, "base": base if SHA.fullmatch(base) else "",
             "head": head, "tree": tree, "policy": {"revision": policy_revision, "rollout": "shadow" if shadow else "enforced"},
@@ -481,7 +487,9 @@ def select_plan(event: str, base: str, head: str, ref: str, root: Path, action:
             "gateway_smoke": gateway_smoke_selection(root, event, base, head, changed),
             "configurations": configurations(jobs), "jobs": jobs, "proposed": proposed,
             "reasons": ["shadow rollout preserves current validation responsibility"] + reasons if shadow else reasons,
-            "not_applicable": {job: f"not required for {mode}/{event}" for job, result in jobs.items() if result == "skipped"}}
+            "not_applicable": {job: f"proven by {proof['url']}" if proven else f"not required for {mode}/{event}"
+                               for job, result in jobs.items() if result == "skipped"},
+            "proven_by": proof if proven else None}
     return plan
 
 
@@ -513,6 +521,12 @@ def validate_plan(plan: object, event: str, ref: str, head: str = "", tree: str
         raise ValueError("invalid plan packages")
     if plan.get("mode") != "docs" and not packages:
         raise ValueError("code validation plan must not be empty")
+    proof = plan.get("proven_by")
+    if (proof is not None or plan.get("mode") == "proven") and not (
+            isinstance(proof, dict) and plan.get("mode") == "proven" and proof.get("tree") == plan["tree"]
+            and type(proof.get("run_id")) is int and proof["run_id"] > 0
+            and re.fullmatch(rf"https://github\.com/[\w.-]+/[\w.-]+/actions/runs/{proof['run_id']}", str(proof.get("url")))):
+        raise ValueError("a proven push requires the merge-group run that validated its tree")
     expected = expected_results(plan.get("mode"), event, ref, plan["connect"], plan.get("action", ""), plan["rehearse_shards"])
     if plan.get("jobs") != expected:
         raise ValueError("plan job contract differs from selected responsibilities")
@@ -800,7 +814,7 @@ def check_workflow(root: Path) -> list[str]:
     # Truth tables protect behavior while allowing equivalent expression and YAML formatting.
     for event, rehearse_shards in [(event, rehearsal) for event in EVENTS
                                    for rehearsal in ((False, True) if event == "workflow_dispatch" else (False,))]:
-        for mode in (("docs", "e2e", "full") if event == "pull_request" else ("full",)):
+        for mode in {"pull_request": ("docs", "e2e", "full"), "push": ("full", "proven")}.get(event, ("full",)):
             for ref in ("refs/heads/main", "refs/heads/release"):
                 for connect in (True, False) if event == "pull_request" and mode == "full" else (mode == "full",):
                     for action in ("opened", "closed") if event == "pull_request" else ("",):
@@ -877,9 +891,14 @@ def main() -> int:
         return not unittest.TextTestRunner(verbosity=2).run(suite).wasSuccessful()
     if sys.argv[1:] == ["select"]:
         try:
+            # scripts/ci_merge_group_proof.py output; select_plan accepts it only for a push of the same tree.
+            try:
+                proof = json.loads(os.environ.get("CI_MERGE_GROUP_PROOF") or "null")
+            except ValueError:
+                proof = None
             plan = select_plan(os.environ.get("GITHUB_EVENT_NAME", ""), os.environ.get("CI_BASE_SHA", ""),
-                               os.environ.get("GITHUB_SHA", ""), os.environ.get("GITHUB_REF", ""),
-                               Path.cwd(), os.environ.get("CI_EVENT_ACTION", ""), os.environ.get("CI_REHEARSE_SHARDS", "false") == "true")
+                               os.environ.get("GITHUB_SHA", ""), os.environ.get("GITHUB_REF", ""), Path.cwd(),
+                               os.environ.get("CI_EVENT_ACTION", ""), os.environ.get("CI_REHEARSE_SHARDS", "false") == "true", proof=proof)
             with open(os.environ["GITHUB_OUTPUT"], "a") as output:
                 for name, value in (("mode", plan["mode"]), ("plan", json.dumps(plan, separators=(",", ":"))),
                                     ("connect", str(plan["connect"]).lower()), ("packages", json.dumps(plan["packages"], separators=(",", ":"))),
@@ -907,7 +926,16 @@ def main() -> int:
                 base = (payload["pull_request"]["base"]["sha"] if event == "pull_request" else
                         payload["merge_group"]["base_sha"] if event == "merge_group" else payload.get("before", ""))
                 action = payload.get("action", "")
-            errors = verify_results(json.loads(os.environ["CI_NEEDS"]), event, os.environ.get("GITHUB_REF", ""), head, tree, base, action)
+            needs = json.loads(os.environ["CI_NEEDS"])
+            errors = verify_results(needs, event, os.environ.get("GITHUB_REF", ""), head, tree, base, action)
+            plan = json.loads(needs["classify-changes"]["outputs"].get("plan") or "null")
+            if not errors and isinstance(plan, dict) and plan["mode"] == "proven":
+                # The check list shows these jobs as skipped like any other; say why.
+                note = f"Validation jobs were skipped as proven: merge_group run {plan['proven_by']['url']} passed tree {plan['tree']}."
+                print(note)
+                if os.environ.get("GITHUB_STEP_SUMMARY"):
+                    with open(os.environ["GITHUB_STEP_SUMMARY"], "a") as summary:
+                        summary.write(note + "\n")
         except (OSError, KeyError, TypeError, ValueError, subprocess.CalledProcessError) as error:
             errors = [str(error)]
     elif sys.argv[1:] == ["--check-workflow"]:

scripts/ci_merge_group_proof.py

@@ -0,0 +1,66 @@
+#!/usr/bin/env python3
+"""Find the successful merge-group run of ci.yml that already validated a pushed tree.
+
+A merge-queue landing fast-forwards the branch to the group's commit, whose
+tree that run validated. Any doubt (no token, an API or network error, a
+rate limit, an unexpected payload, no matching run) returns None, and the
+caller then validates the push in full. A push that did not come through the
+queue has no such run.
+"""
+import json
+import os
+import re
+import subprocess
+import urllib.request
+
+SHA = re.compile(r"[0-9a-f]{40}")
+
+
+def lookup(repository: str, tree: str, branch: str, token: str, opener=None) -> dict | None:
+    if not (token and SHA.fullmatch(tree) and re.fullmatch(r"[\w.-]+/[\w.-]+", repository)
+            and re.fullmatch(r"[\w.-]+(/[\w.-]+)*", branch) and ".." not in branch):
+        return None
+    request = urllib.request.Request(
+        f"https://api.github.com/repos/{repository}/actions/workflows/ci.yml/runs?event=merge_group&status=success&per_page=100",
+        headers={"Authorization": f"Bearer {token}", "Accept": "application/vnd.github+json", "X-GitHub-Api-Version": "2022-11-28"})
+    try:
+        with (opener or urllib.request.urlopen)(request, timeout=20) as response:
+            runs = json.load(response)["workflow_runs"]
+    except (OSError, ValueError, KeyError, TypeError):
+        return None
+    if not isinstance(runs, list):
+        return None
+    for run in runs:
+        if proves(run, repository, branch, tree):
+            return {"run_id": run["id"], "url": run["html_url"], "tree": tree}
+    return None
+
+
+def proves(run, repository: str, branch: str, tree: str) -> bool:
+    """Whether `run`, listed from ci.yml's runs, is a successful merge-queue run of `branch` in `repository` on `tree`.
+
+    The push plan and the release pipeline both accept a tree on this rule alone.
+    """
+    return (isinstance(run, dict) and run.get("event") == "merge_group" and run.get("status") == "completed"
+            and run.get("conclusion") == "success" and isinstance(run.get("head_branch"), str)
+            and run["head_branch"].startswith(f"gh-readonly-queue/{branch}/")
+            and isinstance(run.get("head_commit"), dict) and run["head_commit"].get("tree_id") == tree
+            and type(run.get("id")) is int and run["id"] > 0 and isinstance(run.get("html_url"), str)
+            and run["html_url"] == f"https://github.com/{repository}/actions/runs/{run['id']}")
+
+
+def main() -> int:
+    """Print the proof for GITHUB_SHA on GITHUB_REF as JSON, or null; never fail the step."""
+    ref = os.environ.get("GITHUB_REF", "")
+    try:
+        tree = subprocess.run(["git", "rev-parse", "--verify", "--end-of-options", f"{os.environ.get('GITHUB_SHA', '')}^{{tree}}"],
+                              capture_output=True, text=True, check=True).stdout.strip()
+    except (OSError, subprocess.CalledProcessError):
+        tree = ""
+    print(json.dumps(lookup(os.environ.get("GITHUB_REPOSITORY", ""), tree,
+                            ref.removeprefix("refs/heads/") if ref.startswith("refs/heads/") else "", os.environ.get("GITHUB_TOKEN", ""))))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/release/pipeline.py

@@ -12,6 +12,7 @@ from pathlib import Path
 import re
 import shutil
 import subprocess
+import sys
 import tempfile
 import urllib.error
 import zipfile
@@ -19,6 +20,9 @@ import zipfile
 import candidate as c
 import release_gate
 
+sys.path.insert(0, str(c.ROOT / "scripts"))
+import ci_merge_group_proof as proof
+
 
 def current_run(workflow):
     c.require(os.environ.get("GITHUB_REPOSITORY") == c.REPOSITORY and os.environ.get("GITHUB_REF") == "refs/heads/main",
@@ -44,9 +48,26 @@ def output_inputs(inputs):
         c.emit_output(field, inputs[field])
 
 
+CI_GATE = ("Quick Checks", "Test and Lint", "End-to-End Tests (full merge gate)")
+
+
+def run_jobs(run):
+    jobs = []
+    for page in range(1, 11):
+        batch = c.github(f"repos/{c.REPOSITORY}/actions/runs/{run['id']}/attempts/{run['run_attempt']}/jobs?per_page=100&page={page}")["jobs"]
+        jobs.extend(batch)
+        if len(batch) < 100:
+            break
+    return jobs
+
+
+def succeeded(jobs, names):
+    selected = {name: [job for job in jobs if job.get("name") == name] for name in names}
+    return all(len(found) == 1 and found[0].get("status") == "completed" and found[0].get("conclusion") == "success" for found in selected.values())
+
+
 def source_ci(source, version):
     runs = c.github(f"repos/{c.REPOSITORY}/actions/workflows/ci.yml/runs?head_sha={source}&status=success&per_page=10")["workflow_runs"]
-    required = {"Quick Checks", "Test and Lint", "End-to-End Tests (full merge gate)"}
     for run in runs:
         if (run.get("head_sha") != source or run.get("path") != ".github/workflows/ci.yml"
                 or run.get("head_branch") not in ("main", f"release/{version}")
@@ -54,15 +75,18 @@ def source_ci(source, version):
                 or run.get("conclusion") != "success" or run.get("status") != "completed"
                 or (run.get("repository") or {}).get("full_name") != c.REPOSITORY):
             continue
-        jobs = []
-        for page in range(1, 11):
-            batch = c.github(f"repos/{c.REPOSITORY}/actions/runs/{run['id']}/attempts/{run['run_attempt']}/jobs?per_page=100&page={page}")["jobs"]
-            jobs.extend(batch)
-            if len(batch) < 100:
-                break
-        selected = {name: [job for job in jobs if job.get("name") == name] for name in required}
-        if all(len(found) == 1 and found[0].get("status") == "completed" and found[0].get("conclusion") == "success" for found in selected.values()):
+        jobs = run_jobs(run)
+        if succeeded(jobs, CI_GATE):
             return {"run_id": run["id"], "run_attempt": run["run_attempt"]}
+        # The full gate of a queue-landed commit may have run only in the merge-group run of its tree, which
+        # its push run skips as proven. That run is found again here (scripts/ci_merge_group_proof.py) and,
+        # if it passed the whole gate itself, is the evidence.
+        if run["event"] == "push" and succeeded(jobs, CI_GATE[:2]):
+            tree = c.command(["git", "rev-parse", "--verify", "--end-of-options", f"{source}^{{tree}}"], cwd=c.ROOT)
+            queued = c.github(f"repos/{c.REPOSITORY}/actions/workflows/ci.yml/runs?event=merge_group&status=success&per_page=100")
+            for proving in queued["workflow_runs"]:
+                if proof.proves(proving, c.REPOSITORY, run["head_branch"], tree) and succeeded(run_jobs(proving), CI_GATE):
+                    return {"run_id": proving["id"], "run_attempt": proving["run_attempt"]}
     raise ValueError("no complete exact-source CI found; run ci.yml on main or the release branch before creating a candidate")
 
 

scripts/release/test_pipeline.py

@@ -175,6 +175,56 @@ class SourceCITests(unittest.TestCase):
             with self.assertRaises(ValueError):
                 pipeline.source_ci("a" * 40, "1.2.3")
 
+    def test_a_queue_landed_push_is_accepted_only_through_the_queue_run_of_its_tree(self):
+        tree = "d" * 40
+        gate = ("Quick Checks", "Test and Lint", "End-to-End Tests (full merge gate)")
+        push = dict(run_metadata(60, "ci.yml", "a" * 40), event="push")
+
+        def queued(run_id, branch, tree_id=tree):
+            return dict(run_metadata(run_id, "ci.yml", "a" * 40), event="merge_group", run_attempt=3,
+                        head_branch=f"gh-readonly-queue/{branch}/pr-1-" + "c" * 40, head_commit={"tree_id": tree_id},
+                        html_url=f"https://github.com/{c.REPOSITORY}/actions/runs/{run_id}")
+        listing = [queued(62, "main", "e" * 40), queued(63, "release/1.2.3"), queued(61, "main")]
+        # Keyed by (run id, attempt): reading another attempt's jobs is a KeyError, not evidence.
+        jobs = {(60, 2): [{"name": name, "status": "completed", "conclusion": "skipped" if name == gate[2] else "success"} for name in gate]}
+        jobs.update({(run["id"], 3): [{"name": name, "status": "completed", "conclusion": "success"} for name in gate] for run in listing})
+
+        def api(path):
+            if "/jobs?" in path:
+                run_id, attempt = path.split("/runs/")[1].split("/jobs?")[0].split("/attempts/")
+                return {"jobs": jobs[(int(run_id), int(attempt))]}
+            return {"workflow_runs": listing if "event=merge_group" in path else [push]}
+
+        def source_ci():
+            with patch.object(c, "github", side_effect=api), patch.object(c, "command", return_value=tree) as git:
+                found = pipeline.source_ci("a" * 40, "1.2.3")
+            self.assertEqual(git.call_args.args[0], ["git", "rev-parse", "--verify", "--end-of-options", "a" * 40 + "^{tree}"])
+            return found
+        self.assertEqual(source_ci(), {"run_id": 61, "run_attempt": 3})
+        listing.reverse()
+        self.assertEqual(source_ci(), {"run_id": 61, "run_attempt": 3})
+        push["head_branch"] = "release/1.2.3"
+        self.assertEqual(source_ci(), {"run_id": 63, "run_attempt": 3})
+        del listing[listing.index(next(run for run in listing if run["id"] == 63))]
+        with self.assertRaises(ValueError):
+            source_ci()
+        push["head_branch"] = "main"
+        match = next(run for run in listing if run["id"] == 61)
+        refusals = [(match, "head_commit", {"tree_id": "e" * 40}), (match, "head_branch", "gh-readonly-queue/release/1.2.3/pr-1-x"),
+                    (match, "conclusion", "failure"), (match, "event", "push"), (push, "event", "workflow_dispatch")]
+        for record, key, wrong in refusals:
+            old = record[key]; record[key] = wrong
+            with self.subTest(key=key, wrong=wrong), self.assertRaises(ValueError):
+                source_ci()
+            record[key] = old
+        for key, index, conclusion in (((61, 3), 2, "skipped"), ((61, 3), 0, "failure"), ((60, 2), 1, "failure")):
+            old = jobs[key][index]["conclusion"]; jobs[key][index]["conclusion"] = conclusion
+            with self.subTest(run=key, job=gate[index]), self.assertRaises(ValueError):
+                source_ci()
+            jobs[key][index]["conclusion"] = old
+        del jobs[(61, 3)][2]
+        with self.assertRaises(ValueError):
+            source_ci()
 
 class SupplyChainTests(unittest.TestCase):
     def test_actual_generator_uses_frozen_lock_and_all_package_formats(self):

scripts/test_ci_merge_group_proof.py

@@ -0,0 +1,103 @@
+#!/usr/bin/env python3
+"""Fail-closed cases for the merge-group proof of a pushed tree."""
+import io
+import json
+import os
+import subprocess
+import unittest
+import urllib.error
+from unittest.mock import patch
+
+import ci_merge_group_proof as proof
+
+TREE = "b" * 40
+
+
+def run(**changes):
+    value = {"id": 7, "html_url": "https://github.com/rustfs/rustfs/actions/runs/7", "event": "merge_group",
+             "status": "completed", "conclusion": "success", "head_branch": "gh-readonly-queue/main/pr-1-" + "c" * 40,
+             "head_commit": {"tree_id": TREE}}
+    value.update(changes)
+    return value
+
+
+class Response(io.BytesIO):
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *_):
+        self.close()
+
+
+def answering(payload):
+    requests = []
+
+    def opener(request, timeout):
+        requests.append((request, timeout))
+        if isinstance(payload, Exception):
+            raise payload
+        return Response(payload if isinstance(payload, bytes) else json.dumps(payload).encode())
+    return opener, requests
+
+
+class ProofTests(unittest.TestCase):
+    def lookup(self, payload, **overrides):
+        arguments = {"repository": "rustfs/rustfs", "tree": TREE, "branch": "main", "token": "token"}
+        arguments.update(overrides)
+        opener, requests = answering(payload)
+        return proof.lookup(**arguments, opener=opener), requests
+
+    def test_a_successful_merge_group_run_on_the_same_tree_proves_it(self):
+        found, requests = self.lookup({"workflow_runs": [run(id=6, head_commit={"tree_id": "d" * 40}), run()]})
+        self.assertEqual(found, {"run_id": 7, "url": "https://github.com/rustfs/rustfs/actions/runs/7", "tree": TREE})
+        request, timeout = requests[0]
+        self.assertEqual(request.full_url, "https://api.github.com/repos/rustfs/rustfs/actions/workflows/ci.yml/runs"
+                                           "?event=merge_group&status=success&per_page=100")
+        self.assertEqual(request.get_header("Authorization"), "Bearer token")
+        self.assertLessEqual(timeout, 30)
+
+    def test_any_run_that_is_not_a_successful_queue_run_of_this_tree_proves_nothing(self):
+        for changes in ({"head_commit": {"tree_id": "d" * 40}}, {"head_commit": None}, {"head_commit": {}},
+                        {"conclusion": "failure"}, {"conclusion": "cancelled"}, {"conclusion": None},
+                        {"status": "in_progress"}, {"event": "push"}, {"event": "pull_request"},
+                        {"head_branch": "main"}, {"head_branch": "gh-readonly-queue/release/pr-1-x"},
+                        {"head_branch": "gh-readonly-queue/mainline/pr-1-x"},
+                        {"head_branch": None}, {"id": "7"}, {"id": 0}, {"id": True},
+                        {"html_url": "https://example.invalid/runs/7"}, {"html_url": None}):
+            with self.subTest(changes=changes):
+                self.assertIsNone(self.lookup({"workflow_runs": [run(**changes)]})[0])
+
+    def test_every_doubt_about_the_answer_runs_everything(self):
+        rate_limited = urllib.error.HTTPError("https://api.github.com", 403, "rate limit exceeded", {}, None)
+        for payload in (rate_limited, urllib.error.HTTPError("u", 500, "error", {}, None), urllib.error.URLError("dns"),
+                        TimeoutError(), b"not json", b"[]", {"workflow_runs": None}, {"workflow_runs": [None, "x"]}, {}, {"workflow_runs": []}):
+            with self.subTest(payload=payload):
+                self.assertIsNone(self.lookup(payload)[0])
+
+    def test_no_request_without_a_token_or_with_a_malformed_identity(self):
+        for overrides in ({"token": ""}, {"tree": ""}, {"tree": "B" * 40}, {"tree": "b" * 39}, {"branch": ""},
+                          {"branch": "main/../x"}, {"repository": "rustfs"}, {"repository": "a/b/c"}):
+            with self.subTest(overrides=overrides):
+                found, requests = self.lookup({"workflow_runs": [run()]}, **overrides)
+                self.assertIsNone(found)
+                self.assertEqual(requests, [])
+
+    def test_the_step_prints_the_proof_of_the_pushed_commit_or_null(self):
+        head = subprocess.run(["git", "rev-parse", "HEAD"], capture_output=True, text=True, check=True).stdout.strip()
+        tree = subprocess.run(["git", "rev-parse", "HEAD^{tree}"], capture_output=True, text=True, check=True).stdout.strip()
+        environment = {"GITHUB_REPOSITORY": "rustfs/rustfs", "GITHUB_REF": "refs/heads/main", "GITHUB_SHA": head, "GITHUB_TOKEN": "token"}
+        cases = [({}, {"workflow_runs": [run(head_commit={"tree_id": tree})]}, {"run_id": 7, "url": run()["html_url"], "tree": tree}),
+                 ({}, urllib.error.URLError("dns"), None),
+                 ({"GITHUB_SHA": "0" * 40}, {"workflow_runs": [run(head_commit={"tree_id": tree})]}, None),
+                 ({"GITHUB_SHA": "--output=x"}, {"workflow_runs": [run(head_commit={"tree_id": tree})]}, None),
+                 ({"GITHUB_REF": "refs/tags/main"}, {"workflow_runs": [run(head_commit={"tree_id": tree})]}, None)]
+        for overrides, payload, expected in cases:
+            opener, _ = answering(payload)
+            with self.subTest(overrides=overrides, payload=payload), patch.dict(os.environ, dict(environment, **overrides), clear=True), \
+                    patch.object(proof.urllib.request, "urlopen", opener), patch("sys.stdout", new_callable=io.StringIO) as output:
+                self.assertEqual(proof.main(), 0)
+                self.assertEqual(json.loads(output.getvalue()), expected)
+
+
+if __name__ == "__main__":
+    unittest.main()

scripts/test_ci_plan.py

@@ -44,8 +44,8 @@ class Repository:
         self.git('commit', '-qm', 'fixture')
         return self.git('rev-parse', 'HEAD')
 
-    def plan(self, head=None, event='pull_request', base=None, action='opened'):
-        return gate.select_plan(event, base or self.base, head or self.git('rev-parse', 'HEAD'), 'refs/heads/main', self.root, action)
+    def plan(self, head=None, event='pull_request', base=None, action='opened', proof=None):
+        return gate.select_plan(event, base or self.base, head or self.git('rev-parse', 'HEAD'), 'refs/heads/main', self.root, action, proof=proof)
 
 
 class PlannerTests(unittest.TestCase):
@@ -182,6 +182,24 @@ class PlannerTests(unittest.TestCase):
         for event in ('push', 'merge_group', 'schedule', 'workflow_dispatch'):
             self.assertTrue(self.repo.plan(head, event=event, base=baseline)['full_validation'])
 
+    def test_push_skips_validation_only_for_the_tree_a_merge_group_run_proved(self):
+        head, tree = self.repo.git('rev-parse', 'HEAD'), self.repo.git('rev-parse', 'HEAD^{tree}')
+        proof = {'run_id': 7, 'url': 'https://github.com/rustfs/rustfs/actions/runs/7', 'tree': tree}
+        plan = self.repo.plan(head, event='push', proof=proof)
+        self.assertEqual((plan['mode'], plan['proven_by']), ('proven', proof))
+        self.assertEqual({job for job, state in plan['jobs'].items() if state == 'success'}, set(gate.ALWAYS_JOBS))
+        self.assertTrue(all(proof['url'] in plan['not_applicable'][job] for job in gate.CODE_JOBS))
+        self.assertEqual(gate.validate_plan(plan, 'push', 'refs/heads/main', head, tree), plan['jobs'])
+        # A direct push has no proving run; a proof of another tree, or on any other event, proves nothing.
+        for event, candidate in [('push', None), ('push', dict(proof, tree='0' * 40)), ('merge_group', proof),
+                                 ('schedule', proof), ('workflow_dispatch', proof), ('pull_request', proof)]:
+            with self.subTest(event=event, proof=candidate):
+                other = self.repo.plan(head, event=event, proof=candidate)
+                self.assertNotEqual(other['mode'], 'proven')
+                self.assertIsNone(other['proven_by'])
+                if event != 'pull_request':
+                    self.assertTrue(all(other['jobs'][job] == 'success' for job in (gate.LINT_JOB, gate.ARCHIVE_PRODUCER, 'workspace-tests')))
+
     def test_individual_stable_version_fields_without_new_rpm_history_are_metadata(self):
         baseline, files = self.release_metadata_base()
         for path, content in files.items():
@@ -464,7 +482,8 @@ class GateTests(unittest.TestCase):
                 'base': 'c' * 40, 'action': '', 'mode': mode, 'connect': connect, 'full_validation': mode == 'full', 'packages': ['rustfs'] if mode != 'docs' else [],
                 'rehearse_shards': False,
                 'policy': {'revision': 'c' * 40, 'rollout': 'enforced'}, 'jobs': jobs, 'configurations': gate.configurations(jobs),
-                'reasons': ['fixture'], 'not_applicable': {job: 'fixture' for job, state in jobs.items() if state == 'skipped'}}
+                'reasons': ['fixture'], 'not_applicable': {job: 'fixture' for job, state in jobs.items() if state == 'skipped'},
+                'proven_by': {'run_id': 7, 'url': 'https://github.com/rustfs/rustfs/actions/runs/7', 'tree': 'b' * 40} if mode == 'proven' else None}
 
     def needs(self, plan):
         needs = {job: {'result': state} for job, state in plan['jobs'].items()}
@@ -472,7 +491,7 @@ class GateTests(unittest.TestCase):
         return needs
 
     def test_failure_cancel_missing_extra_or_unexpected_skip_is_rejected(self):
-        for event, mode in [('pull_request', 'docs'), ('pull_request', 'e2e'), ('pull_request', 'full'), ('push', 'full'), ('merge_group', 'full'), ('schedule', 'full'), ('workflow_dispatch', 'full')]:
+        for event, mode in [('pull_request', 'docs'), ('pull_request', 'e2e'), ('pull_request', 'full'), ('push', 'full'), ('push', 'proven'), ('merge_group', 'full'), ('schedule', 'full'), ('workflow_dispatch', 'full')]:
             plan = self.plan(event, mode)
             good = self.needs(plan)
             self.assertEqual(gate.verify_results(good, event, plan['ref'], plan['head'], plan['tree']), [])
@@ -502,6 +521,28 @@ class GateTests(unittest.TestCase):
         self.assertTrue(gate.verify_results(good, plan['event'], plan['ref'], base='d' * 40))
         self.assertTrue(gate.verify_results(good, plan['event'], plan['ref'], action='closed'))
 
+    def test_a_proven_push_needs_its_own_tree_proof_and_nothing_else_may_carry_one(self):
+        proven, full = self.plan('push', 'proven'), self.plan('push', 'full')
+        self.assertEqual(gate.verify_results(self.needs(proven), 'push', proven['ref'], proven['head'], proven['tree']), [])
+        proof = proven['proven_by']
+        bad = [dict(proven, proven_by=None), dict(proven, proven_by=dict(proof, tree='c' * 40)),
+               dict(proven, proven_by=dict(proof, run_id='7')), dict(proven, proven_by=dict(proof, run_id=0)),
+               dict(proven, proven_by=dict(proof, url='https://example.invalid/7')), dict(proven, proven_by={}),
+               dict(full, proven_by=proof)]
+        for plan in bad:
+            with self.subTest(plan=plan['proven_by']), self.assertRaises(ValueError):
+                gate.validate_plan(plan, 'push', plan['ref'])
+        for event in ('pull_request', 'merge_group', 'schedule', 'workflow_dispatch'):
+            with self.subTest(event=event), self.assertRaises(ValueError):
+                gate.expected_results('proven', event, 'refs/heads/main')
+
+    def test_the_queue_run_that_proves_a_push_validates_everything_that_push_would(self):
+        queued = {job for job, result in gate.expected_results('full', 'merge_group', 'refs/heads/main').items() if result == 'success'}
+        for ref in gate.INTEGRATION_PUSH_REFS:
+            pushed = {job for job, result in gate.expected_results('full', 'push', ref).items() if result == 'success'}
+            with self.subTest(ref=ref):
+                self.assertLessEqual(pushed, queued)
+
     def test_legacy_base_cannot_authorize_skipped_new_jobs(self):
         expected = gate.expected_results('full', 'pull_request', 'refs/heads/main')
         needs = {job: {'result': result} for job, result in expected.items()}
@@ -544,11 +585,11 @@ class GateTests(unittest.TestCase):
             elif job in gate.OPTIONAL_JOBS[:2]:
                 expression = "contains(fromJSON('[\"schedule\",\"workflow_dispatch\"]'), github.event_name)"
             elif job == 'e2e-full':
-                expression = "github.event_name == 'merge_group' || github.event_name == 'workflow_dispatch' || (github.event_name == 'push' && github.ref == 'refs/heads/main')"
+                expression = "needs.classify-changes.outputs.mode == 'full' && (github.event_name == 'merge_group' || github.event_name == 'workflow_dispatch' || (github.event_name == 'push' && github.ref == 'refs/heads/main'))"
             elif job in gate.SHARD_JOBS:
                 expression = "needs.classify-changes.outputs.rehearse_shards == 'true'"
             elif job in gate.E2E_JOBS:
-                expression = "needs.classify-changes.outputs.mode != 'docs'"
+                expression = "contains(fromJSON('[\"full\",\"e2e\"]'), needs.classify-changes.outputs.mode)"
             else:
                 expression = "needs.classify-changes.outputs.mode == 'full'"
             expression = f"({expression}) && (github.event_name != 'pull_request' || github.event.action != 'closed')"