test(ecstore): restore 30s lock-timeout guard for concurrent resend test (#4370)

concurrent_resend_same_part_commits_one_generation keeps failing on CI
with Lock(Timeout ... 5s) even after the fast-lock lost-wakeup fix
(NOTIFY_WAIT_CAP re-polling) landed — observed on rustfs#4365's Test and
Lint runs. Two independent causes exist and both are real:

1. the lost/stolen wakeup stall — fixed in fast_lock::shard; and
2. the six legitimately serialized cross-disk commits exceeding the
   small 5s default acquire timeout under full-suite CI disk load.

The lost-wakeup fix reverted the earlier 30s test override on the
assumption that cause 1 was the whole story; CI shows cause 2 stands on
its own. Restore the temp_env override (production-default 30s) around
the concurrent section so the regression guard asserts the correctness
property (exactly one intact generation), not CI disk latency. The
NOTIFY_WAIT_CAP fix stays untouched.

Verification:
- cargo test -p rustfs-ecstore --lib concurrent_resend_same_part_commits_one_generation

Ref: rustfs/backlog#882

Co-authored-by: heihutu <heihutu@gmail.com>

crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs

@@ -4717,31 +4717,43 @@ mod tests {
             })
             .collect();
 
-        let mut tasks = tokio::task::JoinSet::new();
+        // Two independent causes can produce a spurious lock-acquire timeout
-        for payload in candidates.iter().cloned() {
+        // here, and both must stay covered:
-            let store = ecstore.clone();
+        // 1. A lost/stolen fast-lock wakeup could strand a waiter until the
-            let bucket = bucket.clone();
+        //    deadline — fixed for real in fast_lock::shard by bounding each
-            let upload_id = upload.upload_id.clone();
+        //    notification wait (NOTIFY_WAIT_CAP re-polling).
-            tasks.spawn(async move {
+        // 2. Under the full nextest suite on loaded CI disks, the six
-                let mut data = PutObjReader::from_vec(payload.clone());
+        //    *legitimately serialized* cross-disk commits can exceed the small
-                store
+        //    default acquire timeout (5s) all by themselves — observed on CI
-                    .put_object_part(&bucket, object, &upload_id, 1, &mut data, &ObjectOptions::default())
+        //    even with fix 1 in place. Raise the timeout to the production
-                    .await
+        //    default (30s) for the concurrent section so the guard asserts the
-                    .map(|info| (info, payload))
+        //    correctness property (exactly one intact generation), not disk
-            });
+        //    latency. `#[serial]` keeps the process-wide env override isolated.
-        }
+        let results = temp_env::async_with_vars([(rustfs_config::ENV_OBJECT_LOCK_ACQUIRE_TIMEOUT, Some("30"))], async {
+            let mut tasks = tokio::task::JoinSet::new();
+            for payload in candidates.iter().cloned() {
+                let store = ecstore.clone();
+                let bucket = bucket.clone();
+                let upload_id = upload.upload_id.clone();
+                tasks.spawn(async move {
+                    let mut data = PutObjReader::from_vec(payload.clone());
+                    store
+                        .put_object_part(&bucket, object, &upload_id, 1, &mut data, &ObjectOptions::default())
+                        .await
+                        .map(|info| (info, payload))
+                });
+            }
 
-        // Every concurrent resend must succeed. The per-uploadId commit lock must
+            // Every concurrent resend must succeed; the commit lock must never
-        // not starve a waiter into a lock-acquire timeout: the shared fast-lock
+            // starve a waiter into a timeout.
-        // notify pool could otherwise route this lock's wakeup to a waiter of an
+            let mut results = Vec::new();
-        // unrelated lock and strand this one until the deadline (fixed in
+            while let Some(joined) = tasks.join_next().await {
-        // fast_lock::shard by bounding each notification wait, so a lost/stolen
+                let outcome = joined.expect("put_object_part task should not panic");
-        // wakeup degrades to bounded re-polling instead of a hard timeout).
+                results.push(outcome.expect("every concurrent same-part resend must succeed without lock timeout"));
-        let mut results = Vec::new();
+            }
-        while let Some(joined) = tasks.join_next().await {
+            results
-            let outcome = joined.expect("put_object_part task should not panic");
+        })
-            results.push(outcome.expect("every concurrent same-part resend must succeed without lock timeout"));
+        .await;
-        }
         assert_eq!(results.len(), candidates.len());
 
         // Exactly one generation is visible after the serialized commits, and its