fix(tests): stabilize main CI concurrency fixtures (#6655)

* fix(tests): retry transient inventory replacement failures

* test(rio): isolate h2 keepalive fixture runtime

crates/rio/src/http_reader.rs

@@ -2172,8 +2172,14 @@ mod tests {
         Some((format!("http://{addr}/stream"), handle))
     }
 
-    async fn start_temporarily_blocked_h2_server(blocked_for: Duration) -> Option<(String, tokio::task::JoinHandle<()>)> {
+    struct BlockedH2Server {
-        let listener = match TcpListener::bind("127.0.0.1:0").await {
+        url: String,
+        shutdown: tokio::sync::oneshot::Sender<()>,
+        handle: std::thread::JoinHandle<()>,
+    }
+
+    fn start_temporarily_blocked_h2_server(blocked_for: Duration) -> Option<BlockedH2Server> {
+        let listener = match std::net::TcpListener::bind("127.0.0.1:0") {
             Ok(listener) => listener,
             Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return None,
             Err(err) => panic!("HTTP/2 test listener should bind: {err}"),
@@ -2181,34 +2187,49 @@ mod tests {
         let addr = listener
             .local_addr()
             .expect("HTTP/2 test listener address should be available");
-        let handle = tokio::spawn(async move {
+        listener
-            let (stream, _) = listener.accept().await.expect("HTTP/2 test server should accept client");
+            .set_nonblocking(true)
-            let service = service_fn(move |_request: Request<Incoming>| async move {
+            .expect("HTTP/2 test listener should become nonblocking");
-                let body_stream =
+        let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel();
-                    stream::once(async { Ok::<_, Infallible>(Bytes::from_static(b"partial")) }).chain(stream::once(async move {
+        let handle = std::thread::spawn(move || {
-                        tokio::time::sleep(Duration::from_millis(10)).await;
+            let runtime = tokio::runtime::Builder::new_current_thread()
-                        // Model a saturated peer whose connection task cannot
+                .enable_all()
-                        // process a PING ACK even though an active response body
+                .build()
-                        // remains valid. Blocking this one connection task is
+                .expect("HTTP/2 test runtime should build");
-                        // intentional; the multi-thread test runtime keeps the
+            runtime.block_on(async move {
-                        // client timer and transport progressing.
+                let listener = TcpListener::from_std(listener).expect("HTTP/2 test listener should attach to its runtime");
-                        tokio::task::block_in_place(|| std::thread::sleep(blocked_for));
+                let (stream, _) = listener.accept().await.expect("HTTP/2 test server should accept client");
-                        Ok::<_, Infallible>(Bytes::from_static(b"complete"))
+                let service = service_fn(move |_request: Request<Incoming>| async move {
-                    }));
+                    let body_stream = stream::once(async { Ok::<_, Infallible>(Bytes::from_static(b"partial")) }).chain(
-                Ok::<_, Infallible>(Response::new(Body::from_stream(body_stream)))
+                        stream::once(async move {
+                            tokio::time::sleep(Duration::from_millis(10)).await;
+                            // Model a saturated peer whose connection task cannot
+                            // process a PING ACK even though an active response body
+                            // remains valid. The server has its own runtime so this
+                            // cannot starve the client-side keepalive timer.
+                            std::thread::sleep(blocked_for);
+                            Ok::<_, Infallible>(Bytes::from_static(b"complete"))
+                        }),
+                    );
+                    Ok::<_, Infallible>(Response::new(Body::from_stream(body_stream)))
+                });
+                tokio::select! {
+                    _ = http2::Builder::new(TokioExecutor::new()).serve_connection(TokioIo::new(stream), service) => {},
+                    _ = shutdown_rx => {},
+                }
             });
-            let _ = http2::Builder::new(TokioExecutor::new())
-                .serve_connection(TokioIo::new(stream), service)
-                .await;
         });
 
-        Some((format!("http://{addr}"), handle))
+        Some(BlockedH2Server {
+            url: format!("http://{addr}"),
+            shutdown: shutdown_tx,
+            handle,
+        })
     }
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
     async fn data_plane_liveness_does_not_abort_a_valid_h2_stream_while_peer_is_busy() {
-        let blocked_for = Duration::from_millis(500);
+        let Some(baseline_server) = start_temporarily_blocked_h2_server(Duration::from_secs(2)) else {
-        let Some((baseline_url, baseline_server)) = start_temporarily_blocked_h2_server(blocked_for).await else {
             return;
         };
         let aggressive_keepalive_client = Client::builder()
@@ -2220,8 +2241,13 @@ mod tests {
             .build()
             .expect("aggressive HTTP/2 keepalive client should build");
 
-        let baseline_result = tokio::time::timeout(Duration::from_secs(2), async {
+        let baseline_result = tokio::time::timeout(Duration::from_secs(4), async {
-            aggressive_keepalive_client.get(&baseline_url).send().await?.bytes().await
+            aggressive_keepalive_client
+                .get(&baseline_server.url)
+                .send()
+                .await?
+                .bytes()
+                .await
         })
         .await
         .expect("aggressive keepalive body should finish before the outer test deadline");
@@ -2229,16 +2255,20 @@ mod tests {
             baseline_result.is_err(),
             "fixture must reproduce a PING timeout while the server connection task is blocked"
         );
-        baseline_server.abort();
+        let _ = baseline_server.shutdown.send(());
+        baseline_server
+            .handle
+            .join()
+            .expect("baseline HTTP/2 test server should exit");
 
-        let Some((candidate_url, candidate_server)) = start_temporarily_blocked_h2_server(blocked_for).await else {
+        let Some(candidate_server) = start_temporarily_blocked_h2_server(Duration::from_millis(500)) else {
             return;
         };
         let candidate_client = apply_http_data_plane_liveness(Client::builder().no_proxy().http2_prior_knowledge())
             .build()
             .expect("internode data-plane HTTP/2 client should build");
         let response = tokio::time::timeout(Duration::from_secs(2), async {
-            let response = candidate_client.get(&candidate_url).send().await?;
+            let response = candidate_client.get(&candidate_server.url).send().await?;
             let version = response.version();
             let body = response.bytes().await?;
             Ok::<_, reqwest::Error>((version, body))
@@ -2249,7 +2279,11 @@ mod tests {
 
         assert_eq!(response.0, Version::HTTP_2);
         assert_eq!(response.1, Bytes::from_static(b"partialcomplete"));
-        candidate_server.abort();
+        let _ = candidate_server.shutdown.send(());
+        candidate_server
+            .handle
+            .join()
+            .expect("candidate HTTP/2 test server should exit");
     }
 
     #[test]

rustfs/src/connect/inventory.rs

@@ -1805,7 +1805,20 @@ mod tests {
                     second_writer.clone()
                 };
                 let bytes = encode_envelope(snapshot, captured_at.to_owned()).expect("envelope");
-                writer.replace_file("latest.json", &bytes, || false).expect("atomic replace");
+                let mut committed = false;
+                for _ in 0..100 {
+                    match writer.replace_file("latest.json", &bytes, || false) {
+                        Ok(()) => {
+                            committed = true;
+                            break;
+                        }
+                        Err(InventoryError::StateIo | InventoryError::PersistenceSecurity) => {
+                            std::thread::sleep(std::time::Duration::from_millis(1));
+                        }
+                        Err(error) => panic!("atomic replace: {error}"),
+                    }
+                }
+                assert!(committed, "atomic replace did not recover from transient fail-closed errors");
             }
         });
         let now = chrono::DateTime::parse_from_rfc3339(captured_at)