From 2518fb5cd7fcc956744ac43f665c80ac39a479cc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=94=90=E5=B0=8F=E9=B8=AD?= Date: Sat, 5 Sep 2026 15:19:46 +0800 Subject: [PATCH] fix(site-replication): broadcast bucket ops to every peer and record each failure (backlog#2293) The generic JSON broadcast (make/delete bucket, bucket-meta hook, bucket ops) returned at the first failing peer, so peers later in deployment-id order never received the request and got no retry event; a transport construction failure recorded nothing at all. Attempt every remote peer like the IAM change hook does: a success settles the peer/path retry event, a failure (transport construction included) enqueues one under the request path, and the first error is returned after all peers were attempted. (cherry picked from commit ce8f73bfd74bac61c383434780f27d17ca16d75e) --- rustfs/src/site_replication/tests.rs | 127 +++++++++++++++++++++++ rustfs/src/site_replication/transport.rs | 30 ++++-- 2 files changed, 151 insertions(+), 6 deletions(-) diff --git a/rustfs/src/site_replication/tests.rs b/rustfs/src/site_replication/tests.rs index a9491e1b9..9960bbdec 100644 --- a/rustfs/src/site_replication/tests.rs +++ b/rustfs/src/site_replication/tests.rs @@ -3367,3 +3367,130 @@ fn test_retry_snapshot_tombstones_removed_service_accounts() { assert!(change.create.is_none()); assert_eq!(tombstone.updated_at, Some(observed_at)); } + +/// Spawns a one-shot HTTP peer that answers 200 and flips the returned flag +/// once a request head has arrived. +async fn spawn_reached_probe_peer() -> (String, Arc, tokio::task::JoinHandle<()>) { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind healthy peer"); + let endpoint = format!("http://{}", listener.local_addr().expect("healthy peer address")); + let reached = Arc::new(AtomicBool::new(false)); + let reached_by_server = reached.clone(); + let server = tokio::spawn(async move { + let Ok((mut stream, _)) = listener.accept().await else { + return; + }; + let mut request = Vec::new(); + let mut buffer = [0_u8; 1024]; + loop { + let Ok(read) = stream.read(&mut buffer).await else { + return; + }; + if read == 0 { + return; + } + request.extend_from_slice(&buffer[..read]); + if request.windows(4).any(|window| window == b"\r\n\r\n") { + break; + } + } + reached_by_server.store(true, Ordering::SeqCst); + let _ = stream + .write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 2\r\nconnection: close\r\n\r\nok") + .await; + }); + (endpoint, reached, server) +} + +/// Three-peer runtime whose local peer is `local`; BTreeMap order visits the +/// failing peer `b` before the healthy peer `c`. +fn broadcast_runtime_with_failing_peer_before_healthy(failing_endpoint: &str, healthy_endpoint: &str) -> SiteReplicationRuntime { + let local_peer = PeerInfo { + deployment_id: "local".to_string(), + ..peer("local", "http://127.0.0.1:9") + }; + let mut state = SiteReplicationState { + name: "local".to_string(), + service_account_access_key: "site-replicator-0".to_string(), + ..Default::default() + }; + state.peers.insert("local".to_string(), local_peer.clone()); + state.peers.insert( + "b".to_string(), + PeerInfo { + deployment_id: "b".to_string(), + ..peer("b", failing_endpoint) + }, + ); + state.peers.insert( + "c".to_string(), + PeerInfo { + deployment_id: "c".to_string(), + ..peer("c", healthy_endpoint) + }, + ); + SiteReplicationRuntime { + state, + local_peer, + service_account_secret_key: "site-replicator-secret".to_string(), + } +} + +const BROADCAST_PROBE_DELETE_BUCKET_PATH: &str = + "/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=photos&operation=delete-bucket"; + +/// The generic JSON broadcast (bucket make/delete, bucket-meta hook, bucket +/// ops) attempts every remote peer: a peer whose request fails must not stop +/// delivery to the peers that follow it in deployment-id order, and the +/// failure is still reported to the caller (backlog#2293). +#[tokio::test] +#[serial] +async fn test_broadcast_json_reaches_healthy_peers_after_a_failed_peer() { + // Peer "b": nothing listens on the port, so the connect is refused. + let refused = TcpListener::bind("127.0.0.1:0").await.expect("bind refused-peer probe"); + let refused_endpoint = format!("http://{}", refused.local_addr().expect("refused-peer address")); + drop(refused); + + let (healthy_endpoint, reached, server) = spawn_reached_probe_peer().await; + let runtime = broadcast_runtime_with_failing_peer_before_healthy(&refused_endpoint, &healthy_endpoint); + + let result = temp_env::async_with_vars([(ALLOW_LOOPBACK_REPLICATION_TARGET_ENV, Some("true"))], async { + broadcast_site_replication_json_with_runtime(&runtime, BROADCAST_PROBE_DELETE_BUCKET_PATH, &serde_json::json!({})).await + }) + .await; + + let err = result.expect_err("peer b refuses connections, the broadcast must report it"); + assert!( + reached.load(Ordering::SeqCst), + "peer c never received the broadcast once peer b failed: {err}" + ); + server.abort(); +} + +/// Same guarantee when the failing peer never gets a transport: an endpoint +/// that `PeerTransport::for_runtime_peer` rejects must be skipped past (and +/// reported), not abort the broadcast before the healthy peers (backlog#2293). +#[tokio::test] +#[serial] +async fn test_broadcast_json_reaches_healthy_peers_after_a_peer_without_transport() { + // Peer "b": a scheme the peer connection validator refuses outright. + let forbidden_endpoint = "ftp://peer-b.example.com"; + + let (healthy_endpoint, reached, server) = spawn_reached_probe_peer().await; + let runtime = broadcast_runtime_with_failing_peer_before_healthy(forbidden_endpoint, &healthy_endpoint); + + let result = temp_env::async_with_vars([(ALLOW_LOOPBACK_REPLICATION_TARGET_ENV, Some("true"))], async { + broadcast_site_replication_json_with_runtime(&runtime, BROADCAST_PROBE_DELETE_BUCKET_PATH, &serde_json::json!({})).await + }) + .await; + + let err = result.expect_err("peer b has no usable transport, the broadcast must report it"); + assert!( + err.to_string().contains("invalid persisted site replication peer"), + "the reported error must be peer b's transport failure: {err}" + ); + assert!( + reached.load(Ordering::SeqCst), + "peer c never received the broadcast once peer b failed to get a transport: {err}" + ); + server.abort(); +} diff --git a/rustfs/src/site_replication/transport.rs b/rustfs/src/site_replication/transport.rs index bc941c9f0..18d4850b9 100644 --- a/rustfs/src/site_replication/transport.rs +++ b/rustfs/src/site_replication/transport.rs @@ -876,6 +876,14 @@ pub(crate) async fn broadcast_site_replication_json(path: &str, bo broadcast_site_replication_json_with_runtime(&runtime, path, body).await } +/// PUT `body` to `path` on every remote peer of the runtime. +/// +/// Every peer is attempted: one peer's failure — transport construction +/// included — must not skip the peers that follow it in deployment-id order, +/// or they silently miss the change with no retry record (backlog#2293). A +/// success settles the peer/path's queued retry event, a failure enqueues one +/// under the request `path` (so the drain classifies it as today), and the +/// first error is returned once all peers were attempted. pub(crate) async fn broadcast_site_replication_json_with_runtime( runtime: &SiteReplicationRuntime, path: &str, @@ -883,20 +891,30 @@ pub(crate) async fn broadcast_site_replication_json_with_runtime( ) -> S3Result<()> { let state = &runtime.state; let local_peer = &runtime.local_peer; + let mut first_error: Option = None; for peer in state.peers.values() { if peer.deployment_id == local_peer.deployment_id || same_identity_endpoint(&peer.endpoint, &local_peer.endpoint) { continue; } - let transport = PeerTransport::for_runtime_peer(peer).await?; - PeerAdminRequest::put(&transport.connection, path, &state.service_account_access_key) - .with_client(&transport.client) - .send_with_retry_event(peer, &runtime.service_account_secret_key, body) - .await?; + let sent = match PeerTransport::for_runtime_peer(peer).await { + Ok(transport) => PeerAdminRequest::put(&transport.connection, path, &state.service_account_access_key) + .with_client(&transport.client) + .send_with_retry_event(peer, &runtime.service_account_secret_key, body) + .await + .map(|_| ()), + Err(err) => { + enqueue_site_replication_retry_event(peer, path, &err).await; + Err(err) + } + }; + if let Err(err) = sent { + first_error.get_or_insert(err); + } } - Ok(()) + first_error.map_or(Ok(()), Err) } pub(crate) fn parse_endpoint_refresh_status(peer: &PeerInfo, body: &[u8]) -> S3Result<()> {