mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-19 02:56:18 +00:00
Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 唐小鸭
|
1937a0152c |
@@ -83,6 +83,16 @@ pub struct SiteReplicationInfo {
|
||||
pub service_account_access_key: String,
|
||||
#[serde(rename = "apiVersion", skip_serializing_if = "Option::is_none")]
|
||||
pub api_version: Option<String>,
|
||||
/// Outstanding peer deliveries. Absent when the retry queue is empty, so a
|
||||
/// healthy site serializes exactly as it did before this field existed.
|
||||
/// Present means peer operations are failing even if `enabled` is true.
|
||||
#[serde(rename = "retryStats", default, skip_serializing_if = "Option::is_none")]
|
||||
pub retry_stats: Option<SRRetryStats>,
|
||||
/// A multi-step lifecycle operation this site has not finished — most
|
||||
/// importantly a removal that could not reach its peers, which makes the
|
||||
/// site reject peer operations while `enabled` may still read true.
|
||||
#[serde(rename = "pendingOperation", default, skip_serializing_if = "Option::is_none")]
|
||||
pub pending_operation: Option<SRPendingOperation>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
|
||||
|
||||
@@ -14,7 +14,7 @@
|
||||
|
||||
use crate::admin::storage_api::cluster::{CapabilityState, CapabilityStatus, ObservabilitySnapshot, TopologySnapshot};
|
||||
use crate::admin::{
|
||||
auth::authorize_admin_request,
|
||||
auth::validate_admin_request,
|
||||
router::{AdminOperation, Operation, S3Router},
|
||||
runtime_sources::default_admin_usecase,
|
||||
storage_api::cluster::{
|
||||
@@ -24,10 +24,11 @@ use crate::admin::{
|
||||
},
|
||||
system,
|
||||
};
|
||||
use crate::auth::{check_key_valid, get_session_token};
|
||||
use crate::cluster_snapshot::{
|
||||
ClusterReadOnlySnapshot, ClusterRuntimeReadinessState, ClusterRuntimeStatusSnapshot, cluster_has_actionable_pressure,
|
||||
};
|
||||
use crate::server::{ADMIN_PREFIX, ReadinessDegradedReason};
|
||||
use crate::server::{ADMIN_PREFIX, ReadinessDegradedReason, RemoteAddr};
|
||||
use http::{HeaderMap, HeaderValue, StatusCode};
|
||||
use hyper::Method;
|
||||
use matchit::Params;
|
||||
@@ -65,15 +66,23 @@ pub(crate) struct ClusterSnapshotDiscoveryResponse {
|
||||
pub components: Option<ClusterComponentStatusView>,
|
||||
}
|
||||
|
||||
/// The pre-check keeps this endpoint's historical missing-credentials message;
|
||||
/// the shared gate reports "get cred failed".
|
||||
async fn authorize_cluster_snapshot_request(req: &S3Request<Body>) -> S3Result<()> {
|
||||
if req.credentials.is_none() {
|
||||
let Some(input_cred) = &req.credentials else {
|
||||
return Err(s3_error!(InvalidRequest, "authentication required"));
|
||||
}
|
||||
};
|
||||
|
||||
authorize_admin_request(req, vec![Action::AdminAction(AdminAction::ServerInfoAdminAction)]).await?;
|
||||
Ok(())
|
||||
let (cred, owner) =
|
||||
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
|
||||
|
||||
validate_admin_request(
|
||||
&req.headers,
|
||||
&cred,
|
||||
owner,
|
||||
false,
|
||||
vec![Action::AdminAction(AdminAction::ServerInfoAdminAction)],
|
||||
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
fn build_json_response(
|
||||
@@ -944,30 +953,6 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
/// This endpoint authorizes through the shared admin gate, which reports
|
||||
/// "get cred failed" for a credential-less request. The pre-check keeps the
|
||||
/// message it has always returned (rustfs/backlog#1829).
|
||||
#[tokio::test]
|
||||
async fn cluster_snapshot_gate_keeps_its_missing_credentials_message() {
|
||||
let req = s3s::S3Request {
|
||||
input: s3s::Body::from(String::new()),
|
||||
method: http::Method::GET,
|
||||
uri: http::Uri::from_static("/rustfs/admin/v4/cluster/snapshot"),
|
||||
headers: http::HeaderMap::new(),
|
||||
extensions: http::Extensions::new(),
|
||||
credentials: None,
|
||||
region: None,
|
||||
service: None,
|
||||
trailing_headers: None,
|
||||
};
|
||||
|
||||
let err = super::authorize_cluster_snapshot_request(&req)
|
||||
.await
|
||||
.expect_err("a request without credentials must be rejected");
|
||||
assert_eq!(err.code(), &s3s::S3ErrorCode::InvalidRequest);
|
||||
assert_eq!(err.message(), Some("authentication required"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn cluster_snapshot_response_serializes_none_snapshot() {
|
||||
let value = serde_json::to_value(ClusterSnapshotResponse { snapshot: None }).expect("serialize response");
|
||||
|
||||
@@ -14,7 +14,7 @@
|
||||
|
||||
use crate::admin::storage_api::cluster::CapabilityStatus;
|
||||
use crate::admin::{
|
||||
auth::authorize_admin_request,
|
||||
auth::validate_admin_request,
|
||||
handlers::{cluster_snapshot, plugins_instances, system},
|
||||
plugin_contract::{
|
||||
PluginContractDomain, PluginInstanceDiagnosticCode, PluginInstanceDiagnosticCount, PluginInstanceEntry,
|
||||
@@ -22,7 +22,8 @@ use crate::admin::{
|
||||
},
|
||||
router::{AdminOperation, Operation, S3Router},
|
||||
};
|
||||
use crate::server::ADMIN_PREFIX;
|
||||
use crate::auth::{check_key_valid, get_session_token};
|
||||
use crate::server::{ADMIN_PREFIX, RemoteAddr};
|
||||
use http::{HeaderMap, HeaderValue, StatusCode};
|
||||
use hyper::Method;
|
||||
use matchit::Params;
|
||||
@@ -182,26 +183,42 @@ fn map_extension_instance(instance: PluginInstanceEntry) -> ExtensionInstanceEnt
|
||||
}
|
||||
}
|
||||
|
||||
/// The pre-check keeps this endpoint's historical missing-credentials message;
|
||||
/// the shared gate reports "get cred failed".
|
||||
async fn authorize_extension_catalog_request(req: &S3Request<Body>) -> S3Result<()> {
|
||||
if req.credentials.is_none() {
|
||||
let Some(input_cred) = &req.credentials else {
|
||||
return Err(s3_error!(InvalidRequest, "authentication required"));
|
||||
}
|
||||
};
|
||||
|
||||
authorize_admin_request(req, vec![Action::AdminAction(AdminAction::ServerInfoAdminAction)]).await?;
|
||||
Ok(())
|
||||
let (cred, owner) =
|
||||
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
|
||||
|
||||
validate_admin_request(
|
||||
&req.headers,
|
||||
&cred,
|
||||
owner,
|
||||
false,
|
||||
vec![Action::AdminAction(AdminAction::ServerInfoAdminAction)],
|
||||
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
/// The pre-check keeps this endpoint's historical missing-credentials message;
|
||||
/// the shared gate reports "get cred failed".
|
||||
async fn authorize_extension_instance_request(req: &S3Request<Body>) -> S3Result<()> {
|
||||
if req.credentials.is_none() {
|
||||
let Some(input_cred) = &req.credentials else {
|
||||
return Err(s3_error!(InvalidRequest, "authentication required"));
|
||||
}
|
||||
};
|
||||
|
||||
authorize_admin_request(req, vec![Action::AdminAction(AdminAction::GetBucketTargetAction)]).await?;
|
||||
Ok(())
|
||||
let (cred, owner) =
|
||||
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
|
||||
|
||||
validate_admin_request(
|
||||
&req.headers,
|
||||
&cred,
|
||||
owner,
|
||||
false,
|
||||
vec![Action::AdminAction(AdminAction::GetBucketTargetAction)],
|
||||
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
fn build_json_response(
|
||||
@@ -303,36 +320,6 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
/// Both extension gates authorize through the shared admin gate, which reports
|
||||
/// "get cred failed" for a credential-less request. The pre-check keeps the
|
||||
/// message these endpoints have always returned (rustfs/backlog#1829).
|
||||
#[tokio::test]
|
||||
async fn extension_gates_keep_their_missing_credentials_message() {
|
||||
let credential_less_request = || s3s::S3Request {
|
||||
input: s3s::Body::from(String::new()),
|
||||
method: http::Method::GET,
|
||||
uri: http::Uri::from_static("/rustfs/admin/v4/extensions/catalog"),
|
||||
headers: http::HeaderMap::new(),
|
||||
extensions: http::Extensions::new(),
|
||||
credentials: None,
|
||||
region: None,
|
||||
service: None,
|
||||
trailing_headers: None,
|
||||
};
|
||||
|
||||
for err in [
|
||||
super::authorize_extension_catalog_request(&credential_less_request())
|
||||
.await
|
||||
.expect_err("a request without credentials must be rejected"),
|
||||
super::authorize_extension_instance_request(&credential_less_request())
|
||||
.await
|
||||
.expect_err("a request without credentials must be rejected"),
|
||||
] {
|
||||
assert_eq!(err.code(), &s3s::S3ErrorCode::InvalidRequest);
|
||||
assert_eq!(err.message(), Some("authentication required"));
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn builtin_ops_schemas_register_cleanly_in_runtime_registries() {
|
||||
let mut diagnostics_registry = rustfs_targets::OpsDiagnosticsRegistry::new();
|
||||
|
||||
@@ -21,11 +21,12 @@
|
||||
//! that bucket, and with `bucket`+`object` it flushes that one identity — the
|
||||
//! only remediation for a poisoned entry short of a node restart.
|
||||
|
||||
use crate::admin::auth::authorize_admin_request;
|
||||
use crate::admin::auth::validate_admin_request;
|
||||
use crate::admin::router::{AdminOperation, Operation, S3Router};
|
||||
use crate::admin::runtime_sources::current_object_data_cache;
|
||||
use crate::app::object_data_cache::ObjectDataCacheAdapter;
|
||||
use crate::server::ADMIN_PREFIX;
|
||||
use crate::auth::{check_key_valid, get_session_token};
|
||||
use crate::server::{ADMIN_PREFIX, RemoteAddr};
|
||||
use http::{HeaderMap, HeaderValue};
|
||||
use hyper::{Method, StatusCode};
|
||||
use matchit::Params;
|
||||
@@ -75,14 +76,17 @@ pub fn register_object_data_cache_route(r: &mut S3Router<AdminOperation>) -> std
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// The pre-check keeps these endpoints' historical missing-credentials message;
|
||||
/// the shared gate reports "get cred failed".
|
||||
async fn authorize(req: &S3Request<Body>, action: AdminAction) -> S3Result<()> {
|
||||
if req.credentials.is_none() {
|
||||
let Some(input_cred) = req.credentials.as_ref() else {
|
||||
return Err(s3_error!(InvalidRequest, "missing credentials"));
|
||||
}
|
||||
authorize_admin_request(req, vec![Action::AdminAction(action)]).await?;
|
||||
Ok(())
|
||||
};
|
||||
let (cred, owner) =
|
||||
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
|
||||
let remote_addr = req
|
||||
.extensions
|
||||
.get::<Option<RemoteAddr>>()
|
||||
.and_then(|opt| opt.map(|addr| addr.0));
|
||||
validate_admin_request(&req.headers, &cred, owner, false, vec![Action::AdminAction(action)], remote_addr).await
|
||||
}
|
||||
|
||||
fn json_response<T: Serialize>(body: &T) -> S3Result<S3Response<(StatusCode, Body)>> {
|
||||
@@ -204,30 +208,6 @@ mod tests {
|
||||
assert_eq!(invalidation_outcome(&ObjectDataCacheInvalidationResult::NoOp), ("noop", 0));
|
||||
}
|
||||
|
||||
/// These endpoints authorize through the shared admin gate, which reports
|
||||
/// "get cred failed" for a credential-less request. The pre-check keeps the
|
||||
/// message they have always returned (rustfs/backlog#1829).
|
||||
#[tokio::test]
|
||||
async fn authorize_keeps_its_missing_credentials_message() {
|
||||
let req = S3Request {
|
||||
input: Body::from(String::new()),
|
||||
method: Method::GET,
|
||||
uri: "/rustfs/admin/v3/object-data-cache/stats".parse().expect("uri should parse"),
|
||||
headers: HeaderMap::new(),
|
||||
extensions: http::Extensions::new(),
|
||||
credentials: None,
|
||||
region: None,
|
||||
service: None,
|
||||
trailing_headers: None,
|
||||
};
|
||||
|
||||
let err = authorize(&req, AdminAction::ServerInfoAdminAction)
|
||||
.await
|
||||
.expect_err("a request without credentials must be rejected");
|
||||
assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
|
||||
assert_eq!(err.message(), Some("missing credentials"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stats_handler_requires_server_info_action() {
|
||||
// Guard the auth contract: the stats endpoint is a read, the flush
|
||||
|
||||
@@ -13,7 +13,7 @@
|
||||
// limitations under the License.
|
||||
|
||||
use crate::admin::{
|
||||
auth::authorize_admin_request,
|
||||
auth::validate_admin_request,
|
||||
plugin_contract::{
|
||||
PluginCatalogAdminDiscovery, PluginCatalogDomainEntry, PluginCatalogEntry, PluginCatalogResponse, PluginContractDomain,
|
||||
PluginContractEntrypointKind, PluginContractPackaging, PluginDistributionContract, PluginRuntimeContract,
|
||||
@@ -21,7 +21,8 @@ use crate::admin::{
|
||||
router::{AdminOperation, Operation, S3Router},
|
||||
runtime_sources::default_admin_usecase,
|
||||
};
|
||||
use crate::server::ADMIN_PREFIX;
|
||||
use crate::auth::{check_key_valid, get_session_token};
|
||||
use crate::server::{ADMIN_PREFIX, RemoteAddr};
|
||||
use http::{HeaderMap, HeaderValue, StatusCode};
|
||||
use hyper::Method;
|
||||
use matchit::Params;
|
||||
@@ -113,15 +114,23 @@ fn merge_catalog_descriptor(plugins: &mut HashMap<&'static str, PluginCatalogEnt
|
||||
}
|
||||
}
|
||||
|
||||
/// The pre-check keeps this endpoint's historical missing-credentials message;
|
||||
/// the shared gate reports "get cred failed".
|
||||
async fn authorize_plugin_catalog_request(req: &S3Request<Body>) -> S3Result<()> {
|
||||
if req.credentials.is_none() {
|
||||
let Some(input_cred) = &req.credentials else {
|
||||
return Err(s3_error!(InvalidRequest, "authentication required"));
|
||||
}
|
||||
};
|
||||
|
||||
authorize_admin_request(req, vec![Action::AdminAction(AdminAction::ServerInfoAdminAction)]).await?;
|
||||
Ok(())
|
||||
let (cred, owner) =
|
||||
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
|
||||
|
||||
validate_admin_request(
|
||||
&req.headers,
|
||||
&cred,
|
||||
owner,
|
||||
false,
|
||||
vec![Action::AdminAction(AdminAction::ServerInfoAdminAction)],
|
||||
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
fn build_json_response(
|
||||
@@ -166,30 +175,6 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
/// This endpoint authorizes through the shared admin gate, which reports
|
||||
/// "get cred failed" for a credential-less request. The pre-check keeps the
|
||||
/// message it has always returned (rustfs/backlog#1829).
|
||||
#[tokio::test]
|
||||
async fn plugin_catalog_gate_keeps_its_missing_credentials_message() {
|
||||
let req = s3s::S3Request {
|
||||
input: s3s::Body::from(String::new()),
|
||||
method: http::Method::GET,
|
||||
uri: http::Uri::from_static("/rustfs/admin/v4/plugins/catalog"),
|
||||
headers: http::HeaderMap::new(),
|
||||
extensions: http::Extensions::new(),
|
||||
credentials: None,
|
||||
region: None,
|
||||
service: None,
|
||||
trailing_headers: None,
|
||||
};
|
||||
|
||||
let err = super::authorize_plugin_catalog_request(&req)
|
||||
.await
|
||||
.expect_err("a request without credentials must be rejected");
|
||||
assert_eq!(err.code(), &s3s::S3ErrorCode::InvalidRequest);
|
||||
assert_eq!(err.message(), Some("authentication required"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn plugin_catalog_contains_representative_builtin_targets() {
|
||||
let response = build_catalog_response();
|
||||
|
||||
@@ -13,7 +13,7 @@
|
||||
// limitations under the License.
|
||||
|
||||
use crate::admin::{
|
||||
auth::authorize_admin_request,
|
||||
auth::validate_admin_request,
|
||||
handlers::audit_runtime_config::{load_server_config_from_store, remove_audit_target_config, set_audit_target_config},
|
||||
handlers::notify_runtime_access::{
|
||||
load_notification_config_snapshot, remove_notification_target_config, set_notification_target_config,
|
||||
@@ -29,9 +29,10 @@ use crate::admin::{
|
||||
},
|
||||
router::{AdminOperation, Operation, S3Router},
|
||||
};
|
||||
use crate::auth::{check_key_valid, get_session_token};
|
||||
use crate::server::{
|
||||
ADMIN_PREFIX, is_audit_module_enabled, is_notify_module_enabled, refresh_audit_module_enabled, refresh_notify_module_enabled,
|
||||
refresh_persisted_module_switches_from_store,
|
||||
ADMIN_PREFIX, RemoteAddr, is_audit_module_enabled, is_notify_module_enabled, refresh_audit_module_enabled,
|
||||
refresh_notify_module_enabled, refresh_persisted_module_switches_from_store,
|
||||
};
|
||||
use hyper::{Method, StatusCode};
|
||||
use matchit::Params;
|
||||
@@ -562,26 +563,42 @@ fn plugin_instance_matches_query(instance: &PluginInstanceEntry, query: &str) ->
|
||||
.any(|field| field.to_ascii_lowercase().contains(&query))
|
||||
}
|
||||
|
||||
/// The pre-check keeps this endpoint's historical missing-credentials message;
|
||||
/// the shared gate reports "get cred failed".
|
||||
async fn authorize_plugin_instance_request(req: &S3Request<Body>) -> S3Result<()> {
|
||||
if req.credentials.is_none() {
|
||||
let Some(input_cred) = &req.credentials else {
|
||||
return Err(s3_error!(InvalidRequest, "authentication required"));
|
||||
}
|
||||
};
|
||||
|
||||
authorize_admin_request(req, vec![Action::AdminAction(AdminAction::GetBucketTargetAction)]).await?;
|
||||
Ok(())
|
||||
let (cred, owner) =
|
||||
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
|
||||
|
||||
validate_admin_request(
|
||||
&req.headers,
|
||||
&cred,
|
||||
owner,
|
||||
false,
|
||||
vec![Action::AdminAction(AdminAction::GetBucketTargetAction)],
|
||||
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
/// The pre-check keeps this endpoint's historical missing-credentials message;
|
||||
/// the shared gate reports "get cred failed".
|
||||
async fn authorize_plugin_instance_write_request(req: &S3Request<Body>) -> S3Result<()> {
|
||||
if req.credentials.is_none() {
|
||||
let Some(input_cred) = &req.credentials else {
|
||||
return Err(s3_error!(InvalidRequest, "authentication required"));
|
||||
}
|
||||
};
|
||||
|
||||
authorize_admin_request(req, vec![Action::AdminAction(AdminAction::SetBucketTargetAction)]).await?;
|
||||
Ok(())
|
||||
let (cred, owner) =
|
||||
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
|
||||
|
||||
validate_admin_request(
|
||||
&req.headers,
|
||||
&cred,
|
||||
owner,
|
||||
false,
|
||||
vec![Action::AdminAction(AdminAction::SetBucketTargetAction)],
|
||||
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
fn plugin_instance_mutation_block_reason(
|
||||
@@ -925,36 +942,6 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
/// Both instance gates authorize through the shared admin gate, which reports
|
||||
/// "get cred failed" for a credential-less request. The pre-check keeps the
|
||||
/// message these endpoints have always returned (rustfs/backlog#1829).
|
||||
#[tokio::test]
|
||||
async fn plugin_instance_gates_keep_their_missing_credentials_message() {
|
||||
let credential_less_request = || S3Request {
|
||||
input: Body::from(String::new()),
|
||||
method: Method::GET,
|
||||
uri: Uri::from_static("/rustfs/admin/v4/plugins/instances"),
|
||||
headers: HeaderMap::new(),
|
||||
extensions: Extensions::new(),
|
||||
credentials: None,
|
||||
region: None,
|
||||
service: None,
|
||||
trailing_headers: None,
|
||||
};
|
||||
|
||||
for err in [
|
||||
super::authorize_plugin_instance_request(&credential_less_request())
|
||||
.await
|
||||
.expect_err("a request without credentials must be rejected"),
|
||||
super::authorize_plugin_instance_write_request(&credential_less_request())
|
||||
.await
|
||||
.expect_err("a request without credentials must be rejected"),
|
||||
] {
|
||||
assert_eq!(err.code(), &s3s::S3ErrorCode::InvalidRequest);
|
||||
assert_eq!(err.message(), Some("authentication required"));
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn configured_instance_without_runtime_appears_offline() {
|
||||
let config = Config(HashMap::from([(
|
||||
|
||||
@@ -126,6 +126,10 @@ const SITE_REPLICATION_JOIN_ADMISSION_LOCK_PATH: &str = "config/site-replication
|
||||
const SITE_REPL_ADD_SUCCESS: &str = "Requested sites were configured for replication successfully.";
|
||||
const SITE_REPL_EDIT_SUCCESS: &str = "Requested site was updated successfully.";
|
||||
const SITE_REPL_REMOVE_SUCCESS: &str = "Requested site(s) were removed from cluster replication successfully.";
|
||||
/// Local removal committed, but at least one peer could not be told. The
|
||||
/// cluster is diverged until the removal finishes — the reconcile tick keeps
|
||||
/// retrying it, and `replicate info` reports the pending operation meanwhile.
|
||||
const SITE_REPL_REMOVE_PARTIAL: &str = "Partial";
|
||||
const SITE_REPL_RESYNC_START: &str = "start";
|
||||
const SITE_REPL_RESYNC_CANCEL: &str = "cancel";
|
||||
const SITE_REPL_RESYNC_STATUS: &str = "status";
|
||||
@@ -713,6 +717,16 @@ struct SRPeerJoinResponse {
|
||||
peer: PeerInfo,
|
||||
#[serde(rename = "initialSyncErrorMessage", default, skip_serializing_if = "String::is_empty")]
|
||||
initial_sync_error_message: String,
|
||||
/// Whether the receiving site actually applied this join.
|
||||
///
|
||||
/// Three-valued on purpose. `None` means the peer did not report — MinIO
|
||||
/// answers a successful `SRPeerJoin` with an empty body, and RustFS peers
|
||||
/// older than this field say nothing either — so the initiator must NOT
|
||||
/// read it as a failure. `Some(false)` is an explicit no-op: the peer had
|
||||
/// already moved past the snapshot it was sent and wrote nothing, which
|
||||
/// used to be indistinguishable from success (rustfs/rustfs#5963).
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
applied: Option<bool>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
|
||||
@@ -2568,6 +2582,21 @@ fn apply_peer_join(
|
||||
state.peers = normalize_join_peers_for_local(local_peer, join_req.peers);
|
||||
initialize_join_peer_sync_state(&mut state.peers, defer_sync_state_enable);
|
||||
state.sync_state_initialized = true;
|
||||
// An accepted join supersedes a half-finished removal this site started:
|
||||
// the sender's snapshot IS the new topology, while the pending record only
|
||||
// exists to keep notifying peers about the OLD one. Leaving it set is what
|
||||
// kept a recovered site rejecting every peer bucket-op forever —
|
||||
// `SRPeerBucketOpsHandler` short-circuits on `pending_remove` BEFORE it
|
||||
// consults `enabled()`, so a successful re-add restored the topology on
|
||||
// both sides while replication stayed dead (rustfs/rustfs#5963).
|
||||
//
|
||||
// Safe against a concurrent removal: `SiteReplicationRemoveHandler` and
|
||||
// the join admission both hold the lifecycle guard, so a join is only ever
|
||||
// admitted before that handler starts or after it has returned.
|
||||
//
|
||||
// Deliberately NOT cleared here: the peer-edit high-water marks (see this
|
||||
// function's doc comment) — those fence edit ordering, not lifecycle.
|
||||
state.pending_remove = None;
|
||||
state.name = state
|
||||
.peers
|
||||
.get(&local_peer.deployment_id)
|
||||
@@ -3002,8 +3031,17 @@ fn reconcile_site_replication_wiring() -> std::pin::Pin<Box<dyn std::future::Fut
|
||||
|
||||
match load_site_replication_state().await {
|
||||
Ok(state) => {
|
||||
if state.pending_endpoint_refresh.is_some() || state.pending_remove.is_some() || state.pending_rotation.is_some()
|
||||
{
|
||||
if state.pending_endpoint_refresh.is_some() || state.pending_rotation.is_some() {
|
||||
return;
|
||||
}
|
||||
// A removal whose peers were unreachable is the one pending
|
||||
// marker that nothing else re-drives, and it wedges the site
|
||||
// while it sits there. Push it forward here rather than giving
|
||||
// up the round (rustfs/rustfs#5963). The reconcilers below
|
||||
// still skip this round either way: the topology is only
|
||||
// settled once the removal clears, and the next tick sees it.
|
||||
if let Some(pending_remove) = state.pending_remove.clone() {
|
||||
resume_pending_remove(&state, &pending_remove).await;
|
||||
return;
|
||||
}
|
||||
}
|
||||
@@ -7041,15 +7079,31 @@ async fn dequeue_site_replication_retry_event_for_generation(peer: &PeerInfo, pa
|
||||
}
|
||||
}
|
||||
|
||||
/// The removal's client-facing verdict.
|
||||
///
|
||||
/// A fully-notified removal keeps answering with the historical success string,
|
||||
/// byte for byte, so healthy runs stay wire-identical for every existing
|
||||
/// client. Only the path that used to LIE — peers that could not be notified,
|
||||
/// reported as unqualified success while the cluster silently diverged
|
||||
/// (rustfs/rustfs#5963) — now says `Partial`, matching the vocabulary
|
||||
/// `SRRotateServiceAccountHandler` already uses for the same situation.
|
||||
fn site_replication_remove_status(peer_errors: &[String]) -> ReplicateRemoveStatus {
|
||||
if peer_errors.is_empty() {
|
||||
return ReplicateRemoveStatus {
|
||||
status: SITE_REPL_REMOVE_SUCCESS.to_string(),
|
||||
err_detail: String::new(),
|
||||
api_version: Some(SITE_REPL_API_VERSION.to_string()),
|
||||
};
|
||||
}
|
||||
|
||||
let summaries: Vec<String> = peer_errors.iter().map(|error| summarize_peer_error_detail(error)).collect();
|
||||
ReplicateRemoveStatus {
|
||||
status: SITE_REPL_REMOVE_SUCCESS.to_string(),
|
||||
err_detail: if peer_errors.is_empty() {
|
||||
String::new()
|
||||
} else {
|
||||
let summaries: Vec<String> = peer_errors.iter().map(|error| summarize_peer_error_detail(error)).collect();
|
||||
summarize_peer_error_detail(&format!("failed to notify {} peer(s): {}", summaries.len(), summaries.join("; ")))
|
||||
},
|
||||
status: SITE_REPL_REMOVE_PARTIAL.to_string(),
|
||||
err_detail: summarize_peer_error_detail(&format!(
|
||||
"failed to notify {} peer(s): {}",
|
||||
summaries.len(),
|
||||
summaries.join("; ")
|
||||
)),
|
||||
api_version: Some(SITE_REPL_API_VERSION.to_string()),
|
||||
}
|
||||
}
|
||||
@@ -7226,6 +7280,137 @@ async fn clear_pending_remove(remove_id: &str) -> S3Result<()> {
|
||||
.await
|
||||
}
|
||||
|
||||
/// Push a half-finished removal one step forward: notify every peer that has
|
||||
/// not acked yet, then finalize locally if that completed the set. Returns the
|
||||
/// per-peer failures and whether the removal is now finished.
|
||||
///
|
||||
/// Shared by the operator-driven `SiteReplicationRemoveHandler` and the
|
||||
/// reconcile tick. The tick is what makes this self-healing: a removal whose
|
||||
/// peers were unreachable used to sit in `pending_remove` forever, and that one
|
||||
/// field gates every peer bucket-op (`SRPeerBucketOpsHandler` checks it first)
|
||||
/// plus every reconciler — so the site stayed wedged until an operator happened
|
||||
/// to re-run `replicate remove` (rustfs/rustfs#5963).
|
||||
///
|
||||
/// Callers must hold the lifecycle guard: this both notifies peers and, on the
|
||||
/// final step, takes the bucket-op write lock to clean up local rules.
|
||||
async fn drive_pending_remove(pending_remove: &PendingRemove, local_peer: &PeerInfo) -> S3Result<(Vec<String>, bool)> {
|
||||
let mut peer_errors = Vec::new();
|
||||
let mut secret_candidates = pending_remove.secret_candidates.clone();
|
||||
if pending_remove.service_account_access_key.is_empty() {
|
||||
peer_errors.push("site replication service account unavailable".to_string());
|
||||
} else if let Ok(service_account_secret_key) =
|
||||
site_replicator_service_account_secret(&pending_remove.service_account_access_key).await
|
||||
{
|
||||
record_pending_remove_secret_candidate(&pending_remove.id, service_account_secret_key.clone()).await?;
|
||||
push_unique_secret_candidate(&mut secret_candidates, service_account_secret_key);
|
||||
}
|
||||
|
||||
if secret_candidates.is_empty() {
|
||||
peer_errors.push("site replication service account secret unavailable".to_string());
|
||||
} else {
|
||||
for peer in pending_remove.original_peers.values() {
|
||||
if same_identity_endpoint(&peer.endpoint, &local_peer.endpoint)
|
||||
|| pending_remove.acked_deployment_ids.contains(&peer.deployment_id)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
if let Err(err) = send_peer_admin_request_with_secret_candidates(
|
||||
&runtime_peer_connection(peer)?,
|
||||
SITE_REPLICATION_PEER_REMOVE_PATH,
|
||||
&pending_remove.service_account_access_key,
|
||||
&secret_candidates,
|
||||
&pending_remove.req,
|
||||
)
|
||||
.await
|
||||
{
|
||||
let err_detail = summarize_peer_error_detail(&format!("{}: {err}", peer.endpoint));
|
||||
warn!(
|
||||
event = EVENT_ADMIN_SITE_REPLICATION_STATE,
|
||||
component = LOG_COMPONENT_ADMIN,
|
||||
subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
|
||||
peer = %peer.endpoint,
|
||||
result = "peer_remove_notification_failed",
|
||||
error = %err_detail,
|
||||
"admin site replication state"
|
||||
);
|
||||
peer_errors.push(err_detail);
|
||||
} else {
|
||||
mark_pending_remove_peer_acked(&pending_remove.id, &peer.deployment_id).await?;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let finalize_candidate = pending_remove_ready_to_finalize(&pending_remove.id, local_peer).await?;
|
||||
let complete = if let Some(finalized_remove) = finalize_candidate {
|
||||
let _bucket_op_guard = SITE_REPLICATION_BUCKET_OP_LOCK.write().await;
|
||||
let removed_deployment_ids = removed_deployment_ids_for_pending_remove(&finalized_remove, local_peer);
|
||||
match cleanup_removed_site_replication_buckets(&removed_deployment_ids).await {
|
||||
Ok(removed) => {
|
||||
if removed > 0 {
|
||||
info!(
|
||||
event = EVENT_ADMIN_SITE_REPLICATION_STATE,
|
||||
component = LOG_COMPONENT_ADMIN,
|
||||
subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
|
||||
removed,
|
||||
result = "remove_cleanup_completed",
|
||||
"admin site replication state"
|
||||
);
|
||||
}
|
||||
clear_pending_remove(&pending_remove.id).await?;
|
||||
true
|
||||
}
|
||||
Err(err) => {
|
||||
peer_errors.push(summarize_peer_error_detail(&format!("local remove cleanup failed: {err}")));
|
||||
false
|
||||
}
|
||||
}
|
||||
} else {
|
||||
false
|
||||
};
|
||||
|
||||
Ok((peer_errors, complete))
|
||||
}
|
||||
|
||||
/// The reconcile tick's half of [`drive_pending_remove`]: resume the removal
|
||||
/// this site could not finish, and report the outcome. Runs under the tick's
|
||||
/// lifecycle guard, which is what keeps it from racing an operator re-running
|
||||
/// `replicate remove` (that handler takes the same guard).
|
||||
async fn resume_pending_remove(state: &SiteReplicationState, pending_remove: &PendingRemove) {
|
||||
let local_peer = current_local_runtime_peer(state);
|
||||
match drive_pending_remove(pending_remove, &local_peer).await {
|
||||
Ok((peer_errors, complete)) => {
|
||||
if complete && peer_errors.is_empty() {
|
||||
info!(
|
||||
event = EVENT_ADMIN_SITE_REPLICATION_STATE,
|
||||
component = LOG_COMPONENT_ADMIN,
|
||||
subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
|
||||
result = "pending_remove_resumed",
|
||||
"admin site replication state"
|
||||
);
|
||||
} else {
|
||||
warn!(
|
||||
event = EVENT_ADMIN_SITE_REPLICATION_STATE,
|
||||
component = LOG_COMPONENT_ADMIN,
|
||||
subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
|
||||
result = "pending_remove_still_pending",
|
||||
error_count = peer_errors.len(),
|
||||
"admin site replication state"
|
||||
);
|
||||
}
|
||||
}
|
||||
Err(err) => {
|
||||
warn!(
|
||||
event = EVENT_ADMIN_SITE_REPLICATION_STATE,
|
||||
component = LOG_COMPONENT_ADMIN,
|
||||
subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
|
||||
result = "pending_remove_resume_failed",
|
||||
error = ?err,
|
||||
"admin site replication state"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn removed_deployment_ids_for_pending_remove(pending: &PendingRemove, local_peer: &PeerInfo) -> HashSet<String> {
|
||||
if pending.req.remove_all || pending.req.site_names.iter().any(|name| name == &local_peer.name) {
|
||||
return pending
|
||||
@@ -9656,9 +9841,12 @@ pub struct SiteReplicationAddHandler {}
|
||||
/// peer identity from the add preflight metainfo in that case.
|
||||
fn parse_peer_join_response(body: &[u8], fallback_peer: PeerInfo) -> Result<SRPeerJoinResponse, serde_json::Error> {
|
||||
if body.iter().all(u8::is_ascii_whitespace) {
|
||||
// MinIO's empty-body success. `applied` stays `None`: the peer told us
|
||||
// nothing, which must not be reported as a no-op join.
|
||||
return Ok(SRPeerJoinResponse {
|
||||
peer: fallback_peer,
|
||||
initial_sync_error_message: String::new(),
|
||||
applied: None,
|
||||
});
|
||||
}
|
||||
serde_json::from_slice(body)
|
||||
@@ -9761,6 +9949,19 @@ impl Operation for SiteReplicationAddHandler {
|
||||
if !join_response.initial_sync_error_message.is_empty() {
|
||||
initial_sync_errors.push(format!("{}: {}", site.endpoint, join_response.initial_sync_error_message));
|
||||
}
|
||||
// An explicit no-op join. The peer answered 200 but wrote nothing —
|
||||
// its persisted state is already newer than the snapshot it was
|
||||
// sent — so the add is only PARTIALLY configured and saying
|
||||
// "configured successfully" would be a lie (rustfs/rustfs#5963).
|
||||
// `None` (a MinIO peer, or one older than the field) is not a
|
||||
// no-op signal and is deliberately not reported.
|
||||
if join_response.applied == Some(false) {
|
||||
initial_sync_errors.push(format!(
|
||||
"{}: peer did not apply the join (its site replication state is newer than the snapshot it was sent); \
|
||||
the site is not configured against this peer",
|
||||
site.endpoint
|
||||
));
|
||||
}
|
||||
state = reconcile_peer_with_actual_identity(state, join_response.peer);
|
||||
let reconciled_peer = existing_peer_for_endpoint(&state, &site.endpoint).ok_or_else(|| {
|
||||
S3Error::with_message(
|
||||
@@ -9933,79 +10134,7 @@ impl Operation for SiteReplicationRemoveHandler {
|
||||
.await?
|
||||
};
|
||||
|
||||
let mut peer_errors = Vec::new();
|
||||
let mut secret_candidates = pending_remove.secret_candidates.clone();
|
||||
if pending_remove.service_account_access_key.is_empty() {
|
||||
peer_errors.push("site replication service account unavailable".to_string());
|
||||
} else if let Ok(service_account_secret_key) =
|
||||
site_replicator_service_account_secret(&pending_remove.service_account_access_key).await
|
||||
{
|
||||
record_pending_remove_secret_candidate(&pending_remove.id, service_account_secret_key.clone()).await?;
|
||||
push_unique_secret_candidate(&mut secret_candidates, service_account_secret_key);
|
||||
}
|
||||
|
||||
if secret_candidates.is_empty() {
|
||||
peer_errors.push("site replication service account secret unavailable".to_string());
|
||||
} else {
|
||||
for peer in pending_remove.original_peers.values() {
|
||||
if same_identity_endpoint(&peer.endpoint, &local_peer.endpoint)
|
||||
|| pending_remove.acked_deployment_ids.contains(&peer.deployment_id)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
if let Err(err) = send_peer_admin_request_with_secret_candidates(
|
||||
&runtime_peer_connection(peer)?,
|
||||
SITE_REPLICATION_PEER_REMOVE_PATH,
|
||||
&pending_remove.service_account_access_key,
|
||||
&secret_candidates,
|
||||
&pending_remove.req,
|
||||
)
|
||||
.await
|
||||
{
|
||||
let err_detail = summarize_peer_error_detail(&format!("{}: {err}", peer.endpoint));
|
||||
warn!(
|
||||
event = EVENT_ADMIN_SITE_REPLICATION_STATE,
|
||||
component = LOG_COMPONENT_ADMIN,
|
||||
subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
|
||||
peer = %peer.endpoint,
|
||||
result = "peer_remove_notification_failed",
|
||||
error = %err_detail,
|
||||
"admin site replication state"
|
||||
);
|
||||
peer_errors.push(err_detail);
|
||||
} else {
|
||||
mark_pending_remove_peer_acked(&pending_remove.id, &peer.deployment_id).await?;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let finalize_candidate = pending_remove_ready_to_finalize(&pending_remove.id, &local_peer).await?;
|
||||
let complete = if let Some(finalized_remove) = finalize_candidate {
|
||||
let _bucket_op_guard = SITE_REPLICATION_BUCKET_OP_LOCK.write().await;
|
||||
let removed_deployment_ids = removed_deployment_ids_for_pending_remove(&finalized_remove, &local_peer);
|
||||
match cleanup_removed_site_replication_buckets(&removed_deployment_ids).await {
|
||||
Ok(removed) => {
|
||||
if removed > 0 {
|
||||
info!(
|
||||
event = EVENT_ADMIN_SITE_REPLICATION_STATE,
|
||||
component = LOG_COMPONENT_ADMIN,
|
||||
subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
|
||||
removed,
|
||||
result = "remove_cleanup_completed",
|
||||
"admin site replication state"
|
||||
);
|
||||
}
|
||||
clear_pending_remove(&pending_remove.id).await?;
|
||||
true
|
||||
}
|
||||
Err(err) => {
|
||||
peer_errors.push(summarize_peer_error_detail(&format!("local remove cleanup failed: {err}")));
|
||||
false
|
||||
}
|
||||
}
|
||||
} else {
|
||||
false
|
||||
};
|
||||
let (mut peer_errors, complete) = drive_pending_remove(&pending_remove, &local_peer).await?;
|
||||
if !complete && peer_errors.is_empty() {
|
||||
peer_errors.push("site replication remove is still pending".to_string());
|
||||
}
|
||||
@@ -10019,6 +10148,25 @@ impl Operation for SiteReplicationRemoveHandler {
|
||||
}
|
||||
}
|
||||
|
||||
/// The `replicate info` projection.
|
||||
///
|
||||
/// Carries the peer-facing health this endpoint used to omit entirely: a peer
|
||||
/// rejecting every operation, or a removal stuck mid-flight, left `info`
|
||||
/// reporting a perfectly healthy cluster while replication was dead — both were
|
||||
/// only visible through `replicate status --json` (rustfs/rustfs#5963). Split
|
||||
/// out so that omission is a test failure rather than an invisible regression.
|
||||
fn site_replication_info_for(state: &SiteReplicationState, local_peer: &PeerInfo) -> SiteReplicationInfo {
|
||||
SiteReplicationInfo {
|
||||
enabled: state.enabled(),
|
||||
name: local_peer.name.clone(),
|
||||
sites: state.peers.values().cloned().collect(),
|
||||
service_account_access_key: state.service_account_access_key.clone(),
|
||||
api_version: Some(SITE_REPL_API_VERSION.to_string()),
|
||||
retry_stats: retry_stats_for_state(state),
|
||||
pending_operation: pending_operation_for_state(state, local_peer),
|
||||
}
|
||||
}
|
||||
|
||||
pub struct SiteReplicationInfoHandler {}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
@@ -10027,14 +10175,7 @@ impl Operation for SiteReplicationInfoHandler {
|
||||
validate_site_replication_admin_request(&req, AdminAction::SiteReplicationInfoAction).await?;
|
||||
let state = load_site_replication_state().await?;
|
||||
let local_peer = current_local_peer(&req, &state);
|
||||
let info = SiteReplicationInfo {
|
||||
enabled: state.enabled(),
|
||||
name: local_peer.name,
|
||||
sites: state.peers.values().cloned().collect(),
|
||||
service_account_access_key: state.service_account_access_key,
|
||||
api_version: Some(SITE_REPL_API_VERSION.to_string()),
|
||||
};
|
||||
json_response(&info)
|
||||
json_response(&site_replication_info_for(&state, &local_peer))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -10253,6 +10394,28 @@ async fn apply_peer_join_service_account(join_req: SRPeerJoinReq) -> S3Result<()
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// The answer to a join this site refused to apply because it had already
|
||||
/// moved past the sender's snapshot. Split out so the verdict itself is
|
||||
/// testable: answering `applied: Some(true)` here (or omitting the field) is
|
||||
/// exactly the silent no-op that made `replicate add` report success against a
|
||||
/// peer that wrote nothing (rustfs/rustfs#5963).
|
||||
fn superseded_join_response(peer: PeerInfo) -> SRPeerJoinResponse {
|
||||
SRPeerJoinResponse {
|
||||
peer,
|
||||
initial_sync_error_message: String::new(),
|
||||
applied: Some(false),
|
||||
}
|
||||
}
|
||||
|
||||
/// The answer to a join this site committed.
|
||||
fn applied_join_response(peer: PeerInfo, initial_sync_error_message: String) -> SRPeerJoinResponse {
|
||||
SRPeerJoinResponse {
|
||||
peer,
|
||||
initial_sync_error_message,
|
||||
applied: Some(true),
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl Operation for SRPeerJoinHandler {
|
||||
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
|
||||
@@ -10275,10 +10438,14 @@ impl Operation for SRPeerJoinHandler {
|
||||
let (state, local_peer) = match committed {
|
||||
PeerJoinOutcome::Applied(state, local_peer) => (*state, local_peer),
|
||||
PeerJoinOutcome::Superseded(peer) => {
|
||||
return json_response(&SRPeerJoinResponse {
|
||||
peer,
|
||||
..Default::default()
|
||||
});
|
||||
warn!(
|
||||
event = EVENT_ADMIN_SITE_REPLICATION_STATE,
|
||||
component = LOG_COMPONENT_ADMIN,
|
||||
subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
|
||||
result = "join_superseded",
|
||||
"admin site replication state"
|
||||
);
|
||||
return json_response(&superseded_join_response(peer));
|
||||
}
|
||||
};
|
||||
// Fix 1 (receiving side): ensure the joining peer also sets up replication for any
|
||||
@@ -10297,10 +10464,10 @@ impl Operation for SRPeerJoinHandler {
|
||||
"admin site replication state"
|
||||
);
|
||||
}
|
||||
json_response(&SRPeerJoinResponse {
|
||||
peer: state.peers.get(&local_peer.deployment_id).cloned().unwrap_or(local_peer),
|
||||
initial_sync_error_message: backfill_errors.render(),
|
||||
})
|
||||
json_response(&applied_join_response(
|
||||
state.peers.get(&local_peer.deployment_id).cloned().unwrap_or(local_peer),
|
||||
backfill_errors.render(),
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -11344,7 +11511,11 @@ impl Operation for SRRotateServiceAccountHandler {
|
||||
{
|
||||
continue;
|
||||
}
|
||||
if let Err(err) = send_peer_admin_request_with_secret_candidates(
|
||||
// A superseded join returns BEFORE `apply_iam`, so a no-op answer
|
||||
// means the peer never installed the new secret. Acking it would
|
||||
// finalize a rotation half the mesh cannot authenticate against
|
||||
// (rustfs/rustfs#5963).
|
||||
let rotation_error = match send_peer_admin_request_with_secret_candidates(
|
||||
&runtime_peer_connection(peer)?,
|
||||
SITE_REPLICATION_PEER_JOIN_PATH,
|
||||
&pending_rotation.access_key,
|
||||
@@ -11353,7 +11524,20 @@ impl Operation for SRRotateServiceAccountHandler {
|
||||
)
|
||||
.await
|
||||
{
|
||||
let detail = summarize_peer_error_detail(&format!("{}: {err}", peer.endpoint));
|
||||
Err(err) => Some(summarize_peer_error_detail(&format!("{}: {err}", peer.endpoint))),
|
||||
Ok(body) => match parse_peer_join_response(&body, peer.clone()) {
|
||||
Ok(response) if response.applied == Some(false) => Some(summarize_peer_error_detail(&format!(
|
||||
"{}: peer did not apply the rotation join (its site replication state is newer than the snapshot it \
|
||||
was sent); the new service account secret was not installed",
|
||||
peer.endpoint
|
||||
))),
|
||||
// Unparseable bodies keep the pre-existing behaviour: the
|
||||
// transport succeeded, and MinIO peers answer with an empty
|
||||
// body this helper already tolerates.
|
||||
Ok(_) | Err(_) => None,
|
||||
},
|
||||
};
|
||||
if let Some(detail) = rotation_error {
|
||||
warn!(
|
||||
event = EVENT_ADMIN_SITE_REPLICATION_STATE,
|
||||
component = LOG_COMPONENT_ADMIN,
|
||||
@@ -15639,9 +15823,17 @@ mod tests {
|
||||
site_replication_remove_status(&["peer request to https://remote.example.com failed with 403 Forbidden".to_string()]);
|
||||
|
||||
assert!(state.peers.is_empty());
|
||||
assert_eq!(status.status, SITE_REPL_REMOVE_SUCCESS);
|
||||
assert_eq!(
|
||||
status.status, SITE_REPL_REMOVE_PARTIAL,
|
||||
"a removal whose peer could not be notified must not report unqualified success"
|
||||
);
|
||||
assert!(status.err_detail.contains("failed to notify 1 peer"));
|
||||
assert!(status.err_detail.contains("403 Forbidden"));
|
||||
|
||||
// The fully-notified path stays byte-identical for existing clients.
|
||||
let clean = site_replication_remove_status(&[]);
|
||||
assert_eq!(clean.status, SITE_REPL_REMOVE_SUCCESS);
|
||||
assert!(clean.err_detail.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -16982,16 +17174,22 @@ mod tests {
|
||||
assert_eq!(response.peer.deployment_id, "remote-deployment");
|
||||
assert_eq!(response.peer.endpoint, "https://remote.example.com");
|
||||
assert!(response.initial_sync_error_message.is_empty());
|
||||
assert_eq!(
|
||||
response.applied, None,
|
||||
"a MinIO empty-body success reports nothing; it must not read as a no-op join"
|
||||
);
|
||||
}
|
||||
|
||||
let json = serde_json::to_vec(&SRPeerJoinResponse {
|
||||
peer: peer("actual", "https://actual.example.com"),
|
||||
initial_sync_error_message: "sync failed".to_string(),
|
||||
applied: Some(true),
|
||||
})
|
||||
.expect("serialize join response");
|
||||
let response = parse_peer_join_response(&json, fallback.clone()).expect("parse join response body");
|
||||
assert_eq!(response.peer.endpoint, "https://actual.example.com");
|
||||
assert_eq!(response.initial_sync_error_message, "sync failed");
|
||||
assert_eq!(response.applied, Some(true));
|
||||
|
||||
assert!(parse_peer_join_response(b"not-json", fallback).is_err());
|
||||
}
|
||||
@@ -17712,13 +17910,319 @@ mod tests {
|
||||
.expect("parse legacy peer join response");
|
||||
|
||||
assert!(response.initial_sync_error_message.is_empty());
|
||||
assert_eq!(
|
||||
response.applied, None,
|
||||
"a peer older than the field says nothing about whether it applied the join"
|
||||
);
|
||||
|
||||
let value = serde_json::to_value(SRPeerJoinResponse {
|
||||
peer: peer("remote", "https://remote.example.com"),
|
||||
initial_sync_error_message: "bucket setup failed".to_string(),
|
||||
applied: Some(true),
|
||||
})
|
||||
.expect("serialize peer join response");
|
||||
assert_eq!(value.get("initialSyncErrorMessage").and_then(Value::as_str), Some("bucket setup failed"));
|
||||
assert_eq!(value.get("applied").and_then(Value::as_bool), Some(true));
|
||||
|
||||
// An unset verdict must not appear on the wire, so a peer that never
|
||||
// learned the field keeps deserializing byte-identical payloads.
|
||||
let value = serde_json::to_value(SRPeerJoinResponse {
|
||||
peer: peer("remote", "https://remote.example.com"),
|
||||
initial_sync_error_message: String::new(),
|
||||
applied: None,
|
||||
})
|
||||
.expect("serialize peer join response");
|
||||
assert!(value.get("applied").is_none(), "an unset verdict must be omitted: {value}");
|
||||
}
|
||||
|
||||
/// rustfs/rustfs#5963: a removal that could not notify its peers leaves
|
||||
/// `pending_remove` set, and that field alone makes `SRPeerBucketOpsHandler`
|
||||
/// reject every peer operation — before it ever consults `enabled()`. A
|
||||
/// later join restored the topology but left the marker, so a "successful"
|
||||
/// re-add produced a cluster that reported Enabled/2-sites on both sides
|
||||
/// while replication stayed dead. The join must clear it.
|
||||
#[test]
|
||||
fn peer_join_clears_a_stuck_pending_remove() {
|
||||
let local = PeerInfo {
|
||||
deployment_id: "site-b".to_string(),
|
||||
..peer("site-b", "https://site-b.example.com")
|
||||
};
|
||||
let remote = PeerInfo {
|
||||
deployment_id: "site-a".to_string(),
|
||||
..peer("site-a", "https://site-a.example.com")
|
||||
};
|
||||
let mut state = SiteReplicationState {
|
||||
peers: BTreeMap::from([(local.deployment_id.clone(), local.clone())]),
|
||||
pending_remove: Some(PendingRemove {
|
||||
id: "stuck-remove".to_string(),
|
||||
req: SRRemoveReq {
|
||||
remove_all: true,
|
||||
..Default::default()
|
||||
},
|
||||
service_account_access_key: SITE_REPLICATOR_SERVICE_ACCOUNT.to_string(),
|
||||
secret_candidates: Vec::new(),
|
||||
original_peers: BTreeMap::from([
|
||||
(local.deployment_id.clone(), local.clone()),
|
||||
(remote.deployment_id.clone(), remote.clone()),
|
||||
]),
|
||||
acked_deployment_ids: BTreeSet::new(),
|
||||
updated_at: Some(OffsetDateTime::now_utc()),
|
||||
}),
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
apply_peer_join(
|
||||
&mut state,
|
||||
&local,
|
||||
SRPeerJoinReq {
|
||||
svc_acct_access_key: SITE_REPLICATOR_SERVICE_ACCOUNT.to_string(),
|
||||
svc_acct_secret_key: "svc-secret".to_string(),
|
||||
svc_acct_parent: "root".to_string(),
|
||||
peers: BTreeMap::from([
|
||||
(local.deployment_id.clone(), local.clone()),
|
||||
(remote.deployment_id.clone(), remote),
|
||||
]),
|
||||
updated_at: Some(OffsetDateTime::now_utc()),
|
||||
},
|
||||
false,
|
||||
);
|
||||
|
||||
assert!(
|
||||
state.pending_remove.is_none(),
|
||||
"an accepted join supersedes the half-finished removal it lands on"
|
||||
);
|
||||
assert!(state.enabled(), "the join restores the two-site topology");
|
||||
// The guard `SRPeerBucketOpsHandler` evaluates, asserted directly: with
|
||||
// the marker cleared and the topology back, peer bucket-ops are
|
||||
// admitted again.
|
||||
assert!(
|
||||
state.pending_remove.is_none() && state.enabled(),
|
||||
"the bucket-ops admission predicate must now pass"
|
||||
);
|
||||
}
|
||||
|
||||
/// The fence marks are lifecycle-independent and must survive the clearing
|
||||
/// above — wiping them would reopen the rollback window the fence closes.
|
||||
#[test]
|
||||
fn peer_join_clearing_pending_remove_keeps_edit_generation_marks() {
|
||||
let local = PeerInfo {
|
||||
deployment_id: "site-b".to_string(),
|
||||
..peer("site-b", "https://site-b.example.com")
|
||||
};
|
||||
let remote = PeerInfo {
|
||||
deployment_id: "site-a".to_string(),
|
||||
..peer("site-a", "https://site-a.example.com")
|
||||
};
|
||||
let mut state = SiteReplicationState {
|
||||
peers: BTreeMap::from([(local.deployment_id.clone(), local.clone())]),
|
||||
applied_edit_generations: BTreeMap::from([(remote.deployment_id.clone(), 7)]),
|
||||
pending_remove: Some(PendingRemove {
|
||||
id: "stuck-remove".to_string(),
|
||||
req: SRRemoveReq {
|
||||
remove_all: true,
|
||||
..Default::default()
|
||||
},
|
||||
service_account_access_key: SITE_REPLICATOR_SERVICE_ACCOUNT.to_string(),
|
||||
secret_candidates: Vec::new(),
|
||||
original_peers: BTreeMap::from([
|
||||
(local.deployment_id.clone(), local.clone()),
|
||||
(remote.deployment_id.clone(), remote.clone()),
|
||||
]),
|
||||
acked_deployment_ids: BTreeSet::new(),
|
||||
updated_at: Some(OffsetDateTime::now_utc()),
|
||||
}),
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
apply_peer_join(
|
||||
&mut state,
|
||||
&local,
|
||||
SRPeerJoinReq {
|
||||
svc_acct_access_key: SITE_REPLICATOR_SERVICE_ACCOUNT.to_string(),
|
||||
svc_acct_secret_key: "svc-secret".to_string(),
|
||||
svc_acct_parent: "root".to_string(),
|
||||
peers: BTreeMap::from([
|
||||
(local.deployment_id.clone(), local.clone()),
|
||||
(remote.deployment_id.clone(), remote.clone()),
|
||||
]),
|
||||
updated_at: Some(OffsetDateTime::now_utc()),
|
||||
},
|
||||
false,
|
||||
);
|
||||
|
||||
assert!(state.pending_remove.is_none());
|
||||
assert_eq!(
|
||||
state.applied_edit_generations.get(&remote.deployment_id),
|
||||
Some(&7),
|
||||
"clearing the lifecycle marker must not touch the ordering fence"
|
||||
);
|
||||
}
|
||||
|
||||
/// rustfs/rustfs#5963: the two join verdicts must be distinguishable on the
|
||||
/// wire. `Some(true)`/`Some(false)` is what lets the initiator tell a real
|
||||
/// configuration from a 200 that wrote nothing; flipping either one back to
|
||||
/// an unset verdict re-hides the no-op.
|
||||
#[test]
|
||||
fn join_verdicts_are_distinguishable_on_the_wire() {
|
||||
let remote = peer("remote", "https://remote.example.com");
|
||||
|
||||
let superseded = superseded_join_response(remote.clone());
|
||||
assert_eq!(
|
||||
superseded.applied,
|
||||
Some(false),
|
||||
"a join this site refused to apply must say so explicitly"
|
||||
);
|
||||
assert!(superseded.initial_sync_error_message.is_empty());
|
||||
|
||||
let applied = applied_join_response(remote, "bucket setup failed".to_string());
|
||||
assert_eq!(applied.applied, Some(true));
|
||||
assert_eq!(applied.initial_sync_error_message, "bucket setup failed");
|
||||
|
||||
// Round-tripping through the wire keeps the two apart — the initiator
|
||||
// only ever sees the serialized form.
|
||||
let decoded: SRPeerJoinResponse =
|
||||
serde_json::from_slice(&serde_json::to_vec(&superseded_join_response(peer("r", "https://r.example.com"))).unwrap())
|
||||
.expect("round-trip superseded verdict");
|
||||
assert_eq!(decoded.applied, Some(false));
|
||||
}
|
||||
|
||||
/// rustfs/rustfs#5963: a stuck removal must be visible on the endpoint
|
||||
/// operators actually run. `replicate info` used to report only
|
||||
/// `enabled: false`, which reads as "never configured" rather than "a
|
||||
/// removal is wedged here and this site rejects every peer operation".
|
||||
#[test]
|
||||
fn site_replication_info_reports_a_wedged_removal() {
|
||||
let local = PeerInfo {
|
||||
deployment_id: "site-b".to_string(),
|
||||
..peer("site-b", "https://site-b.example.com")
|
||||
};
|
||||
let remote = PeerInfo {
|
||||
deployment_id: "site-a".to_string(),
|
||||
..peer("site-a", "https://site-a.example.com")
|
||||
};
|
||||
let state = SiteReplicationState {
|
||||
name: "site-b".to_string(),
|
||||
peers: BTreeMap::from([(local.deployment_id.clone(), local.clone())]),
|
||||
pending_remove: Some(PendingRemove {
|
||||
id: "stuck-remove".to_string(),
|
||||
req: SRRemoveReq {
|
||||
remove_all: true,
|
||||
..Default::default()
|
||||
},
|
||||
service_account_access_key: SITE_REPLICATOR_SERVICE_ACCOUNT.to_string(),
|
||||
secret_candidates: Vec::new(),
|
||||
original_peers: BTreeMap::from([
|
||||
(local.deployment_id.clone(), local.clone()),
|
||||
(remote.deployment_id.clone(), remote.clone()),
|
||||
]),
|
||||
acked_deployment_ids: BTreeSet::new(),
|
||||
updated_at: Some(OffsetDateTime::now_utc()),
|
||||
}),
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
let info = site_replication_info_for(&state, &local);
|
||||
assert!(!info.enabled, "the peer set is already torn down");
|
||||
let pending = info
|
||||
.pending_operation
|
||||
.as_ref()
|
||||
.expect("a wedged removal must surface as a pending operation");
|
||||
assert_eq!(pending.operation, "remove");
|
||||
assert!(
|
||||
pending.pending_peers.contains(&remote.deployment_id),
|
||||
"the peer that was never notified must be named: {pending:?}"
|
||||
);
|
||||
}
|
||||
|
||||
/// The source side of the same failure: peer operations are being rejected,
|
||||
/// the topology still looks like a healthy two-site cluster, and `info` has
|
||||
/// to say the deliveries are failing.
|
||||
#[test]
|
||||
fn site_replication_info_reports_failing_peer_deliveries() {
|
||||
let local = PeerInfo {
|
||||
deployment_id: "site-a".to_string(),
|
||||
..peer("site-a", "https://site-a.example.com")
|
||||
};
|
||||
let remote = PeerInfo {
|
||||
deployment_id: "site-b".to_string(),
|
||||
..peer("site-b", "https://site-b.example.com")
|
||||
};
|
||||
let state = SiteReplicationState {
|
||||
name: "site-a".to_string(),
|
||||
peers: BTreeMap::from([
|
||||
(local.deployment_id.clone(), local.clone()),
|
||||
(remote.deployment_id.clone(), remote.clone()),
|
||||
]),
|
||||
retry_queue: vec![SiteReplicationRetryEvent {
|
||||
id: "evt".to_string(),
|
||||
peer_deployment_id: remote.deployment_id.clone(),
|
||||
peer_endpoint: remote.endpoint,
|
||||
path: "/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=demo&operation=make-with-versioning".to_string(),
|
||||
retry_count: 9,
|
||||
failed: true,
|
||||
last_error: "site replication is not enabled".to_string(),
|
||||
updated_at: Some(OffsetDateTime::now_utc()),
|
||||
edit_generation: None,
|
||||
}],
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
let info = site_replication_info_for(&state, &local);
|
||||
assert!(info.enabled, "the topology still reports two sites — that was the trap");
|
||||
let stats = info
|
||||
.retry_stats
|
||||
.as_ref()
|
||||
.expect("a peer rejecting every delivery must be visible in `info`");
|
||||
assert_eq!(stats.failed, 1);
|
||||
assert_eq!(stats.last_error, "site replication is not enabled");
|
||||
|
||||
// A healthy site must stay wire-identical to before the field existed.
|
||||
let healthy = SiteReplicationState {
|
||||
retry_queue: Vec::new(),
|
||||
..state
|
||||
};
|
||||
let info = site_replication_info_for(&healthy, &local);
|
||||
assert!(info.retry_stats.is_none());
|
||||
assert!(info.pending_operation.is_none());
|
||||
}
|
||||
|
||||
/// rustfs/rustfs#5963: `replicate info` reported a healthy cluster while
|
||||
/// every peer operation was failing. The health it used to omit now rides
|
||||
/// along, and a healthy site still serializes without the new fields.
|
||||
#[test]
|
||||
fn site_replication_info_health_fields_are_absent_when_healthy() {
|
||||
let healthy = SiteReplicationInfo {
|
||||
enabled: true,
|
||||
name: "site-a".to_string(),
|
||||
sites: vec![peer("site-a", "https://site-a.example.com")],
|
||||
service_account_access_key: SITE_REPLICATOR_SERVICE_ACCOUNT.to_string(),
|
||||
api_version: Some(SITE_REPL_API_VERSION.to_string()),
|
||||
retry_stats: None,
|
||||
pending_operation: None,
|
||||
};
|
||||
let value = serde_json::to_value(&healthy).expect("serialize info");
|
||||
assert!(value.get("retryStats").is_none(), "a healthy site must not grow fields: {value}");
|
||||
assert!(value.get("pendingOperation").is_none(), "a healthy site must not grow fields: {value}");
|
||||
|
||||
let degraded = SiteReplicationInfo {
|
||||
retry_stats: Some(SRRetryStats {
|
||||
pending: 1,
|
||||
failed: 4,
|
||||
last_error: "site replication is not enabled".to_string(),
|
||||
api_version: Some(SITE_REPL_API_VERSION.to_string()),
|
||||
}),
|
||||
..healthy
|
||||
};
|
||||
let value = serde_json::to_value(°raded).expect("serialize info");
|
||||
assert_eq!(
|
||||
value.pointer("/retryStats/failed").and_then(Value::as_u64),
|
||||
Some(4),
|
||||
"a source site whose peer rejects everything must say so in `info`"
|
||||
);
|
||||
assert_eq!(
|
||||
value.pointer("/retryStats/lastError").and_then(Value::as_str),
|
||||
Some("site replication is not enabled")
|
||||
);
|
||||
}
|
||||
|
||||
// Fix 5: remove --all must purge local state unconditionally even when peer errors occur
|
||||
@@ -17766,12 +18270,14 @@ mod tests {
|
||||
assert!(state.peers.is_empty(), "peers must be cleared on remove --all");
|
||||
assert!(state.resync_status.is_empty(), "resync_status must be cleared on remove --all");
|
||||
|
||||
// Even if peers returned 403 (desynced account), status still reports success
|
||||
// The local side is torn down either way, but a peer that returned 403
|
||||
// (desynced account) leaves the cluster diverged — the response must
|
||||
// say so instead of reporting unqualified success (rustfs/rustfs#5963).
|
||||
let status =
|
||||
site_replication_remove_status(&["https://remote.example.com: peer/remove returned 403 Forbidden".to_string()]);
|
||||
assert_eq!(
|
||||
status.status, SITE_REPL_REMOVE_SUCCESS,
|
||||
"local remove reports success even when peer notifications fail"
|
||||
status.status, SITE_REPL_REMOVE_PARTIAL,
|
||||
"local remove must report a partial result when peer notifications fail"
|
||||
);
|
||||
assert!(
|
||||
status.err_detail.contains("403 Forbidden"),
|
||||
|
||||
@@ -27,6 +27,7 @@ Usage:
|
||||
./scripts/test/site_replication_smoke.py # up: start both + pair
|
||||
./scripts/test/site_replication_smoke.py status # process + pair status
|
||||
./scripts/test/site_replication_smoke.py smoke # bidirectional object check
|
||||
./scripts/test/site_replication_smoke.py diverge # rustfs/rustfs#5963 regression
|
||||
./scripts/test/site_replication_smoke.py logs # tail both server logs
|
||||
./scripts/test/site_replication_smoke.py down # stop both processes
|
||||
./scripts/test/site_replication_smoke.py clean # down + wipe site data
|
||||
@@ -337,11 +338,12 @@ def ensure_pair(site_a: Site, site_b: Site) -> None:
|
||||
print(f"[ok] site replication configured: {result.get('status', '')}")
|
||||
|
||||
|
||||
def remove_pair(site: Site) -> None:
|
||||
def remove_pair(site: Site) -> dict:
|
||||
status, body = admin(site, "PUT", "site-replication/remove", payload={"all": True})
|
||||
if status != 200:
|
||||
raise SystemExit(f"[fail] site-replication remove: HTTP {status} {body.decode(errors='replace')}")
|
||||
print(f"[ok] site replication removed: {body.decode(errors='replace')}")
|
||||
return json.loads(body)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -397,6 +399,112 @@ def smoke(site_a: Site, site_b: Site, timeout: float) -> None:
|
||||
print(f"[ok] bidirectional replication verified via bucket {bucket}")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Divergence regression (rustfs/rustfs#5963)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def wait_for(description: str, probe, timeout: float):
|
||||
"""Poll `probe` until it returns a truthy value; return it. SystemExit on timeout."""
|
||||
deadline = time.monotonic() + timeout
|
||||
last = None
|
||||
while time.monotonic() < deadline:
|
||||
try:
|
||||
result = probe()
|
||||
except (urllib.error.URLError, OSError, TimeoutError, SystemExit) as err:
|
||||
last = err
|
||||
result = None
|
||||
if result:
|
||||
return result
|
||||
time.sleep(1.0)
|
||||
raise SystemExit(f"[fail] {description} within {timeout:.0f}s (last: {last})")
|
||||
|
||||
|
||||
def diverge(site_a: Site, site_b: Site, binary: Path, console: bool, timeout: float) -> None:
|
||||
"""Reproduce rustfs/rustfs#5963 end to end and assert the cluster recovers.
|
||||
|
||||
Before the fix, step 7 left site-b rejecting every peer bucket-op forever:
|
||||
`pending_remove` gates `SRPeerBucketOpsHandler` ahead of `enabled()`, and a
|
||||
join never cleared it — so a *successful* re-add produced a cluster that
|
||||
reported Enabled/2-sites on both sides while replication stayed dead.
|
||||
"""
|
||||
ensure_pair(site_a, site_b)
|
||||
|
||||
# 1. Take site-a down so it cannot be told about the removal.
|
||||
print("[..] step 1: stopping site-a so it cannot be notified")
|
||||
stop_site(site_a)
|
||||
|
||||
# 2. Remove from site-b. The local teardown commits either way, but the
|
||||
# response must NOT claim unqualified success (P2-5).
|
||||
print("[..] step 2: removing site replication from site-b while site-a is down")
|
||||
status = remove_pair(site_b)
|
||||
if not status.get("errorDetail"):
|
||||
raise SystemExit(f"[fail] remove hid the unreachable peer; expected errorDetail: {json.dumps(status)}")
|
||||
if status.get("status") == "Requested site(s) were removed from cluster replication successfully.":
|
||||
raise SystemExit(f"[fail] remove reported unqualified success despite an unnotified peer: {json.dumps(status)}")
|
||||
print(f"[ok] remove reported a partial result: status={status.get('status')!r}")
|
||||
|
||||
# 3. The wedged removal must be visible on `info`, not just in status --json (P1-4).
|
||||
info_b = pair_state(site_b)
|
||||
pending = info_b.get("pendingOperation")
|
||||
if not pending or pending.get("operation") != "remove":
|
||||
raise SystemExit(f"[fail] site-b hides the wedged removal in `info`: {json.dumps(info_b, indent=2)}")
|
||||
print(f"[ok] site-b reports the wedged removal: pendingPeers={pending.get('pendingPeers')}")
|
||||
|
||||
# 4. Bring site-a back. It still believes in a healthy 2-site cluster.
|
||||
print("[..] step 4: restarting site-a")
|
||||
start_site(site_a, binary, console)
|
||||
wait_ready([site_a], timeout)
|
||||
info_a = pair_state(site_a)
|
||||
if not info_a.get("enabled"):
|
||||
raise SystemExit(f"[fail] site-a lost its own state: {json.dumps(info_a, indent=2)}")
|
||||
print("[ok] site-a still reports an enabled cluster (the divergence)")
|
||||
|
||||
# 5. A bucket created on site-a cannot reach site-b. The failure must become
|
||||
# visible on the SOURCE, which used to report a perfectly healthy cluster.
|
||||
bucket = f"sr-diverge-{uuid.uuid4().hex[:8]}"
|
||||
sig_status, body = signed_request(site_a, "PUT", f"/{bucket}")
|
||||
if sig_status != 200:
|
||||
raise SystemExit(f"[fail] create bucket {bucket} on site-a: HTTP {sig_status} {body.decode(errors='replace')}")
|
||||
print(f"[ok] created {bucket} on site-a (locally succeeds, peer push is rejected)")
|
||||
|
||||
stats = wait_for(
|
||||
"site-a did not surface the failing peer deliveries in `info`",
|
||||
lambda: pair_state(site_a).get("retryStats"),
|
||||
timeout,
|
||||
)
|
||||
print(f"[ok] site-a reports failing deliveries: pending={stats.get('pending')} failed={stats.get('failed')} "
|
||||
f"lastError={stats.get('lastError')!r}")
|
||||
|
||||
# 6. Re-add. This is the operator's natural recovery move.
|
||||
print("[..] step 6: re-adding the pair from site-a")
|
||||
peers = [
|
||||
{"name": s.name, "endpoints": s.endpoint, "accessKey": s.access_key, "secretKey": s.secret_key}
|
||||
for s in (site_a, site_b)
|
||||
]
|
||||
add_status, add_body = admin(site_a, "PUT", "site-replication/add", "replicateILMExpiry=false", peers)
|
||||
if add_status != 200:
|
||||
raise SystemExit(f"[fail] re-add: HTTP {add_status} {add_body.decode(errors='replace')}")
|
||||
print(f"[ok] re-add accepted: {add_body.decode(errors='replace')}")
|
||||
|
||||
# 7. The join must have cleared site-b's pending_remove (P0-1). Without the
|
||||
# fix this assertion is exactly what fails while everything above passes.
|
||||
info_b = pair_state(site_b)
|
||||
if info_b.get("pendingOperation"):
|
||||
raise SystemExit(
|
||||
"[fail] the join did not clear site-b's wedged removal; peer bucket-ops stay rejected forever: "
|
||||
f"{json.dumps(info_b, indent=2)}"
|
||||
)
|
||||
if not info_b.get("enabled"):
|
||||
raise SystemExit(f"[fail] site-b did not rejoin: {json.dumps(info_b, indent=2)}")
|
||||
print("[ok] site-b cleared the wedged removal and rejoined")
|
||||
|
||||
# 8. The symptom the issue actually reported: replication works again.
|
||||
print("[..] step 8: verifying replication actually flows again")
|
||||
smoke(site_a, site_b, timeout)
|
||||
print("[ok] rustfs/rustfs#5963 regression passed")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Commands
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -464,7 +572,7 @@ def main() -> None:
|
||||
"command",
|
||||
nargs="?",
|
||||
default="up",
|
||||
choices=["up", "down", "restart", "status", "logs", "smoke", "info", "remove", "clean"],
|
||||
choices=["up", "down", "restart", "status", "logs", "smoke", "diverge", "info", "remove", "clean"],
|
||||
)
|
||||
parser.add_argument("--port-a", type=int, default=9000, help="site A S3 port (default: %(default)s)")
|
||||
parser.add_argument("--port-b", type=int, default=9020, help="site B S3 port (default: %(default)s)")
|
||||
@@ -495,6 +603,8 @@ def main() -> None:
|
||||
cmd_logs(sites, args.lines)
|
||||
elif args.command == "smoke":
|
||||
smoke(site_a, site_b, args.timeout)
|
||||
elif args.command == "diverge":
|
||||
diverge(site_a, site_b, args.binary, args.console, args.timeout)
|
||||
elif args.command == "info":
|
||||
print(json.dumps(pair_state(site_a), indent=2, ensure_ascii=False))
|
||||
elif args.command == "remove":
|
||||
|
||||
Reference in New Issue
Block a user