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.

.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.

crates/data-usage/src/data_usage.rs

@@ -1764,7 +1764,12 @@ impl DataUsageCache {
         let mut found = HashSet::new();
         found.insert(top);
         mark(self, &top_e, &mut found);
-        self.cache.retain(|k, _| found.contains(k));
+        self.cache.retain(|k, _| {
+            if !found.contains(k) {
+                return false;
+            }
+            true
+        });
     }
 
     pub fn reduce_children_of(&mut self, path: &DataUsageHash, limit: usize, compact_self: bool) {

crates/e2e_test/src/distributed/multipart_rpc_delay_test.rs

@@ -766,9 +766,9 @@ impl Run {
             || async {
                 let mut ready = true;
                 let mut observations = Vec::with_capacity(4);
-                for (node, cluster_node) in self.dist.cluster.nodes.iter().enumerate().take(4) {
+                for node in 0..4 {
                     match http
-                        .get(format!("{}/health/ready", cluster_node.url))
+                        .get(format!("{}/health/ready", self.dist.cluster.nodes[node].url))
                         .timeout(Duration::from_secs(3))
                         .send()
                         .await

crates/e2e_test/src/distributed_startup_regression_test.rs

@@ -1332,15 +1332,14 @@ mod tests {
                 .collect();
             assert_eq!(confirmed.len(), 1);
             let mut receivers = Vec::new();
-            let receiver_endpoints = phase.disks.get(1).map_or(&[][..], |disks| &phase.endpoints[1][..disks.len()]);
-            for endpoint in receiver_endpoints {
+            for drive in 0..phase.disks.get(1).map_or(0, Vec::len) {
                 for cas in [prepares[1], commits[1]] {
                     let matching: Vec<_> = events[1]
                         .iter()
                         .copied()
                         .filter(|event| {
                             event["kind"] == "receiver"
-                                && event["disk"] == *endpoint
+                                && event["disk"] == phase.endpoints[1][drive]
                                 && event["dst_volume"] == ".rustfs.sys"
                                 && event["dst_path"] == "pool.bin"
                                 && event["etag"] == cas["etag"]

crates/ecstore/src/bucket/migration.rs

@@ -580,7 +580,7 @@ mod tests {
             let first: Error = make_error("buckets/first/.metadata.bin".into()).into();
             let second: Error = make_error("buckets/second/.metadata.bin".into()).into();
             assert_eq!(first, second, "record paths must not fragment error grouping");
-            assert_eq!(first, second, "cloning must preserve error grouping");
+            assert_eq!(first.clone(), second, "cloning must preserve error grouping");
 
             let io_error = std::io::Error::from(first);
             let detail = io_error

crates/ecstore/src/core/pools.rs

@@ -20945,12 +20945,12 @@ impl ECStore {
         let locator = parse_decommission_durable_ilm_receipt_path(&receipt_path)?;
         let mut proof = None::<DecommissionDurableIlmReceipt>;
         let mut nonterminal_receipt_found = false;
-        for (pool_idx, pool) in self.pools.iter().enumerate() {
+        for pool_idx in 0..self.pools.len() {
             if pool_idx == source_pool_idx {
                 continue;
             }
             let data = match read_config_limited_preserve_empty(
-                pool.clone(),
+                self.pools[pool_idx].clone(),
                 &receipt_path,
                 DECOMMISSION_DURABLE_ILM_RECEIPT_MAX_SIZE,
             )
@@ -28413,7 +28413,7 @@ mod tests {
             for wrapped in [
                 err.clone(),
                 data_movement::data_movement_stage_error_for_test("decommission_object", "put_object", "b", "o", err.clone()),
-                with_decommission_entry_context("migrate_object", "b", "o", err),
+                with_decommission_entry_context("migrate_object", "b", "o", err.clone()),
             ] {
                 assert_eq!(
                     classify_decommission_capacity_failure(&wrapped),
@@ -40281,7 +40281,7 @@ mod pools_tests {
             )
             .expect("finalize records the completion");
 
-            let mut adopted = meta;
+            let mut adopted = meta.clone();
             let plan = decommission_capacity_intent_adoption_plan(&adopted, 0, generation)
                 .expect("plan adoption")
                 .expect("the interrupted completion should be planned");

crates/ecstore/src/data_movement/mod.rs

@@ -4242,7 +4242,7 @@ mod tests {
                 "compatible aliases for {suffix} should match"
             );
 
-            let mut source_without_marker = source;
+            let mut source_without_marker = source.clone();
             rustfs_utils::http::remove_str(Arc::make_mut(&mut source_without_marker.user_defined), suffix);
             let mut target_with_generated_marker = source_without_marker.clone();
             rustfs_utils::http::insert_str(

crates/ecstore/src/data_usage/mod.rs

@@ -1512,7 +1512,7 @@ async fn normalize_loaded_data_usage(
             && !cfg.role.is_empty()
         {
             data_usage_info.replication_info.insert(
-                cfg.role,
+                cfg.role.clone(),
                 BucketTargetUsageInfo {
                     replication_failed_size: bui.replication_failed_size_v1,
                     replication_failed_count: bui.replication_failed_count_v1,
@@ -1725,7 +1725,7 @@ fn merge_snapshot(aggregated: &mut DataUsageInfo, mut snapshot: LocalUsageSnapsh
         match aggregated.buckets_usage.entry(bucket.clone()) {
             Entry::Occupied(mut entry) => entry.get_mut().merge(&usage),
             Entry::Vacant(entry) => {
-                entry.insert(usage);
+                entry.insert(usage.clone());
             }
         }
 

crates/ecstore/src/diagnostics/admin_server_info.rs

@@ -521,7 +521,7 @@ impl EndpointTopology {
                     }
                 };
 
-                topology.aliases.entry(host_port).or_insert(member_index);
+                topology.aliases.entry(host_port.clone()).or_insert(member_index);
                 topology.aliases.entry(ep.to_string()).or_insert(member_index);
                 if let Some(display_endpoint) = ep.url.host_str() {
                     match display_aliases.entry(display_endpoint.to_owned()) {

crates/ecstore/src/disk/error.rs

@@ -1626,7 +1626,7 @@ mod tests {
 
         for disk_error in disk_errors {
             let original_message = disk_error.to_string();
-            let io_error: std::io::Error = disk_error.into();
+            let io_error: std::io::Error = disk_error.clone().into();
 
             // The io::Error should contain the original error message
             assert!(io_error.to_string().contains(&original_message));

crates/ecstore/src/disk/local.rs

@@ -3603,7 +3603,11 @@ pub(crate) fn has_delete_version_test_failpoints() -> bool {
     if os::prepared_publication_test_hooks::has_any_hooks() {
         return true;
     }
-    cfg!(feature = "e2e-test-hooks") && std::env::var_os("RUSTFS_DELETE_RECOVERY_BARRIER_DIR").is_some()
+    #[cfg(feature = "e2e-test-hooks")]
+    if std::env::var_os("RUSTFS_DELETE_RECOVERY_BARRIER_DIR").is_some() {
+        return true;
+    }
+    false
 }
 
 #[cfg(test)]

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

@@ -295,11 +295,17 @@ pub(super) fn put_active_multipart_source_head(source: impl AsRef<Path>, raw: By
 }
 
 fn stamp_matches_active_disk(source: &Path, expected_head_stamp: FileStamp) -> bool {
-    FileStamp::from_path(&source.join(STORAGE_FORMAT_FILE)) == Some(expected_head_stamp)
-        && !(cfg!(test)
-            && (source.join(STATE).exists()
-                || source.join(crate::disk::MULTIPART_RECOVERY_DIR).exists()
-                || source.join(".mpu-completions").exists()))
+    if FileStamp::from_path(&source.join(STORAGE_FORMAT_FILE)) != Some(expected_head_stamp) {
+        return false;
+    }
+    #[cfg(test)]
+    if source.join(STATE).exists()
+        || source.join(crate::disk::MULTIPART_RECOVERY_DIR).exists()
+        || source.join(".mpu-completions").exists()
+    {
+        return false;
+    }
+    true
 }
 
 pub(super) fn get_active_multipart_source_head(source: &Path) -> Option<(Bytes, FileInfo)> {

crates/ecstore/src/disk/multipart_source.rs

@@ -2137,7 +2137,7 @@ mod multipart_proof_authority_tests {
             assert_eq!(expected, snapshot);
             assert_eq!(actual_round, round);
             let publication = PublicationIdentity {
-                lineage: round.lineage,
+                lineage: round.lineage.clone(),
                 source_generation: snapshot.generation,
                 attempt: Uuid::new_v4(),
                 version_id: None,
@@ -2145,7 +2145,7 @@ mod multipart_proof_authority_tests {
                 modified_unix_nanos: nanos,
                 completion_identity: [6; 32],
             };
-            snapshot.publication = Some(publication);
+            snapshot.publication = Some(publication.clone());
             let response = MultipartSourceResponse::TerminalConfirmed(
                 MultipartSourceProof::issue(
                     &authority,

crates/ecstore/src/disk/os.rs

@@ -392,11 +392,14 @@ pub(crate) mod prepared_publication_test_hooks {
     }
 
     pub(crate) fn has_any_hooks() -> bool {
+        if !BEFORE_PUBLICATION.lock().is_empty() {
+            return true;
+        }
         #[cfg(test)]
         if !RENAME_DESTINATIONS.lock().is_empty() || !HARDLINK_DESTINATIONS.lock().is_empty() {
             return true;
         }
-        !BEFORE_PUBLICATION.lock().is_empty()
+        false
     }
 
     #[cfg(test)]

crates/ecstore/src/erasure/codec/bridge.rs

@@ -439,7 +439,7 @@ mod tests {
             reconstruct_with(&engine, &mut shards).expect("restore missing data and parity");
             assert_eq!(shards, expected, "every restored byte must match at width {}", shards.len());
 
-            let mut corrupt = expected;
+            let mut corrupt = expected.clone();
             corrupt[0] = None;
             corrupt[data_shards] = None;
             corrupt[data_shards + 1].as_mut().expect("surplus parity present")[0] ^= 1;

crates/ecstore/src/error/mod.rs

@@ -1996,7 +1996,7 @@ mod tests {
 
         for storage_error in storage_errors {
             let original_message = storage_error.to_string();
-            let io_error: std::io::Error = storage_error.into();
+            let io_error: std::io::Error = storage_error.clone().into();
 
             // The io::Error should contain the original error message or info
             assert!(io_error.to_string().contains(&original_message));

crates/ecstore/src/layout/endpoints.rs

@@ -391,7 +391,7 @@ impl PoolEndpointList {
                             e.get_mut().extend(host_ip_set.iter());
                         }
                         Entry::Vacant(e) => {
-                            e.insert(host_ip_set);
+                            e.insert(host_ip_set.clone());
                         }
                     }
                 }

crates/ecstore/src/layout/set_heal.rs

@@ -47,7 +47,7 @@ pub(crate) fn formats_to_drives_info(
         before_drives.push(HealDriveInfo {
             uuid,
             endpoint: drive,
-            state,
+            state: state.to_string(),
         });
     }
     before_drives

crates/ecstore/src/runtime/sources.rs

@@ -526,7 +526,7 @@ pub(crate) async fn record_local_disks(instance_ctx: &Arc<InstanceContext>, disk
     let mut global_local_disk_map = map.write().await;
     for disk in disks {
         let path = disk.endpoint().to_string();
-        global_local_disk_map.insert(path, Some(disk));
+        global_local_disk_map.insert(path, Some(disk.clone()));
     }
 }
 
@@ -625,7 +625,7 @@ pub(crate) async fn initialize_local_disk_maps(
             })?;
 
             global_local_disk_map.insert(path, Some(disk.clone()));
-            global_set_drives[pool_idx][set_idx][disk_idx] = Some(disk);
+            global_set_drives[pool_idx][set_idx][disk_idx] = Some(disk.clone());
         }
     }
 

crates/ecstore/src/services/notification_sys.rs

@@ -3168,7 +3168,7 @@ impl NotificationSys {
                     return Err(Error::other("scanner scoped dirty usage capability requires scoped acknowledgements"));
                 }
             };
-            if by_host.insert(host, acknowledgement).is_some() {
+            if by_host.insert(host.clone(), acknowledgement).is_some() {
                 return Err(Error::other("duplicate scanner dirty usage acknowledgement target"));
             }
         }
@@ -3218,7 +3218,7 @@ impl NotificationSys {
                 ScannerDirtyUsageAcknowledgement::Generation { host, .. }
                 | ScannerDirtyUsageAcknowledgement::Scoped { host, .. } => host.clone(),
             };
-            if by_host.insert(host, acknowledgement).is_some() {
+            if by_host.insert(host.clone(), acknowledgement).is_some() {
                 return Err(Error::other("duplicate scanner dirty usage acknowledgement target"));
             }
         }

crates/ecstore/src/services/rebalance/entry.rs

@@ -802,6 +802,7 @@ impl ECStore {
             let set = set.clone();
             let rx = rx.clone();
             let bucket = bucket.clone();
+            let entry_tasks = entry_tasks.clone();
 
             let job = tokio::spawn(async move {
                 let list_rx = rx.clone();

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

@@ -8144,9 +8144,9 @@ mod tests {
                     assert_eq!(calls.total(disk_call_counters::KIND_DISK_SHARD_OPEN), 4);
                     assert_eq!(setup.available_shards(), 4);
                     assert_eq!(setup.deferred_shards(), 4);
-                    for (index, attempted) in setup.attempted.iter().enumerate().take(8).skip(4) {
+                    for index in 4..8 {
                         assert_eq!(calls.for_disk(disk_call_counters::KIND_DISK_SHARD_OPEN, index), 0);
-                        assert!(!attempted, "parity must stay unopened until needed");
+                        assert!(!setup.attempted[index], "parity must stay unopened until needed");
                     }
                     let (output, error) = local_batch_quorum_decode(setup, files[0].erasure.block_size).await;
                     assert!(error.is_none(), "healthy decode failed: {error:?}");

crates/ecstore/src/set_disk/mod.rs

@@ -6045,10 +6045,14 @@ fn should_use_inline_fast_path_with_plan(
     // a stale/corrupt marker cannot route an oversized payload into the
     // in-memory decoder. The independent direct-memory reader remains capped
     // at 128 KiB below.
-    plan.is_plain()
-        && shape.object_size <= usize::try_from(INLINE_FAST_PATH_MAX_OBJECT_SIZE).expect("inline fast path limit fits usize")
-        && fi.erasure.data_blocks != 0
-        && fi.erasure.block_size != 0
+    if !plan.is_plain()
+        || shape.object_size > usize::try_from(INLINE_FAST_PATH_MAX_OBJECT_SIZE).expect("inline fast path limit fits usize")
+        || fi.erasure.data_blocks == 0
+        || fi.erasure.block_size == 0
+    {
+        return false;
+    }
+    true
 }
 
 enum SmallWritePath {

crates/ecstore/src/set_disk/ops/heal.rs

@@ -1285,7 +1285,7 @@ impl SetDisks {
                             result.after.drives.push(HealDriveInfo {
                                 uuid: "".to_string(),
                                 endpoint: self.set_endpoints[index].to_string(),
-                                state: drive_state,
+                                state: drive_state.to_string(),
                             });
                         }
 
@@ -3102,7 +3102,7 @@ impl SetDisks {
             result.after.drives.push(HealDriveInfo {
                 uuid: "".to_string(),
                 endpoint,
-                state: drive_state,
+                state: drive_state.to_string(),
             });
         }
 
@@ -3147,7 +3147,7 @@ impl SetDisks {
                         _ => DriveState::Corrupt.to_string(),
                     },
                 };
-                result.after.drives[index].state = drive_state;
+                result.after.drives[index].state = drive_state.to_string();
             }
         }
 
@@ -10994,7 +10994,7 @@ mod heal_result_report_tests {
                     set.heal_object(bucket, object, "", &heal_opts),
                 );
                 let put_info = put_res.expect("overwrite must succeed under lock serialization");
-                last_etag = put_info.etag.unwrap_or_default();
+                last_etag = put_info.etag.clone().unwrap_or_default();
                 // Heal outcome is unconstrained (may hit the tail race and report a
                 // retryable error); the invariant is checked on the end state.
                 drop(heal_res);

crates/ecstore/src/set_disk/ops/locking.rs

@@ -690,10 +690,9 @@ impl SetDisks {
             return Err(Error::other("DriveID: offline"));
         }
 
-        let set_drive_count = self.format.erasure.sets[0].len();
-        for (i, set) in self.format.erasure.sets.iter().enumerate() {
-            for (j, drive) in set.iter().enumerate().take(set_drive_count) {
-                if fm.erasure.this == *drive {
+        for i in 0..self.format.erasure.sets.len() {
+            for j in 0..self.format.erasure.sets[0].len() {
+                if fm.erasure.this == self.format.erasure.sets[i][j] {
                     return Ok((i, j));
                 }
             }

crates/ecstore/src/set_disk/ops/multipart.rs

@@ -11560,7 +11560,7 @@ mod tests {
                 break;
             }
             key_marker = page.next_key_marker.clone();
-            upload_id_marker = page.next_upload_id_marker;
+            upload_id_marker = page.next_upload_id_marker.clone();
             assert!(key_marker.is_some(), "a truncated page must carry a next key marker");
             assert!(upload_id_marker.is_some(), "a truncated upload page must carry a next upload marker");
         }

crates/ecstore/src/store/init.rs

@@ -17932,9 +17932,10 @@ mod tests {
             .expect("shared-tuple recursive prefix delete should commit");
 
         assert_eq!(tier_delete_journal_count(store.clone()).await, 1);
-        for (pool_idx, pool) in store.pools.iter().enumerate().take(2) {
+        for pool_idx in 0..2 {
             assert!(
-                pool.get_disks_by_key(object)
+                store.pools[pool_idx]
+                    .get_disks_by_key(object)
                     .load_file_info_versions_exact(bucket, object)
                     .await
                     .expect("shared-tuple post-delete metadata should decode")
@@ -19767,8 +19768,8 @@ mod tests {
         }
         wait_for_expiry_workers_idle(&store).await;
         let mut free_version_ids = Vec::new();
-        for (pool_idx, pool) in store.pools.iter().enumerate().take(2) {
-            let retained = pool
+        for pool_idx in 0..2 {
+            let retained = store.pools[pool_idx]
                 .get_disks_by_key(object)
                 .load_file_info_versions_exact(&bucket, object)
                 .await
@@ -19804,9 +19805,10 @@ mod tests {
         );
         wait_for_expiry_workers_idle(&store).await;
         assert_eq!(backend.remove_count().await, 1, "shared remote tuple should be deleted exactly once");
-        for (pool_idx, pool) in store.pools.iter().enumerate().take(2) {
+        for pool_idx in 0..2 {
             assert!(
-                pool.get_disks_by_key(object)
+                store.pools[pool_idx]
+                    .get_disks_by_key(object)
                     .load_file_info_versions_exact(&bucket, object)
                     .await
                     .expect("post-delete metadata should decode")
@@ -19856,9 +19858,9 @@ mod tests {
             .expect("multi-pool source bucket should be created");
 
         let mut remote_tuples = Vec::new();
-        for (pool_idx, pool) in store.pools.iter().enumerate().take(2) {
+        for pool_idx in 0..2 {
             let mut reader = PutObjReader::from_vec(vec![b'a' + pool_idx as u8; 1024 * 1024]);
-            let source = pool
+            let source = store.pools[pool_idx]
                 .put_object(
                     bucket,
                     object,
@@ -19870,27 +19872,31 @@ mod tests {
                 )
                 .await
                 .expect("pool-local source should be written");
-            pool.transition_object(
-                bucket,
-                object,
-                &ObjectOptions {
-                    transition: TransitionOptions {
-                        status: TRANSITION_PENDING.to_string(),
-                        tier: tier_name.to_string(),
-                        etag: source.etag.clone().expect("pool-local source should have an etag"),
+            store.pools[pool_idx]
+                .transition_object(
+                    bucket,
+                    object,
+                    &ObjectOptions {
+                        transition: TransitionOptions {
+                            status: TRANSITION_PENDING.to_string(),
+                            tier: tier_name.to_string(),
+                            etag: source.etag.clone().expect("pool-local source should have an etag"),
+                            ..Default::default()
+                        },
+                        mod_time: source.mod_time,
                         ..Default::default()
                     },
-                    mod_time: source.mod_time,
-                    ..Default::default()
-                },
-            )
-            .await
-            .expect("pool-local source should transition");
-            let transitioned = pool
+                )
+                .await
+                .expect("pool-local source should transition");
+            let transitioned = store.pools[pool_idx]
                 .get_object_info(bucket, object, &ObjectOptions::default())
                 .await
                 .expect("pool-local transitioned source should be readable");
-            remote_tuples.push((transitioned.transitioned_object.name.clone(), transitioned.transitioned_object.version_id));
+            remote_tuples.push((
+                transitioned.transitioned_object.name.clone(),
+                transitioned.transitioned_object.version_id.clone(),
+            ));
         }
         assert_ne!(remote_tuples[0], remote_tuples[1], "fixture must use distinct remote cleanup tuples");
         assert_eq!(backend.object_count().await, 2);
@@ -19900,8 +19906,8 @@ mod tests {
             .delete_object_with_tier_delete_journal(bucket, object, ObjectOptions::default())
             .await
             .expect("store-level delete should remove both pool copies");
-        for (pool_idx, pool) in store.pools.iter().enumerate().take(2) {
-            let retained = pool
+        for pool_idx in 0..2 {
+            let retained = store.pools[pool_idx]
                 .get_disks_by_key(object)
                 .load_file_info_versions_exact(bucket, object)
                 .await
@@ -19940,9 +19946,10 @@ mod tests {
         removed.sort();
         remote_tuples.sort();
         assert_eq!(removed, remote_tuples, "recovery must use each pool's exact remote tuple");
-        for (pool_idx, pool) in store.pools.iter().enumerate().take(2) {
+        for pool_idx in 0..2 {
             assert!(
-                pool.get_disks_by_key(object)
+                store.pools[pool_idx]
+                    .get_disks_by_key(object)
                     .load_file_info_versions_exact(bucket, object)
                     .await
                     .expect("post-recovery pool metadata should decode")

crates/ecstore/src/store/list_objects.rs

@@ -8504,7 +8504,7 @@ mod test {
 
         let mut results = Vec::new();
         while let Some(entry) = out_rx.recv().await {
-            results.push(entry.name);
+            results.push(entry.name.clone());
         }
 
         handle.await.unwrap().unwrap();
@@ -8533,7 +8533,7 @@ mod test {
 
         let mut results = Vec::new();
         while let Some(entry) = out_rx.recv().await {
-            results.push(entry.name);
+            results.push(entry.name.clone());
         }
 
         handle.await.unwrap().unwrap();
@@ -9177,7 +9177,7 @@ mod test {
 
         let mut results = Vec::new();
         while let Some(entry) = out_rx.recv().await {
-            results.push(entry.name);
+            results.push(entry.name.clone());
         }
 
         handle.await.unwrap().unwrap();
@@ -9265,7 +9265,7 @@ mod test {
 
         let mut results = Vec::new();
         while let Some(entry) = out_rx.recv().await {
-            results.push(entry.name);
+            results.push(entry.name.clone());
         }
         handle.await.unwrap().unwrap();
 
@@ -9295,7 +9295,7 @@ mod test {
 
         let mut results = Vec::new();
         while let Some(entry) = out_rx.recv().await {
-            results.push(entry.name);
+            results.push(entry.name.clone());
         }
         handle.await.unwrap().unwrap();
 
@@ -9367,7 +9367,7 @@ mod test {
 
         let mut results = Vec::new();
         while let Some(entry) = out_rx.recv().await {
-            results.push(entry.name);
+            results.push(entry.name.clone());
         }
         handle.await.unwrap().unwrap();
 
@@ -9401,7 +9401,7 @@ mod test {
 
         let mut results = Vec::new();
         while let Some(entry) = out_rx.recv().await {
-            results.push(entry.name);
+            results.push(entry.name.clone());
         }
         handle.await.unwrap().unwrap();
 
@@ -9491,7 +9491,7 @@ mod test {
         let mut results = Vec::new();
         while let Some(entry) = out_rx.recv().await {
             assert!(entry.is_object(), "the directory-marker object must shadow the same-named prefix dir");
-            results.push(entry.name);
+            results.push(entry.name.clone());
         }
         handle.await.unwrap().unwrap();
 
@@ -9514,7 +9514,7 @@ mod test {
 
         let mut results = Vec::new();
         while let Some(entry) = out_rx.recv().await {
-            results.push(entry.name);
+            results.push(entry.name.clone());
         }
         handle.await.unwrap().unwrap();
 
@@ -9538,7 +9538,7 @@ mod test {
 
         let mut results = Vec::new();
         while let Some(entry) = out_rx.recv().await {
-            results.push(entry.name);
+            results.push(entry.name.clone());
         }
         handle.await.unwrap().unwrap();
 

crates/ecstore/src/store/rebalance.rs

@@ -1722,19 +1722,20 @@ mod tests {
             .await
             .expect("bucket should be created in both pools");
 
-        for (pool_index, pool) in store.pools.iter().enumerate().take(2) {
+        for pool_index in 0..2 {
             let mut reader = PutObjReader::from_vec(format!("pool-{pool_index}-history").into_bytes());
-            pool.put_object(
-                &bucket,
-                object,
-                &mut reader,
-                &ObjectOptions {
-                    versioned: true,
-                    ..Default::default()
-                },
-            )
-            .await
-            .expect("historical version should be stored");
+            store.pools[pool_index]
+                .put_object(
+                    &bucket,
+                    object,
+                    &mut reader,
+                    &ObjectOptions {
+                        versioned: true,
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("historical version should be stored");
         }
         let marker = store.pools[0]
             .delete_object(

crates/filemeta/src/filemeta/inline_data.rs

@@ -908,7 +908,7 @@ mod tests {
                     assert_indexed_matches_scalar(&original, selected);
                 }
                 if inline_mode > 0 {
-                    let mut legacy_null = original;
+                    let mut legacy_null = original.clone();
                     let payload = legacy_null
                         .data
                         .find(NULL_VERSION_ID)

crates/filemeta/src/filemeta/validation.rs

@@ -113,7 +113,11 @@ impl FileMetaVersionHeader {
         }
 
         // Check erasure coding parameters
-        !(self.has_ec() && (self.ec_n == 0 || self.ec_m == 0 || self.ec_m < self.ec_n))
+        if self.has_ec() && (self.ec_n == 0 || self.ec_m == 0 || self.ec_m < self.ec_n) {
+            return false;
+        }
+
+        true
     }
 }
 

crates/filemeta/src/metacache.rs

@@ -4223,10 +4223,9 @@ mod tests {
                         o_hdr.ec_m = o_m;
                         o_hdr.ec_n = o_n;
 
-                        let actual = FileMeta::headers_match_strict(&s_hdr, &o_hdr);
                         let expected = {
-                            let mut s_cmp = s_hdr;
-                            let mut o_cmp = o_hdr;
+                            let mut s_cmp = s_hdr.clone();
+                            let mut o_cmp = o_hdr.clone();
                             if s_cmp.has_ec() != o_cmp.has_ec() {
                                 s_cmp.ec_m = 0;
                                 s_cmp.ec_n = 0;
@@ -4235,7 +4234,11 @@ mod tests {
                             }
                             s_cmp == o_cmp
                         };
-                        assert_eq!(actual, expected, "mismatch for s=({s_m},{s_n}) vs o=({o_m},{o_n})");
+                        assert_eq!(
+                            FileMeta::headers_match_strict(&s_hdr, &o_hdr),
+                            expected,
+                            "mismatch for s=({s_m},{s_n}) vs o=({o_m},{o_n})"
+                        );
                     }
                 }
             }

crates/heal/src/error.rs

@@ -574,7 +574,7 @@ mod tests {
             tonic::Status::internal("operation was canceled"),
         ] {
             assert!(
-                !Error::Storage(EcstoreError::from(status)).is_recoverable_heal(),
+                !Error::Storage(EcstoreError::from(status).clone()).is_recoverable_heal(),
                 "cancellation text alone must not change application error classification"
             );
         }
@@ -722,7 +722,7 @@ mod tests {
             for error in [
                 Error::Disk(DiskError::from(status.clone())),
                 Error::Storage(EcstoreError::from(status.clone())),
-                Error::Io(std::io::Error::from(DiskError::from(status))),
+                Error::Io(std::io::Error::from(DiskError::from(status.clone()))),
             ] {
                 assert!(!error.is_recoverable_heal(), "application RPC errors must stay terminal: {error:?}");
             }

crates/heal/src/heal/manager/auto_scan.rs

@@ -543,7 +543,7 @@ impl HealManager {
                                 .filter_map(|target| local_endpoints.get(target).cloned())
                                 .collect::<Vec<_>>();
                             if targets.len() != state.replacement_targets.len() {
-                                deferred_recovery_sets.insert(state.set_disk_id);
+                                deferred_recovery_sets.insert(state.set_disk_id.clone());
                                 continue;
                             }
                             // A pending handoff may resolve the predecessor to

crates/heal/src/heal/manager/scheduler.rs

@@ -257,6 +257,8 @@ impl HealManager {
                 let notify_clone = notify.clone();
                 let root_recovery_clone = root_recovery.clone();
                 let manager_cancel_token = cancel_token.clone();
+                let task_type_label_for_spawn = task_type_label.clone();
+                let task_set_label_for_spawn = task_set_label.clone();
                 let config_for_spawn = config.clone();
 
                 // start heal task
@@ -268,8 +270,8 @@ impl HealManager {
                         subsystem = LOG_SUBSYSTEM_MANAGER,
                         task_id,
                         priority = ?task_priority,
-                        heal_type = %task_type_label,
-                        set = %task_set_label,
+                        heal_type = %task_type_label_for_spawn,
+                        set = %task_set_label_for_spawn,
                         state = "task_started",
                         "Heal scheduler task started"
                     );
@@ -284,8 +286,8 @@ impl HealManager {
                                 component = LOG_COMPONENT_HEAL,
                                 subsystem = LOG_SUBSYSTEM_MANAGER,
                                 task_id,
-                                heal_type = %task_type_label,
-                                set = %task_set_label,
+                                heal_type = %task_type_label_for_spawn,
+                                set = %task_set_label_for_spawn,
                                 state = "task_completed",
                                 "Heal scheduler task completed"
                             );
@@ -298,8 +300,8 @@ impl HealManager {
                                 subsystem = LOG_SUBSYSTEM_MANAGER,
                                 state = "verification_unavailable",
                                 task_id,
-                                heal_type = %task_type_label,
-                                set = %task_set_label,
+                                heal_type = %task_type_label_for_spawn,
+                                set = %task_set_label_for_spawn,
                                 detail = %e,
                                 "Heal attempt completed without independent verification"
                             );
@@ -312,8 +314,8 @@ impl HealManager {
                                     component = LOG_COMPONENT_HEAL,
                                     subsystem = LOG_SUBSYSTEM_MANAGER,
                                     task_id,
-                                    heal_type = %task_type_label,
-                                    set = %task_set_label,
+                                    heal_type = %task_type_label_for_spawn,
+                                    set = %task_set_label_for_spawn,
                                     state = "task_retrying",
                                     retry_attempt = task.retry_attempts.saturating_add(1),
                                     error = %e,
@@ -326,8 +328,8 @@ impl HealManager {
                                     subsystem = LOG_SUBSYSTEM_MANAGER,
                                     state = "task_failed",
                                     task_id,
-                                    heal_type = %task_type_label,
-                                    set = %task_set_label,
+                                    heal_type = %task_type_label_for_spawn,
+                                    set = %task_set_label_for_spawn,
                                     error = %e,
                                     "Heal scheduler task failed"
                                 });

crates/heal/src/heal/mrf_queue.rs

@@ -2199,7 +2199,7 @@ async fn replay_into(
                 // runtime dispatches it after publishing the successor, avoiding
                 // a duplicate task in the startup-replay and steady-state paths.
                 if let Ok(Some(anchor)) = manager.durable_mrf_repair_anchor(&intent).await {
-                    durable_replay_anchors.push(anchor);
+                    durable_replay_anchors.push(anchor.clone());
                 }
                 let restored = intent_digest(&intent).and_then(|digest| take_lifecycle_record(&mut lifecycle, digest));
                 partial_writes.push((intent, restored));

crates/iam/src/mfa/recovery.rs

@@ -379,8 +379,8 @@ mod tests {
         let mut stored = generated.stored;
         assert_eq!(remaining(&stored), RECOVERY_CODE_COUNT as u32);
 
-        for code in generated.plaintext.iter().take(3) {
-            consume(&mut stored, code, now());
+        for index in 0..3 {
+            consume(&mut stored, &generated.plaintext[index], now());
         }
         assert_eq!(remaining(&stored), (RECOVERY_CODE_COUNT - 3) as u32);
     }

crates/keystone/src/middleware.rs

@@ -181,7 +181,7 @@ where
                         );
 
                         let body: BoxBody = Full::new(Bytes::from(error_xml))
-                            .map_err(|never| match never {})
+                            .map_err(|e| -> BoxError { Box::new(e) })
                             .boxed_unsync();
 
                         let response = Response::builder()

crates/kms/src/api_types.rs

@@ -1348,7 +1348,7 @@ mod tests {
             let summary = KmsConfigSummary::from(&config);
             let response = KmsStatusResponse {
                 status: KmsServiceStatus::Configured,
-                backend_type: Some(config.backend),
+                backend_type: Some(config.backend.clone()),
                 healthy: None,
                 config_summary: Some(summary),
             };

crates/kms/src/backup/local_restore.rs

@@ -977,7 +977,7 @@ async fn enumerate_target_conflicts(target: &Path, manifest: &BackupManifest) ->
             (stem.to_string(), format!("target already has key file {entry:?}"))
         } else if entry == LOCAL_KMS_MASTER_KEY_SALT_FILE {
             (
-                entry,
+                entry.clone(),
                 "target holds an orphan master-key KDF salt; it would poison every key created after the restore".to_string(),
             )
         } else {

crates/lifecycle/src/core.rs

@@ -718,7 +718,7 @@ impl Lifecycle for PersistedLifecycleConfiguration {
 
             let predicted = Event {
                 action: IlmAction::DeleteAction,
-                rule_id: rule.id.unwrap_or_default(),
+                rule_id: rule.id.clone().unwrap_or_default(),
                 due: Some(due),
                 noncurrent_days: 0,
                 newer_noncurrent_versions: 0,

crates/lock/src/fast_lock/tests.rs

@@ -487,7 +487,7 @@ mod fast_lock_tests {
             // Waiter: starts a touch later so the holder wins the lock first, then
             // must survive the whole hold and acquire promptly after the release.
             let waiter_mgr = manager.clone();
-            let waiter_key = key;
+            let waiter_key = key.clone();
             handles.push(tokio::spawn(async move {
                 sleep(Duration::from_millis(10)).await;
                 let request = ObjectLockRequest::new_write(waiter_key, "waiter").with_acquire_timeout(Duration::from_secs(5));

crates/madmin/src/metrics.rs

@@ -1537,7 +1537,7 @@ impl RealtimeMetrics {
         self.hosts.sort();
 
         for (k, v) in other.by_disk.into_iter() {
-            self.by_disk.entry(k).and_modify(|h| *h = v.clone()).or_insert(v);
+            self.by_disk.entry(k.to_string()).and_modify(|h| *h = v.clone()).or_insert(v);
         }
     }
 }

crates/obs/src/metrics/collectors/cluster_erasure_set.rs

@@ -120,8 +120,8 @@ pub fn collect_erasure_set_metrics(stats: &[ErasureSetStats]) -> Vec<PrometheusM
         );
         metrics.push(
             PrometheusMetric::from_descriptor(&ERASURE_SET_WRITE_HEALTH_MD, stat.write_health as f64)
-                .with_label_owned(POOL_ID_L, pool_id_label)
-                .with_label_owned(SET_ID_L, set_id_label),
+                .with_label_owned(POOL_ID_L, pool_id_label.clone())
+                .with_label_owned(SET_ID_L, set_id_label.clone()),
         );
     }
 

crates/policy/src/policy/opa.rs

@@ -102,7 +102,7 @@ fn check() -> Result<(), OpaConfigError> {
     let prefix = format!("{ENV_PREFIX}{POLICY_PLUGIN_SUB_SYS}").to_uppercase();
     for (key, value) in env_list {
         if key.starts_with(&prefix) {
-            candidate.insert(key, value);
+            candidate.insert(key.to_string(), value);
         }
     }
 

crates/policy/src/policy/policy.rs

@@ -141,7 +141,7 @@ impl Policy {
                 merged.version = p.version.clone();
             }
             for st in p.statements {
-                merged.statements.push(st);
+                merged.statements.push(st.clone());
             }
         }
         merged.drop_duplicate_statements();

crates/scanner-metrics/src/metrics.rs

@@ -3975,8 +3975,8 @@ impl Metrics {
         m.partial_cycles = self.partial_scan_cycles.load(Ordering::Relaxed);
 
         // Lifetime operation counts
-        for (i, operation) in self.operations.iter().enumerate().take(Metric::Last as usize) {
-            let count = operation.load(Ordering::Relaxed);
+        for i in 0..Metric::Last as usize {
+            let count = self.operations[i].load(Ordering::Relaxed);
             if count > 0
                 && let Some(metric) = Metric::from_index(i)
             {
@@ -3985,8 +3985,8 @@ impl Metrics {
         }
 
         // Last-minute stats for realtime metrics — now plain sync calls
-        for (i, latency) in self.latency.iter().enumerate().take(Metric::LastRealtime as usize) {
-            let last_min = latency.total();
+        for i in 0..Metric::LastRealtime as usize {
+            let last_min = self.latency[i].total();
             if last_min.n > 0
                 && let Some(metric) = Metric::from_index(i)
             {
@@ -4002,8 +4002,8 @@ impl Metrics {
         }
 
         // Lifetime ILM counts
-        for (i, action_count) in self.actions.iter().enumerate().take(IlmAction::ActionCount as usize) {
-            let count = action_count.load(Ordering::Relaxed);
+        for i in 0..IlmAction::ActionCount as usize {
+            let count = self.actions[i].load(Ordering::Relaxed);
             if count > 0
                 && let Some(action) = IlmAction::from_index(i)
             {
@@ -4012,8 +4012,8 @@ impl Metrics {
         }
 
         // Last-minute ILM latency — plain sync calls
-        for (i, action_latency) in self.actions_latency.iter().enumerate().take(IlmAction::ActionCount as usize) {
-            let last_min = action_latency.total();
+        for i in 0..IlmAction::ActionCount as usize {
+            let last_min = self.actions_latency[i].total();
             if last_min.n > 0
                 && let Some(action) = IlmAction::from_index(i)
             {
@@ -5324,37 +5324,6 @@ mod tests {
         assert_eq!(report.life_time_ops.get("scan_bucket_drive_failure"), Some(&1));
     }
 
-    #[tokio::test]
-    async fn report_includes_the_last_slot_of_each_realtime_and_ilm_table() {
-        assert_eq!(Metric::HealAbandonedObject as usize + 1, Metric::LastRealtime as usize);
-        assert_eq!(IlmAction::DelMarkerDeleteAllVersionsAction as usize + 1, IlmAction::ActionCount as usize);
-        let metrics = Metrics::new();
-        let last_ilm = IlmAction::DelMarkerDeleteAllVersionsAction as usize;
-        metrics.latency[Metric::HealAbandonedObject as usize].add(Duration::from_millis(1));
-        metrics.actions[last_ilm].fetch_add(1, Ordering::Relaxed);
-        metrics.actions_latency[last_ilm].add(Duration::from_millis(1));
-
-        let report = metrics.report().await;
-
-        assert_eq!(
-            report
-                .last_minute
-                .actions
-                .get("heal_abandoned_object")
-                .map(|action| action.count),
-            Some(1)
-        );
-        assert_eq!(report.life_time_ilm.get("del_marker_delete_all_versions"), Some(&1));
-        assert_eq!(
-            report
-                .last_minute
-                .ilm
-                .get("del_marker_delete_all_versions")
-                .map(|action| action.count),
-            Some(1)
-        );
-    }
-
     #[tokio::test]
     async fn active_bucket_drive_snapshot_is_structured_and_retired_on_end() {
         let metrics = Metrics::new();

crates/scanner/src/data_usage_define/tests.rs

@@ -3995,7 +3995,7 @@ fn test_lifecycle_cache_legacy_msgpack_oracle() {
             [
                 rmp_serde::to_vec(&frozen_tuple_read).expect("encode the fixed legacy tuple after production read"),
                 rmp_serde::to_vec(&frozen_map_read).expect("encode the fixed legacy map after production read"),
-                current_map,
+                current_map.clone(),
                 rmp_serde::to_vec(&crate::lifecycle_cache::LifecycleCache(&current.lifecycle))
                     .expect("encode the independent domain lifecycle payload"),
             ],

crates/scanner/src/scanner_folder/tests/checkpoint_fixture/periodic_snapshot.rs

@@ -578,7 +578,7 @@ async fn periodic_nested_snapshot_checks_suspended_ancestors_before_certifying_f
                 .new_cache
                 .copy_with_children(&scanner.old_cache, &hash_path(frontier), &Some(hash_path("bucket")));
         }
-        let root_entry = scanner.new_cache.root().expect("suspended root owner");
+        let root_entry = scanner.new_cache.root().expect("suspended root owner").clone();
         let mut ancestor_entry = scanner.new_cache.find(ancestor).expect("suspended subtree owner").clone();
         ancestor_entry.objects = 1;
         ancestor_entry.size = 2;

crates/tls-runtime/src/certs.rs

@@ -708,7 +708,7 @@ pub fn create_multi_cert_resolver(
     for (domain, (certs, key)) in cert_key_pairs {
         let certified_key = validate_certified_key(&certs, &key, &domain, true).map_err(io::Error::other)?;
         if domain == "default" {
-            default_cert = Some(Arc::new(certified_key));
+            default_cert = Some(Arc::new(certified_key.clone()));
         } else {
             resolver
                 .add(&domain, certified_key)

crates/tls-runtime/src/server.rs

@@ -74,7 +74,7 @@ impl ResolverState {
             let certified_key = CertifiedKey::new(certs, signing_key);
 
             if domain.as_str() == "default" {
-                default_cert = Some(Arc::new(certified_key));
+                default_cert = Some(Arc::new(certified_key.clone()));
             } else {
                 cert_resolver
                     .add(&domain, certified_key)

crates/trusted-proxies/src/utils/validation.rs

@@ -127,7 +127,11 @@ impl ValidationUtils {
             }
         }
 
-        value.len() <= 8192
+        if value.len() > 8192 {
+            return false;
+        }
+
+        true
     }
 
     /// Validates an entire HeaderMap for security.

crates/utils/src/string.rs

@@ -443,7 +443,10 @@ pub fn has_ellipses<T: AsRef<str>>(s: &[T]) -> bool {
         }
 
         // Bare `...` without braces (e.g. "1...64") is still an ellipsis attempt.
-        v.contains(ELLIPSES)
+        if v.contains(ELLIPSES) {
+            return true;
+        }
+        false
     })
 }
 

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

@@ -72,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.
 
@@ -88,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/handlers/kms_dynamic.rs

@@ -2132,7 +2132,7 @@ mod tests {
                 normalize_configure_request_secrets(&mut decode_vault_request(changed), Some(&existing))
                     .expect_err("the request cannot select an arbitrary server file");
                 if variant != "TokenFile" {
-                    let mut changed = value;
+                    let mut changed = value.clone();
                     changed["auth_method"][variant]["mount"] = serde_json::json!("attacker-auth");
                     normalize_configure_request_secrets(&mut decode_vault_request(changed), Some(&existing))
                         .expect_err("file contents cannot be sent to a different authentication mount");

rustfs/src/app/object/get.rs

@@ -2376,7 +2376,11 @@ fn should_buffer_get_object_in_memory_with_threshold(
         );
     }
 
-    response_content_length <= effective_threshold
+    if response_content_length > effective_threshold {
+        return false;
+    }
+
+    true
 }
 
 impl DefaultObjectUsecase {

rustfs/src/cgroup_resources/live.rs

@@ -255,7 +255,7 @@ fn resolve_scope_with_root(
                     depth,
                     Scope {
                         leaf,
-                        mount,
+                        mount: mount.clone(),
                         components,
                         directories,
                     },

rustfs/src/error.rs

@@ -2418,7 +2418,7 @@ mod tests {
             StorageError::PartMissingOrCorrupt,
         ] {
             assert!(is_object_read_unavailable(&source));
-            let api_error = slow_down_read_api_error(source);
+            let api_error = slow_down_read_api_error(source.clone());
             assert_eq!(api_error.code, S3ErrorCode::Custom(SLOW_DOWN_READ_CODE.into()));
             assert_eq!(api_error.message, SLOW_DOWN_READ_MESSAGE);
         }

rustfs/src/server/http/tests/production_batch_lock_rpc_tests.rs

@@ -496,8 +496,8 @@ impl Cluster {
             locks,
             _root: root,
         };
-        for (index, client) in cluster.clients.iter().enumerate().skip(1).take(3) {
-            let peer = DistributedLock::new("warmup".to_owned(), vec![Arc::clone(client)], 1);
+        for index in 1..=3 {
+            let peer = DistributedLock::new("warmup".to_owned(), vec![Arc::clone(&cluster.clients[index])], 1);
             drop(holder(&peer, keys(&format!("warmup-{index}"), 1).remove(0), "warmup-holder").await);
         }
         cluster.idle().await;

rustfs/src/server/rate_limit.rs

@@ -455,7 +455,9 @@ fn s3_too_many_requests_response(request_id: Option<&str>, limit_rpm: u32, throt
          <Message>Request rate limit exceeded. Reduce your request rate and retry after the indicated delay.</Message>\
          {request_id_xml}</Error>"
     );
-    let body: BoxBody = Full::new(Bytes::from(body)).map_err(|never| match never {}).boxed_unsync();
+    let body: BoxBody = Full::new(Bytes::from(body))
+        .map_err(|e| -> BoxError { Box::new(e) })
+        .boxed_unsync();
 
     let mut response = Response::new(body);
     *response.status_mut() = StatusCode::TOO_MANY_REQUESTS;
@@ -949,7 +951,7 @@ mod tests {
 
         fn call(&mut self, _req: Request<ReqBody>) -> Self::Future {
             let body: BoxBody = Full::new(Bytes::from_static(b"ok"))
-                .map_err(|never| match never {})
+                .map_err(|e| -> BoxError { Box::new(e) })
                 .boxed_unsync();
             ready(Ok(Response::new(body)))
         }

rustfs/src/server/readiness.rs

@@ -177,7 +177,7 @@ fn readiness_pending_dependency(stage: rustfs_common::SystemStage) -> &'static s
 fn service_not_ready_response(stage: rustfs_common::SystemStage) -> Response<BoxBody> {
     let pending = readiness_pending_dependency(stage);
     let body: BoxBody = Full::new(Bytes::from(format!("Service not ready: waiting for {pending}")))
-        .map_err(|never| match never {})
+        .map_err(|e| -> BoxError { Box::new(e) })
         .boxed_unsync();
 
     let mut response = Response::new(body);

rustfs/src/server/ssec_transport.rs

@@ -137,7 +137,9 @@ fn ssec_requires_tls_response(request_id: Option<&str>) -> Response<BoxBody> {
          <Message>Requests specifying Server Side Encryption with Customer provided keys must be made over a secure connection.</Message>\
          {request_id_xml}</Error>"
     );
-    let body: BoxBody = Full::new(Bytes::from(body)).map_err(|never| match never {}).boxed_unsync();
+    let body: BoxBody = Full::new(Bytes::from(body))
+        .map_err(|e| -> BoxError { Box::new(e) })
+        .boxed_unsync();
 
     let mut response = Response::new(body);
     *response.status_mut() = StatusCode::BAD_REQUEST;

rustfs/src/table_catalog/store/object.rs

@@ -2591,7 +2591,8 @@ where
                     TableCatalogEntryState::Active => &mut active,
                     TableCatalogEntryState::Deleted => {
                         if let Some(current) = deleted.as_ref() {
-                            deleted_overlap = Some((current.warehouse_object_prefix.clone(), resource.warehouse_object_prefix));
+                            deleted_overlap =
+                                Some((current.warehouse_object_prefix.clone(), resource.warehouse_object_prefix.clone()));
                             continue;
                         }
                         &mut deleted

rustfs/tests/connect_offline_enrollment.rs

@@ -1199,7 +1199,7 @@ fn enrollment_opens_no_descriptor_and_is_a_pure_byte_transform() {
         fields.push((
             challenge.challenge_id.clone(),
             challenge.nonce.clone(),
-            challenge.challenge_proof,
+            challenge.challenge_proof.clone(),
             signed_octets(&serde_json::from_slice::<Value>(&response).expect("JSON")),
         ));
     }

scripts/ci_gate.py

@@ -378,7 +378,7 @@ def configurations(jobs: dict[str, str]) -> list[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:
+                *, 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()
@@ -898,7 +898,7 @@ def main() -> int:
                 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", proof)
+                               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=(",", ":"))),

scripts/ci_merge_group_proof.py

@@ -31,16 +31,24 @@ def lookup(repository: str, tree: str, branch: str, token: str, opener=None) ->
     if not isinstance(runs, list):
         return None
     for run in runs:
-        if (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']}"):
+        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", "")

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

@@ -61,6 +61,7 @@ class ProofTests(unittest.TestCase):
                         {"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):

scripts/test_ci_plan.py

@@ -536,6 +536,13 @@ class GateTests(unittest.TestCase):
             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()}