diff --git a/Cargo.lock b/Cargo.lock index 637202fd5..b7b1ec220 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -12688,6 +12688,7 @@ dependencies = [ "js-sys", "rand 0.10.2", "serde_core", + "sha1_smol", "wasm-bindgen", ] diff --git a/rustfs/Cargo.toml b/rustfs/Cargo.toml index 28f4ad259..af5514870 100644 --- a/rustfs/Cargo.toml +++ b/rustfs/Cargo.toml @@ -322,7 +322,7 @@ thiserror = { workspace = true } tracing.workspace = true url = { workspace = true } urlencoding = { workspace = true } -uuid = { workspace = true, features = ["v4", "fast-rng", "macro-diagnostics"] } +uuid = { workspace = true, features = ["v4", "v5", "fast-rng", "macro-diagnostics"] } zip = { workspace = true } libc = { workspace = true } rand = { workspace = true, features = ["serde"] } @@ -345,7 +345,7 @@ libsystemd.workspace = true libmimalloc-sys.workspace = true [dev-dependencies] -uuid = { workspace = true, features = ["v4", "fast-rng", "macro-diagnostics"] } +uuid = { workspace = true, features = ["v4", "v5", "fast-rng", "macro-diagnostics"] } serial_test = { workspace = true } tempfile = { workspace = true } aws-config = { workspace = true } diff --git a/rustfs/src/admin/handlers/site_replication.rs b/rustfs/src/admin/handlers/site_replication.rs index 1bb0379df..23c8b680b 100644 --- a/rustfs/src/admin/handlers/site_replication.rs +++ b/rustfs/src/admin/handlers/site_replication.rs @@ -41,7 +41,7 @@ use crate::admin::storage_api::config::save_admin_config; use crate::admin::storage_api::contract::bucket::{ BucketOperations, BucketOptions, DeleteBucketOptions, MakeBucketOptions, SRBucketDeleteOp, }; -use crate::admin::storage_api::error::Error as StorageError; +use crate::admin::storage_api::error::{Error as StorageError, is_err_bucket_not_found}; use crate::admin::storage_api::runtime::ECStore; use crate::admin::utils::{encode_compatible_admin_payload, read_compatible_admin_body}; use crate::auth::constant_time_eq; @@ -55,6 +55,7 @@ use crate::storage::storage_api::{ use base64::Engine; use base64::engine::general_purpose::STANDARD as BASE64_STANDARD; use base64::engine::general_purpose::URL_SAFE_NO_PAD; +use futures::StreamExt; use hmac::{Hmac, Mac}; use http::header::{CONTENT_TYPE, HOST}; use http::{HeaderMap, HeaderValue, Uri}; @@ -2096,6 +2097,18 @@ async fn remote_add_preflight_info(site: &PeerSite) -> S3Result Option { query_pairs(uri).get("bootstrapToken").cloned() } -fn bootstrap_bucket_make_op_path(bucket: &SRBucketInfo) -> String { +/// Query for a peer `make-with-versioning` bucket op. `versioningEnabled` +/// always travels so the outbound query matches MinIO's site-replication +/// make-bucket wire contract: MinIO's own create-bucket hook sends +/// `versioningEnabled=true` on this op. RustFS's inbound handler +/// force-enables versioning either way. +fn make_with_versioning_bucket_op_path(bucket: &str, created_at: Option<&str>, lock_enabled: bool) -> String { let mut query = form_urlencoded::Serializer::new(String::new()); - query.append_pair("bucket", &bucket.bucket); - query.append_pair("operation", "make-with-versioning"); - if let Some(created_at) = bucket - .created_at - .and_then(|value| value.format(&time::format_description::well_known::Rfc3339).ok()) - { - query.append_pair("createdAt", &created_at); + query.append_pair("bucket", bucket); + query.append_pair("operation", SITE_REPLICATION_BUCKET_OP_MAKE_WITH_VERSIONING); + query.append_pair("versioningEnabled", "true"); + if let Some(created_at) = created_at { + query.append_pair("createdAt", created_at); } - if bucket.object_lock_config.is_some() { + if lock_enabled { query.append_pair("lockEnabled", "true"); } - format!("/rustfs/admin/v3/site-replication/peer/bucket-ops?{}", query.finish()) + format!("{SITE_REPLICATION_PEER_BUCKET_OPS_PATH}?{}", query.finish()) +} + +fn bootstrap_bucket_make_op_path(bucket: &SRBucketInfo) -> String { + let created_at = bucket + .created_at + .and_then(|value| value.format(&time::format_description::well_known::Rfc3339).ok()); + make_with_versioning_bucket_op_path(&bucket.bucket, created_at.as_deref(), bucket.object_lock_config.is_some()) } fn bootstrap_bucket_meta_item(bucket: &SRBucketInfo, item_type: &str, updated_at: Option) -> SRBucketMeta { @@ -4246,16 +4269,7 @@ async fn broadcast_site_replication_make_bucket( .format(&time::format_description::well_known::Rfc3339) .unwrap_or_default(); - let path = { - let mut query = form_urlencoded::Serializer::new(String::new()); - query.append_pair("bucket", bucket); - query.append_pair("operation", "make-with-versioning"); - query.append_pair("createdAt", &created_at); - if lock_enabled { - query.append_pair("lockEnabled", "true"); - } - format!("/rustfs/admin/v3/site-replication/peer/bucket-ops?{}", query.finish()) - }; + let path = make_with_versioning_bucket_op_path(bucket, Some(&created_at), lock_enabled); let path = if let Some(token) = bootstrap_token { with_site_replication_bootstrap_token(&path, token) } else { @@ -10206,13 +10220,25 @@ impl Operation for SiteReplicationStatusHandler { } } +/// `POST /v3/site-replication/devnull` — peer link-check upload drain. +/// MinIO streams multi-megabyte probe bodies here during site netperf link +/// checks and expects an unbounded discard (its handler copies to io.Discard); +/// buffering through the 1MB admin body cap turned any larger probe into a +/// 400 and a false link failure. Stream and discard instead — no size cap. +async fn drain_site_replication_devnull(mut input: Body) -> S3Result<()> { + while let Some(chunk) = input.next().await { + chunk.map_err(|e| s3_error!(InvalidRequest, "failed to read devnull stream: {}", e))?; + } + Ok(()) +} + pub struct SiteReplicationDevNullHandler {} #[async_trait::async_trait] impl Operation for SiteReplicationDevNullHandler { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { validate_site_replication_admin_request(&req, AdminAction::SiteReplicationOperationAction).await?; - let _ = read_plain_admin_body(req.input).await?; + drain_site_replication_devnull(req.input).await?; Ok(empty_response(StatusCode::NO_CONTENT)) } } @@ -10471,6 +10497,19 @@ impl Operation for SRPeerJoinHandler { } } +/// Outcome of a peer-driven `purge-deleted-bucket` replay. A bucket that is +/// already gone means the purge raced an earlier replay or a local delete — +/// that is success — but any other failure must reach the sender like the +/// sibling delete branches do: swallowing it answered 200 while the bucket +/// survived on this site. +fn purge_deleted_bucket_result(result: Result<(), StorageError>) -> S3Result<()> { + match result { + Ok(()) => Ok(()), + Err(err) if is_err_bucket_not_found(&err) => Ok(()), + Err(err) => Err(ApiError::from(err).into()), + } +} + pub struct SRPeerBucketOpsHandler {} #[async_trait::async_trait] @@ -10570,16 +10609,18 @@ impl Operation for SRPeerBucketOpsHandler { .map_err(ApiError::from)?; } "purge-deleted-bucket" => { - let _ = store - .delete_bucket( - &bucket, - &DeleteBucketOptions { - force: true, - srdelete_op: SRBucketDeleteOp::Purge, - ..Default::default() - }, - ) - .await; + purge_deleted_bucket_result( + store + .delete_bucket( + &bucket, + &DeleteBucketOptions { + force: true, + srdelete_op: SRBucketDeleteOp::Purge, + ..Default::default() + }, + ) + .await, + )?; } _ => return Err(s3_error!(InvalidRequest, "unsupported site replication bucket operation")), } @@ -13925,6 +13966,54 @@ mod tests { assert!(!query_flag(&uri, "missing")); } + /// A5 red-light: a `purge-deleted-bucket` replay must report success when + /// the bucket is already gone, and must propagate every other failure — + /// the swallowed error answered 200 while the bucket survived. + #[test] + fn test_purge_deleted_bucket_result_tolerates_only_missing_bucket() { + assert!(purge_deleted_bucket_result(Ok(())).is_ok()); + assert!(purge_deleted_bucket_result(Err(StorageError::BucketNotFound("photos".to_string()))).is_ok()); + assert!(purge_deleted_bucket_result(Err(StorageError::VolumeNotFound)).is_ok()); + let err = purge_deleted_bucket_result(Err(StorageError::StorageFull)) + .expect_err("non-not-found delete failures must propagate"); + assert_ne!(*err.code(), S3ErrorCode::NoSuchBucket); + } + + /// C5 red-light: the site-replication devnull drain must accept bodies + /// beyond the 1MB admin body cap — MinIO's link check streams large + /// probe bodies and treats a 400 as a broken link. + #[tokio::test] + async fn test_site_replication_devnull_drains_body_beyond_admin_cap() { + let body = Body::from(vec![0u8; MAX_ADMIN_REQUEST_BODY_SIZE + 1]); + drain_site_replication_devnull(body) + .await + .expect("devnull must drain bodies larger than the admin body cap"); + } + + /// A3 red-light: `versioningEnabled` must travel on every outbound + /// make-with-versioning bucket op so the query matches MinIO's + /// site-replication make-bucket wire contract (MinIO's own hook sends + /// `versioningEnabled=true` on this op). + #[test] + fn test_make_with_versioning_op_paths_send_versioning_enabled() { + let bucket = SRBucketInfo { + bucket: "photos".to_string(), + created_at: Some(OffsetDateTime::UNIX_EPOCH), + object_lock_config: Some(BASE64_STANDARD.encode("")), + ..Default::default() + }; + let bootstrap = bootstrap_bucket_make_op_path(&bucket); + assert!(bootstrap.contains("operation=make-with-versioning"), "{bootstrap}"); + assert!(bootstrap.contains("versioningEnabled=true"), "{bootstrap}"); + assert!(bootstrap.contains("createdAt="), "{bootstrap}"); + assert!(bootstrap.contains("lockEnabled=true"), "{bootstrap}"); + + // The broadcast path (create-bucket hook) shares the same builder. + let broadcast = make_with_versioning_bucket_op_path("photos", Some("1970-01-01T00:00:00Z"), false); + assert!(broadcast.contains("versioningEnabled=true"), "{broadcast}"); + assert!(!broadcast.contains("lockEnabled"), "{broadcast}"); + } + #[tokio::test] #[serial] async fn test_add_bootstrap_scope_only_allows_expected_bucket_setup_until_guard_drops() { diff --git a/rustfs/src/admin/site_replication_identity.rs b/rustfs/src/admin/site_replication_identity.rs index dc6440b4d..24784160b 100644 --- a/rustfs/src/admin/site_replication_identity.rs +++ b/rustfs/src/admin/site_replication_identity.rs @@ -13,9 +13,9 @@ // limitations under the License. use rustfs_madmin::{PeerInfo, SyncStatus}; -use std::collections::{BTreeMap, hash_map::DefaultHasher}; -use std::hash::{Hash, Hasher}; +use std::collections::BTreeMap; use url::Url; +use uuid::Uuid; fn has_http_scheme(endpoint: &str) -> bool { endpoint.get(..7).is_some_and(|prefix| prefix.eq_ignore_ascii_case("http://")) @@ -66,10 +66,12 @@ pub fn site_identity_key(endpoint: &str) -> String { .unwrap_or_else(|| trimmed.to_ascii_lowercase()) } +/// Fallback deployment ID for a peer that reported none. UUIDv5 over the +/// canonical endpoint: the ID is persisted in site-replication state and +/// broadcast to peers, so it must be identical across Rust toolchains +/// (`DefaultHasher` is not) and across spellings of the same endpoint. pub fn deployment_id_for_endpoint(endpoint: &str) -> String { - let mut hasher = DefaultHasher::new(); - endpoint.hash(&mut hasher); - format!("{:016x}", hasher.finish()) + Uuid::new_v5(&Uuid::NAMESPACE_URL, canonical_endpoint(endpoint).as_bytes()).to_string() } pub fn same_identity_endpoint(left: &str, right: &str) -> bool { @@ -174,6 +176,23 @@ mod tests { } } + /// B8 red-light: the fallback deployment ID must be a toolchain-stable + /// UUIDv5 over the canonical endpoint — `DefaultHasher` output is not + /// guaranteed stable across Rust releases, yet the ID is persisted in + /// site-replication state and broadcast to peers. + #[test] + fn deployment_id_for_endpoint_is_stable_uuid_v5_over_canonical_endpoint() { + let endpoint = "https://node-a.example.com:9000"; + let id = deployment_id_for_endpoint(endpoint); + let parsed = uuid::Uuid::parse_str(&id).expect("fallback deployment ID must be a UUID"); + assert_eq!(parsed.get_version_num(), 5, "fallback deployment ID must be UUIDv5"); + // Deterministic for the same endpoint and for spelling variants that + // share a canonical form; distinct endpoints stay distinct. + assert_eq!(id, deployment_id_for_endpoint(endpoint)); + assert_eq!(id, deployment_id_for_endpoint(" HTTPS://Node-A.Example.Com:9000/ ")); + assert_ne!(id, deployment_id_for_endpoint("https://node-b.example.com:9000")); + } + #[test] fn canonical_endpoint_accepts_case_insensitive_scheme() { assert_eq!( diff --git a/rustfs/src/admin/storage_api.rs b/rustfs/src/admin/storage_api.rs index 318718b49..bdf28f15e 100644 --- a/rustfs/src/admin/storage_api.rs +++ b/rustfs/src/admin/storage_api.rs @@ -51,7 +51,7 @@ mod ecstore_disk { } mod ecstore_error { - pub(crate) use crate::storage::storage_api::ecstore_error::StorageError; + pub(crate) use crate::storage::storage_api::ecstore_error::{StorageError, is_err_bucket_not_found}; } #[allow(unused_imports)] @@ -919,6 +919,7 @@ pub(crate) mod contract { } pub(crate) mod error { + pub(crate) use super::ecstore_error::is_err_bucket_not_found; pub(crate) use super::{Error, StorageError}; }