Merge branch 'main' into codex/fix-4846-lock-quorum-timeout

.config/nextest.toml

@@ -300,9 +300,6 @@ path = "junit.xml"
 # negative-path siblings of each family stay in as regression guards.
 #   * rustfs#4843 — over-limit archive entry paths hard-reject the whole
 #     archive even under ignore-errors semantics.
-#   * rustfs#4846 — distributed-lock quorum tests misclassify as timeout
-#     under parallel load (multi-node in-process clusters; natural home is
-#     ci-7's nightly cluster lane).
 [profile.e2e-full]
 default-filter = """
     package(e2e_test)
@@ -311,7 +308,6 @@ default-filter = """
     & !test(/^replication_extension_test::/)
     & !test(/^multipart_auth_test::test_signed_put_object_extract_skips_invalid_entry_when_ignore_errors_enabled$/)
     & !test(/^snowball_auto_extract_test::tests::snowball_auto_extract_(ignores_invalid_entries_when_requested|supports_standard_headers_with_combined_extract_options)$/)
-    & !test(/^reliant::lock::test_distributed_lock_(2_nodes_grpc_read_survives_failed_node|4_nodes_grpc_read_write_quorum_split_with_two_failed_nodes)$/)
 """
 fail-fast = false
 

crates/e2e_test/src/reliant/lock.rs

@@ -15,9 +15,7 @@
 
 use super::{grpc_lock_client::GrpcLockClient, grpc_lock_server::spawn_lock_server};
 use rustfs_lock::client::{LockClient, local::LocalClient};
-use rustfs_lock::{
-    GlobalLockManager, LockError, LockInfo, LockRequest, LockResponse, LockStats, LockType, NamespaceLock, ObjectKey,
-};
+use rustfs_lock::{GlobalLockManager, LockInfo, LockRequest, LockResponse, LockStats, LockType, NamespaceLock, ObjectKey};
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -35,7 +33,11 @@ struct FailingClient;
 #[async_trait::async_trait]
 impl rustfs_lock::LockClient for FailingClient {
     async fn acquire_lock(&self, _request: &rustfs_lock::LockRequest) -> rustfs_lock::Result<LockResponse> {
-        Err(LockError::internal("simulated gRPC node failure"))
+        // Match RemoteClient's transport-failure response so the coordinator can count this node toward quorum loss.
+        Ok(LockResponse::failure(
+            "Remote lock RPC failed: simulated gRPC node failure",
+            Duration::ZERO,
+        ))
     }
 
     async fn release(&self, _lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result<bool> {

crates/e2e_test/src/reliant/tiering.rs

@@ -382,6 +382,7 @@ struct ManualTransitionRunReport {
     skipped_delete_marker: u64,
     skipped_directory: u64,
     skipped_replication: u64,
+    skipped_already_transitioned: u64,
     skipped_already_in_flight: u64,
     skipped_queue_full: u64,
     skipped_queue_closed: u64,
@@ -407,6 +408,41 @@ fn assert_completed_or_in_flight_partial(state: &str, report: &ManualTransitionR
     }
 }
 
+fn assert_conflict_winner_report(state: &str, report: &ManualTransitionRunReport, expected_objects: u64, context: &str) {
+    assert_completed_or_in_flight_partial(state, report, context);
+    if report.skipped_already_in_flight > 0 {
+        assert!(
+            report.scanned <= expected_objects,
+            "{context}: scanned more objects than the conflict scope contains: {report:#?}"
+        );
+        assert!(
+            report.eligible <= expected_objects,
+            "{context}: marked more objects eligible than the conflict scope contains: {report:#?}"
+        );
+        assert_eq!(
+            report.enqueued + report.skipped_already_in_flight,
+            report.eligible,
+            "{context}: partial in-flight accounting must cover every eligible object: {report:#?}"
+        );
+    } else {
+        assert_eq!(report.scanned, expected_objects, "{context}: {report:#?}");
+        assert_eq!(
+            report.eligible + report.skipped_already_transitioned,
+            expected_objects,
+            "{context}: {report:#?}"
+        );
+        assert_eq!(
+            report.enqueued + report.skipped_already_in_flight,
+            expected_objects,
+            "{context}: {report:#?}"
+        );
+    }
+    assert_eq!(
+        report.transition_completed, report.enqueued,
+        "{context}: winner must wait for all queued transitions: {report:#?}"
+    );
+}
+
 #[derive(Debug, Deserialize)]
 struct ManualTransitionQueueSnapshot {
     queue_capacity: u64,
@@ -1240,15 +1276,8 @@ async fn test_manual_transition_async_scope_conflicts_report_active_job() -> Tes
     cold_client.create_bucket().bucket(TIER_BUCKET).send().await?;
 
     let mut hot = RustFSTestEnvironment::new().await?;
-    hot.start_rustfs_server_with_env(
-        vec![],
-        &[
-            ("RUSTFS_SCANNER_ENABLED", "false"),
-            ("RUSTFS_SCANNER_CYCLE", "3600"),
-            (MANUAL_TRANSITION_CANCEL_BARRIER_ENV, "1"),
-        ],
-    )
-    .await?;
+    hot.start_rustfs_server_with_env(vec![], &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")])
+        .await?;
     let hot_client = hot.create_s3_client();
     add_rustfs_tier(&hot, &cold).await?;
 
@@ -1318,21 +1347,20 @@ async fn test_manual_transition_async_scope_conflicts_report_active_job() -> Tes
     assert_eq!(conflict.cancel_endpoint, status_endpoint);
     assert!(!conflict.scope_key.is_empty());
 
-    manual_transition_job_cancel(&hot, cancel_endpoint).await?;
-
     let terminal = wait_for_manual_transition_job_terminal(&hot, status_endpoint, MANUAL_ASYNC_CONFLICT_TERMINAL_TIMEOUT).await?;
     assert_eq!(terminal.job_id, job_id);
-    assert_eq!(terminal.status, "cancelled", "terminal conflict winner response: {terminal:#?}");
     assert!(!terminal.report.dry_run);
     assert_eq!(terminal.report.bucket, MANUAL_ASYNC_CONFLICT_BUCKET);
     assert_eq!(terminal.report.prefix, accepted.report.prefix);
-    assert!(terminal.report.cancelled, "terminal conflict winner response: {terminal:#?}");
-    assert_eq!(terminal.report.scanned, 0, "terminal conflict winner response: {terminal:#?}");
-    assert_eq!(terminal.report.enqueued, 0, "terminal conflict winner response: {terminal:#?}");
-    assert_eq!(
-        terminal.report.transition_completed, 0,
-        "terminal conflict winner response: {terminal:#?}"
+    assert_conflict_winner_report(
+        &terminal.status,
+        &terminal.report,
+        MANUAL_ASYNC_CONFLICT_OBJECTS as u64,
+        "terminal conflict winner response",
     );
+    assert_eq!(terminal.report.dry_run_eligible, 0, "terminal conflict winner response: {terminal:#?}");
+    assert_eq!(terminal.report.transition_failed, 0, "terminal conflict winner response: {terminal:#?}");
+    assert_eq!(terminal.report.tier_failure, 0, "terminal conflict winner response: {terminal:#?}");
     let after_remote_count = cold_tier_object_count(&cold_client).await?;
     assert!(after_remote_count >= before_remote_count);
     assert!(after_remote_count <= before_remote_count + MANUAL_ASYNC_CONFLICT_OBJECTS);

crates/scanner/src/scanner_folder.rs

@@ -75,7 +75,7 @@ const DATA_SCANNER_COMPACT_LEAST_OBJECT: usize = 500;
 const DATA_SCANNER_COMPACT_AT_CHILDREN: usize = 10000;
 const DATA_SCANNER_COMPACT_AT_FOLDERS: usize = DATA_SCANNER_COMPACT_AT_CHILDREN / 4;
 const DATA_SCANNER_FORCE_COMPACT_AT_FOLDERS: usize = 250_000;
-const SCANNER_LIST_PATH_RAW_STALL_TIMEOUT: Duration = Duration::from_secs(60);
+const SCANNER_LIST_PATH_RAW_TIMEOUT: Duration = Duration::from_secs(60);
 const SCANNER_ENTRY_PROGRESS_BATCH: u64 = 32;
 const SCANNER_ENTRY_PROGRESS_INTERVAL: Duration = Duration::from_secs(30);
 const DEFAULT_HEAL_OBJECT_SELECT_PROB: u32 = 1024;
@@ -100,21 +100,6 @@ static SCANNER_INLINE_HEAL_WARN_ONCE: Once = Once::new();
 static SCANNER_INLINE_HEAL_METRICS_ONCE: Once = Once::new();
 static SCANNER_ALERT_METRICS_ONCE: Once = Once::new();
 
-#[cfg(test)]
-type ListPathRawTimeoutSnapshot = (bool, Option<Duration>, Option<Duration>);
-
-fn scanner_abandoned_child_list_options() -> ListPathRawOptions {
-    // A complete heal walk scales with bucket size and may legitimately take
-    // longer than a fixed wall-clock budget. Keep the total duration unbounded;
-    // Retain the scanner's per-read stall budget and keep cancellation controlled
-    // by the scanner cycle token.
-    ListPathRawOptions {
-        skip_walkdir_total_timeout: true,
-        walkdir_stall_timeout: Some(SCANNER_LIST_PATH_RAW_STALL_TIMEOUT),
-        ..Default::default()
-    }
-}
-
 pub fn data_usage_update_dir_cycles() -> u32 {
     rustfs_utils::get_env_u32(ENV_DATA_USAGE_UPDATE_DIR_CYCLES, DATA_USAGE_UPDATE_DIR_CYCLES)
 }
@@ -1296,8 +1281,6 @@ pub struct FolderScanner {
     skip_heal: Arc<std::sync::atomic::AtomicBool>,
     local_disk: Arc<Disk>,
     pending_heals_changed: bool,
-    #[cfg(test)]
-    list_path_raw_options_observer: Option<mpsc::UnboundedSender<ListPathRawTimeoutSnapshot>>,
 }
 
 impl FolderScanner {
@@ -2427,79 +2410,75 @@ impl FolderScanner {
                 let bucket_clone = bucket.clone();
                 let prefix_clone = prefix.clone();
                 let child_ctx_clone = child_ctx.clone();
-                #[cfg(test)]
-                let list_path_raw_options_observer = self.list_path_raw_options_observer.clone();
 
                 tokio::spawn(async move {
-                    let options = ListPathRawOptions {
-                        disks,
-                        bucket: bucket_clone.clone(),
-                        path: prefix_clone.clone(),
-                        recursive: true,
-                        report_not_found: true,
-                        min_disks: disks_quorum,
-                        agreed: Some(Box::new(move |entry: MetaCacheEntry| {
-                            let entry_name = entry.name.clone();
-                            let agreed_tx = agreed_tx.clone();
-                            Box::pin(async move {
-                                if let Err(e) = agreed_tx.send(entry_name).await {
-                                    error!(
-                                        target: "rustfs::scanner::folder",
-                                        event = EVENT_SCANNER_FOLDER_STATE,
-                                        component = LOG_COMPONENT_SCANNER,
-                                        subsystem = LOG_SUBSYSTEM_FOLDER,
-                                        entry = %entry.name,
-                                        state = "list_path_agreed_send_failed",
-                                        error = %e,
-                                        "Scanner list_path_raw agreed callback failed"
-                                    );
-                                }
-                            })
-                        })),
-                        partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option<DiskError>]| {
-                            let partial_tx = partial_tx.clone();
-                            Box::pin(async move {
-                                if let Err(e) = partial_tx.send(entries).await {
-                                    error!(
-                                        target: "rustfs::scanner::folder",
-                                        event = EVENT_SCANNER_FOLDER_STATE,
-                                        component = LOG_COMPONENT_SCANNER,
-                                        subsystem = LOG_SUBSYSTEM_FOLDER,
-                                        state = "list_path_partial_send_failed",
-                                        error = %e,
-                                        "Scanner list_path_raw partial callback failed"
-                                    );
-                                }
-                            })
-                        })),
-                        finished: Some(Box::new(move |errs: &[Option<DiskError>]| {
-                            let finished_tx = finished_tx.clone();
-                            let errs_clone = errs.to_vec();
-                            Box::pin(async move {
-                                if let Err(e) = finished_tx.send(errs_clone).await {
-                                    error!(
-                                        target: "rustfs::scanner::folder",
-                                        event = EVENT_SCANNER_FOLDER_STATE,
-                                        component = LOG_COMPONENT_SCANNER,
-                                        subsystem = LOG_SUBSYSTEM_FOLDER,
-                                        state = "list_path_finished_send_failed",
-                                        error = %e,
-                                        "Scanner list_path_raw finished callback failed"
-                                    );
-                                }
-                            })
-                        })),
-                        ..scanner_abandoned_child_list_options()
-                    };
-                    #[cfg(test)]
-                    if let Some(observer) = list_path_raw_options_observer {
-                        let _ = observer.send((
-                            options.skip_walkdir_total_timeout,
-                            options.walkdir_timeout,
-                            options.walkdir_stall_timeout,
-                        ));
-                    }
-                    if let Err(e) = list_path_raw(child_ctx_clone.clone(), options).await {
+                    if let Err(e) = list_path_raw(
+                        child_ctx_clone.clone(),
+                        ListPathRawOptions {
+                            disks,
+                            bucket: bucket_clone.clone(),
+                            path: prefix_clone.clone(),
+                            recursive: true,
+                            report_not_found: true,
+                            min_disks: disks_quorum,
+                            walkdir_timeout: Some(SCANNER_LIST_PATH_RAW_TIMEOUT),
+                            walkdir_stall_timeout: Some(SCANNER_LIST_PATH_RAW_TIMEOUT),
+                            agreed: Some(Box::new(move |entry: MetaCacheEntry| {
+                                let entry_name = entry.name.clone();
+                                let agreed_tx = agreed_tx.clone();
+                                Box::pin(async move {
+                                    if let Err(e) = agreed_tx.send(entry_name).await {
+                                        error!(
+                                            target: "rustfs::scanner::folder",
+                                            event = EVENT_SCANNER_FOLDER_STATE,
+                                            component = LOG_COMPONENT_SCANNER,
+                                            subsystem = LOG_SUBSYSTEM_FOLDER,
+                                            entry = %entry.name,
+                                            state = "list_path_agreed_send_failed",
+                                            error = %e,
+                                            "Scanner list_path_raw agreed callback failed"
+                                        );
+                                    }
+                                })
+                            })),
+                            partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option<DiskError>]| {
+                                let partial_tx = partial_tx.clone();
+                                Box::pin(async move {
+                                    if let Err(e) = partial_tx.send(entries).await {
+                                        error!(
+                                            target: "rustfs::scanner::folder",
+                                            event = EVENT_SCANNER_FOLDER_STATE,
+                                            component = LOG_COMPONENT_SCANNER,
+                                            subsystem = LOG_SUBSYSTEM_FOLDER,
+                                            state = "list_path_partial_send_failed",
+                                            error = %e,
+                                            "Scanner list_path_raw partial callback failed"
+                                        );
+                                    }
+                                })
+                            })),
+                            finished: Some(Box::new(move |errs: &[Option<DiskError>]| {
+                                let finished_tx = finished_tx.clone();
+                                let errs_clone = errs.to_vec();
+                                Box::pin(async move {
+                                    if let Err(e) = finished_tx.send(errs_clone).await {
+                                        error!(
+                                            target: "rustfs::scanner::folder",
+                                            event = EVENT_SCANNER_FOLDER_STATE,
+                                            component = LOG_COMPONENT_SCANNER,
+                                            subsystem = LOG_SUBSYSTEM_FOLDER,
+                                            state = "list_path_finished_send_failed",
+                                            error = %e,
+                                            "Scanner list_path_raw finished callback failed"
+                                        );
+                                    }
+                                })
+                            })),
+                            ..Default::default()
+                        },
+                    )
+                    .await
+                    {
                         if is_missing_path_disk_error(&e) {
                             debug!(
                                 target: "rustfs::scanner::folder",
@@ -2841,8 +2820,6 @@ pub async fn scan_data_folder(
         skip_heal,
         local_disk,
         pending_heals_changed: false,
-        #[cfg(test)]
-        list_path_raw_options_observer: None,
     };
 
     // Check if context is cancelled
@@ -3049,7 +3026,6 @@ mod tests {
             skip_heal: Arc::new(AtomicBool::new(false)),
             local_disk: disk,
             pending_heals_changed: false,
-            list_path_raw_options_observer: None,
         };
 
         (scanner, temp_dir)
@@ -4234,8 +4210,6 @@ mod tests {
         scanner.heal_object_select = 1;
         scanner.disks = disks;
         scanner.disks_quorum = 2;
-        let (options_tx, mut options_rx) = mpsc::unbounded_channel();
-        scanner.list_path_raw_options_observer = Some(options_tx);
         scanner.old_cache.replace(
             &format!("{bucket}/{object}"),
             bucket,
@@ -4257,11 +4231,6 @@ mod tests {
             .expect("scan_folder should not hang after list_path_raw finishes")
             .expect("scan_folder should finish successfully");
 
-        let observed_options = tokio::time::timeout(Duration::from_secs(1), options_rx.recv())
-            .await
-            .expect("abandoned-child listing options should be observed promptly")
-            .expect("abandoned-child listing options channel should remain open");
-        assert_eq!(observed_options, (true, None, Some(SCANNER_LIST_PATH_RAW_STALL_TIMEOUT)));
         let root = scanner
             .new_cache
             .checked_flatten(bucket)