From b4f58b81cfdb2a833429b2299e8728885f686cd0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=94=90=E5=B0=8F=E9=B8=AD?= Date: Fri, 16 Oct 2026 05:03:31 +0800 Subject: [PATCH] test(e2e): gate the rc.5 baseline restarts on drive re-admission (#8595) * test(e2e): gate the rc.5 baseline restarts on drive re-admission The mixed-version rolling upgrade failed three times on unrelated branches with `MultipartAcknowledgementDeadline { node: 2, key: "one-current-node/..." }`. The baseline kills the next rc.5 node about 10 ms after the previous one reports ListBuckets, while that node is still replaying its startup traffic. Two failed operations drop the peer's drive to Offline on the pinned release, and only the 2 s recovery probe brings it back three successes later. In both failing runs node 2 held node 3's drive as `faulty` when the multipart load started three seconds later; it then committed every upload at three of four drives with a heal per object, and on a two-vCPU runner one upload exceeded the 30 s acknowledgement budget. The green run on the same branch only reached `suspect`. An all-rc.5 cluster reproduces the same degradation after the same restart sequence and an all-current cluster shows none of it. Write a slash-free probe object before the first restart and, after each baseline restart, block until every node reports all four drives of that object as `ok` through a dry-run object heal. `storageinfo` and `info` only carry each node's local drives; the object heal runs on the receiving node and classifies a short-circuited remote drive as `faulty`. The gate stays in the all-rc.5 baseline: admin heal first probes every peer's heal-control capability, and a mixed-version cluster fails that probe on both binaries. Record every node's lock table in `mixed-multipart-failure.json`; the stalled writer is a previous-release node whose table the current node cannot show. The 30 s budget, layout, load and assertions are unchanged. * test(e2e): probe drive re-admission with a parity-0 write per node The dry-run heal gate observed only node 0: the pinned rc.5 admin heal forwards start and status requests to the heal-control coordinator whatever node receives them, so the loop checked node 0's drive handles four times and never the restarted peers' handles. Run the cluster with `RUSTFS_STORAGE_CLASS_RRS=EC:0` and probe each node with a reduced-redundancy PUT instead. A parity-0 write on the four-drive set needs all four drives (`write_quorum = data_drives`), so the PUT fails with 503 `ServiceUnavailable` until the serving node writes through every one of its own drive handles. The probe is plain S3 and executes on the receiving node on both binaries, so the gate now also runs before each mixed-version load and after the last upgrade. Standard-class objects keep the default layout. `parity_zero_write_needs_every_drive` pins the premise on the workspace build: with one node down a standard write still commits, the reduced-redundancy write fails with 503, and the gate converges once the node returns. The diagnostic snapshot test now checks the per-node lock probe array. --- .../src/upgrade_compatibility_test.rs | 141 ++++++++++++++++-- 1 file changed, 130 insertions(+), 11 deletions(-) diff --git a/crates/e2e_test/src/upgrade_compatibility_test.rs b/crates/e2e_test/src/upgrade_compatibility_test.rs index 35e0c8eaf..182a60b64 100644 --- a/crates/e2e_test/src/upgrade_compatibility_test.rs +++ b/crates/e2e_test/src/upgrade_compatibility_test.rs @@ -28,7 +28,7 @@ use aws_sdk_s3::types::{ BucketLifecycleConfiguration, BucketVersioningStatus, CompletedMultipartUpload, CompletedPart, DefaultRetention, ExpirationStatus, LifecycleExpiration, LifecycleRule, LifecycleRuleFilter, ObjectAttributes, ObjectLockConfiguration, ObjectLockEnabled, ObjectLockRetentionMode, ObjectLockRule, PublicAccessBlockConfiguration, ServerSideEncryption, - ServerSideEncryptionByDefault, ServerSideEncryptionConfiguration, ServerSideEncryptionRule, Tag, Tagging, + ServerSideEncryptionByDefault, ServerSideEncryptionConfiguration, ServerSideEncryptionRule, StorageClass, Tag, Tagging, VersioningConfiguration, }; use http::{Method, StatusCode}; @@ -66,6 +66,16 @@ const MIXED_MULTIPART_UPLOAD_TIMEOUT: Duration = Duration::from_secs(30); // probe_interval (2s) x success_threshold (3) after it comes back; 30s // comfortably covers that window plus CI scheduling jitter. const LISTING_CONVERGENCE_TIMEOUT: Duration = Duration::from_secs(30); +// Parity 0 for the reduced-redundancy class only. A parity-0 write on the +// four-drive set needs all four drives (`write_quorum = data_drives`), so a +// PUT with that class succeeds only once the serving node writes through every +// one of its own drive handles. Standard-class objects keep the default layout. +const ALL_DRIVES_STORAGE_CLASS_ENV: (&str, &str) = ("RUSTFS_STORAGE_CLASS_RRS", "EC:0"); +// A peer drops a restarted node's drive to Offline after two failed +// operations and only re-admits it through probe_interval (2s) x +// success_threshold (3) once the node answers again. The budget covers that +// for every node after the slowest restart in the sequence, plus CI jitter. +const DRIVE_READMISSION_TIMEOUT: Duration = Duration::from_secs(60); // Bucket-configuration upgrade/rollback scenarios (rustfs#7172, #7183, #7089). const CONFIG_PLAIN_BUCKET: &str = "upgrade-config-plain"; @@ -509,6 +519,9 @@ async fn observe_mixed_multipart_failure( let network_deadline = started + Duration::from_secs(4); let path = format!("/{MIXED_BUCKET}/{}", failure.key); let _ = tokio::time::timeout_at(deadline, async { + // Every node keeps its own lock table, and the writer that stalls is + // usually a previous-release node; the current node's snapshot alone + // cannot show which holder blocks the upload. let (heads, locks) = tokio::join!( futures::future::join_all((0..cluster.nodes.len()).map(|node| probe_mixed_multipart_failure( cluster, @@ -517,7 +530,13 @@ async fn observe_mixed_multipart_failure( true, network_deadline ))), - probe_mixed_multipart_failure(cluster, 0, "/rustfs/admin/v3/top/locks?count=1000", false, network_deadline), + futures::future::join_all((0..cluster.nodes.len()).map(|node| probe_mixed_multipart_failure( + cluster, + node, + "/rustfs/admin/v3/top/locks?count=1000", + false, + network_deadline + ))), ); let snapshot = serde_json::json!({ "phase": phase, "bucket": MIXED_BUCKET, "key": failure.key, @@ -579,6 +598,7 @@ async fn exercise_mixed_cluster( ) -> TestResult { let clients = cluster.create_all_clients()?; wait_for_upgrade_write_readiness(&clients, MIXED_BUCKET, phase, LISTING_CONVERGENCE_TIMEOUT).await?; + wait_for_drive_readmission(&clients, MIXED_BUCKET, phase, DRIVE_READMISSION_TIMEOUT).await?; let current_client = &clients[current_node]; let previous_client = &clients[previous_node]; @@ -656,16 +676,51 @@ pub(crate) async fn wait_for_upgrade_write_readiness( // health also accepts Returning disks whose write health is still FAULTY. // Probe every writer outside the asserted phase prefix; compatibility // writes still execute once and retain their original assertions. + wait_for_upgrade_writes(clients, bucket, phase, "write readiness", None, budget).await +} + +/// Block until every node commits a parity-0 write, which needs all of the +/// node's own drive handles. +/// +/// Write readiness proves that a node accepts a standard write at quorum; it +/// does not prove that the node has re-admitted the drives it marked Offline +/// while a peer restarted. A writer that still excludes one drive commits at +/// exactly three of four drives and cannot ride out any other slow drive. The +/// cluster must run with [`ALL_DRIVES_STORAGE_CLASS_ENV`]; a write with the +/// reduced-redundancy class then fails with 503 `ServiceUnavailable` until the +/// serving node writes through every drive, on the previous release and on +/// the current build alike. +async fn wait_for_drive_readmission(clients: &[Client], bucket: &str, context: &str, budget: Duration) -> TestResult { + wait_for_upgrade_writes( + clients, + bucket, + context, + "drive re-admission", + Some(StorageClass::ReducedRedundancy), + budget, + ) + .await +} + +async fn wait_for_upgrade_writes( + clients: &[Client], + bucket: &str, + phase: &str, + probe: &str, + storage_class: Option, + budget: Duration, +) -> TestResult { let deadline = Instant::now() + budget; for (node, client) in clients.iter().enumerate() { let client = upgrade_probe_client(client); - let key = format!(".upgrade-readiness/{phase}/node-{node}"); + let key = match storage_class { + Some(_) => format!(".upgrade-readiness/{phase}/all-drives/node-{node}"), + None => format!(".upgrade-readiness/{phase}/node-{node}"), + }; let mut last_response = "no response".to_string(); loop { if Instant::now() >= deadline { - return Err( - format!("{phase}: node {node} write readiness deadline exceeded; last response: {last_response}").into(), - ); + return Err(format!("{phase}: node {node} {probe} deadline exceeded; last response: {last_response}").into()); } let response = tokio::time::timeout_at( deadline, @@ -673,6 +728,7 @@ pub(crate) async fn wait_for_upgrade_write_readiness( .put_object() .bucket(bucket) .key(&key) + .set_storage_class(storage_class.clone()) .body(ByteStream::from_static(UPGRADE_READINESS_BODY)) .send(), ) @@ -683,13 +739,13 @@ pub(crate) async fn wait_for_upgrade_write_readiness( if error.raw_response().map(|response| response.status().as_u16()) != Some(503) || error.as_service_error().and_then(ProvideErrorMetadata::code) != Some("ServiceUnavailable") { - return Err(format!("{phase}: node {node} write readiness failed: {error:?}").into()); + return Err(format!("{phase}: node {node} {probe} failed: {error:?}").into()); } last_response = format!("{error:?}"); } Err(_) => { return Err(format!( - "{phase}: node {node} write readiness deadline exceeded during PutObject; last response: {last_response}" + "{phase}: node {node} {probe} deadline exceeded during PutObject; last response: {last_response}" ) .into()); } @@ -707,16 +763,23 @@ async fn prepare_previous_release_baseline(cluster: &mut RustFSTestClusterEnviro // persist data, then restart each old process once with three peers still // readable. This preparation ends before any current binary is started. wait_for_upgrade_write_readiness(&clients[..1], MIXED_BUCKET, "previous-seed", LISTING_CONVERGENCE_TIMEOUT).await?; + // A node that just came back is still replaying its startup traffic against + // every peer. Killing the next node during that burst gives the fresh node + // two failures in a row and it drops that peer's drive to Offline, where + // only the slow recovery probe can bring it back. Let every node re-admit + // every drive before the next kill and before any current binary starts. for node in [1, 2, 3, 0] { cluster.stop_node(node)?; cluster.start_node_from_binary(node, previous_binary).await?; + let context = format!("previous-restart-{node}"); wait_for_upgrade_write_readiness( std::slice::from_ref(&clients[node]), MIXED_BUCKET, - &format!("previous-restart-{node}"), + &context, LISTING_CONVERGENCE_TIMEOUT, ) .await?; + wait_for_drive_readmission(&clients, MIXED_BUCKET, &context, DRIVE_READMISSION_TIMEOUT).await?; } wait_for_upgrade_write_readiness(&clients, MIXED_BUCKET, "previous-baseline", LISTING_CONVERGENCE_TIMEOUT).await?; @@ -738,6 +801,52 @@ async fn prepare_previous_release_baseline(cluster: &mut RustFSTestClusterEnviro Ok(()) } +/// The drive re-admission gate assumes that a parity-0 write needs every +/// drive. Prove it on the workspace build: with one node down a standard +/// write still commits, the reduced-redundancy write fails with 503, and once +/// the node is back the gate converges within its budget. +#[tokio::test] +async fn parity_zero_write_needs_every_drive() -> TestResult { + init_logging(); + let mut cluster = RustFSTestClusterEnvironment::new(MIXED_NODE_COUNT).await?; + cluster.set_env("RUST_LOG", "rustfs=warn,rustfs_notify=warn"); + cluster.set_env(ALL_DRIVES_STORAGE_CLASS_ENV.0, ALL_DRIVES_STORAGE_CLASS_ENV.1); + cluster.start().await?; + cluster.create_test_bucket(MIXED_BUCKET).await?; + let clients = cluster.create_all_clients()?; + wait_for_upgrade_write_readiness(&clients, MIXED_BUCKET, "parity-zero-seed", LISTING_CONVERGENCE_TIMEOUT).await?; + wait_for_drive_readmission(&clients, MIXED_BUCKET, "parity-zero-healthy", DRIVE_READMISSION_TIMEOUT).await?; + + cluster.stop_node(3)?; + let writer = upgrade_probe_client(&clients[0]); + writer + .put_object() + .bucket(MIXED_BUCKET) + .key("parity-zero/standard-with-one-node-down") + .body(ByteStream::from_static(UPGRADE_READINESS_BODY)) + .send() + .await?; + let error = writer + .put_object() + .bucket(MIXED_BUCKET) + .key("parity-zero/all-drives-with-one-node-down") + .storage_class(StorageClass::ReducedRedundancy) + .body(ByteStream::from_static(UPGRADE_READINESS_BODY)) + .send() + .await + .expect_err("a parity-0 write must not commit at three of four drives"); + assert_eq!(error.raw_response().map(|response| response.status().as_u16()), Some(503), "{error:?}"); + assert_eq!( + error.as_service_error().and_then(ProvideErrorMetadata::code), + Some("ServiceUnavailable"), + "{error:?}" + ); + + cluster.start_node(3).await?; + wait_for_drive_readmission(&clients, MIXED_BUCKET, "parity-zero-returned", DRIVE_READMISSION_TIMEOUT).await?; + Ok(()) +} + #[cfg(test)] mod upgrade_write_readiness_tests { use super::*; @@ -963,7 +1072,14 @@ mod upgrade_write_readiness_tests { ); assert_eq!(target.count_requests(FakeTargetOperation::HeadObject, stalled_key), MIXED_NODE_COUNT); assert_eq!(target.count_requests(FakeTargetOperation::CompleteMultipartUpload, stalled_key), 1); - assert_eq!(snapshot["top_locks"]["observation"], "unknown", "the fake peer has no admin lock API"); + let locks = snapshot["top_locks"] + .as_array() + .expect("failure artifact must retain every node's lock probe"); + assert_eq!(locks.len(), MIXED_NODE_COUNT); + for (node, probe) in locks.iter().enumerate() { + assert_eq!(probe["node"], node); + assert_eq!(probe["observation"], "unknown", "the fake peer has no admin lock API"); + } let locks: MixedLockSnapshot = serde_json::from_value(serde_json::json!({ "total": 1, "truncated": true, "capability_note": "partial snapshot", @@ -1239,6 +1355,7 @@ async fn rolling_upgrade_from_rc2_preserves_mixed_version_contracts() -> TestRes let current_binary = rustfs_binary_path(); let mut cluster = RustFSTestClusterEnvironment::new(MIXED_NODE_COUNT).await?; cluster.set_env("RUST_LOG", "rustfs=warn,rustfs_notify=warn"); + cluster.set_env(ALL_DRIVES_STORAGE_CLASS_ENV.0, ALL_DRIVES_STORAGE_CLASS_ENV.1); configure_cluster_logs(&mut cluster)?; cluster.start_with_binary(&previous_binary).await?; cluster.create_test_bucket(MIXED_BUCKET).await?; @@ -1256,8 +1373,10 @@ async fn rolling_upgrade_from_rc2_preserves_mixed_version_contracts() -> TestRes cluster.stop_node(3)?; cluster.start_node_from_binary(3, ¤t_binary).await?; + let clients = cluster.create_all_clients()?; + wait_for_drive_readmission(&clients, MIXED_BUCKET, "homogeneous-current", DRIVE_READMISSION_TIMEOUT).await?; - for (node_idx, client) in cluster.create_all_clients()?.iter().enumerate() { + for (node_idx, client) in clients.iter().enumerate() { assert_eq!( read_object(client, MIXED_BUCKET, PREVIOUS_RELEASE_SEED_KEY, None).await?.1, UPGRADE_READINESS_BODY,