diff --git a/crates/ecstore/src/bucket/bucket_target_sys.rs b/crates/ecstore/src/bucket/bucket_target_sys.rs index dca41cf72..db4e80f29 100644 --- a/crates/ecstore/src/bucket/bucket_target_sys.rs +++ b/crates/ecstore/src/bucket/bucket_target_sys.rs @@ -59,7 +59,7 @@ use rustfs_utils::http::{ insert_header, }; use serde::{Deserialize, Serialize}; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::error::Error; use std::fmt; use std::str::FromStr as _; @@ -376,6 +376,11 @@ pub struct BucketTargetSys { /// [`SsecPassthroughCapability`]; reset alongside `arn_remotes_map`. ssec_passthrough_map: Arc>>, pub targets_map: Arc>>>, + /// Buckets whose persisted `bucket-targets.json` exists but cannot be + /// decoded (rustfs/backlog#2282). Written under the bucket's update mutex + /// alongside `targets_map`, and read before it so an unreadable + /// configuration surfaces as a typed error instead of an empty target set. + unreadable_targets: Arc>>, pub h_mutex: Arc>>, target_h_mutex: Arc>>, pub hc_client: Arc, @@ -419,6 +424,7 @@ impl BucketTargetSys { arn_remotes_map: Arc::new(RwLock::new(HashMap::new())), ssec_passthrough_map: Arc::new(RwLock::new(HashMap::new())), targets_map: Arc::new(RwLock::new(HashMap::new())), + unreadable_targets: Arc::new(RwLock::new(HashSet::new())), h_mutex: Arc::new(RwLock::new(HashMap::new())), target_h_mutex: Arc::new(RwLock::new(HashMap::new())), hc_client: Arc::new(build_health_check_client()), @@ -628,30 +634,40 @@ impl BucketTargetSys { health_map.clone() } - pub async fn list_targets(&self, bucket: &str, arn_type: &str) -> Vec { + /// Targets of one bucket, or of every bucket when `bucket` is empty. + /// + /// A bucket that simply has no targets yields an empty list; a bucket + /// whose persisted configuration cannot be decoded is an error, so an + /// admin listing reports the fault instead of an empty list that reads as + /// "replication is not configured" (rustfs/backlog#2282). + pub async fn list_targets(&self, bucket: &str, arn_type: &str) -> Result, BucketTargetError> { let health_stats = self.target_health_stats().await; let mut targets = Vec::new(); if !bucket.is_empty() { - if let Ok(bucket_targets) = self.list_bucket_targets(bucket).await { - for mut target in bucket_targets.targets { - if arn_type.is_empty() || target.target_type.to_string() == arn_type { - if let Some(health) = health_stats.get(&target.arn) { - target.total_downtime = health.offline_duration; - target.online = health.online; - target.last_online = health.last_online; - target.latency = target::LatencyStat { - curr: health.latency.curr, - avg: health.latency.avg, - max: health.latency.peak, - }; - target.offline_count = health.offline_count; + match self.list_bucket_targets(bucket).await { + Ok(bucket_targets) => { + for mut target in bucket_targets.targets { + if arn_type.is_empty() || target.target_type.to_string() == arn_type { + if let Some(health) = health_stats.get(&target.arn) { + target.total_downtime = health.offline_duration; + target.online = health.online; + target.last_online = health.last_online; + target.latency = target::LatencyStat { + curr: health.latency.curr, + avg: health.latency.avg, + max: health.latency.peak, + }; + target.offline_count = health.offline_count; + } + targets.push(target); } - targets.push(target); } } + Err(BucketTargetError::BucketRemoteTargetNotFound { .. }) => {} + Err(err) => return Err(err), } - return targets; + return Ok(targets); } let targets_map = self.targets_map.read().await; @@ -674,10 +690,16 @@ impl BucketTargetSys { } } - targets + Ok(targets) } pub async fn list_bucket_targets(&self, bucket: &str) -> Result { + if self.unreadable_targets.read().await.contains(bucket) { + return Err(BucketTargetError::BucketRemoteTargetsUnreadable { + bucket: bucket.to_string(), + }); + } + let targets_map = self.targets_map.read().await; if let Some(targets) = targets_map.get(bucket) { Ok(BucketTargets { @@ -690,13 +712,30 @@ impl BucketTargetSys { } } + /// Record that this bucket's persisted targets configuration exists but + /// cannot be decoded (rustfs/backlog#2282). + /// + /// Any snapshot published from an earlier readable load is deliberately + /// left in place: withdrawing it would produce exactly the silent "no + /// targets configured" state this marker exists to prevent. The marker is + /// cleared by the next successful publish, which is what makes a repaired + /// configuration take effect without a restart. + pub async fn mark_targets_unreadable(&self, bucket: &str) { + let update_mutex = self.target_update_mutex(bucket).await; + let _update_guard = update_mutex.lock().await; + + self.unreadable_targets.write().await.insert(bucket.to_string()); + } + pub async fn delete(&self, bucket: &str) { let update_mutex = self.target_update_mutex(bucket).await; let _update_guard = update_mutex.lock().await; - // Lock order: targets_map, then arn_remotes_map, then target_h_mutex, - // then ssec_passthrough_map (always last; also taken standalone by the - // capability accessors). + // Lock order: unreadable_targets, then targets_map, then + // arn_remotes_map, then target_h_mutex, then ssec_passthrough_map + // (always last; also taken standalone by the capability accessors). + self.unreadable_targets.write().await.remove(bucket); + let mut targets_map = self.targets_map.write().await; let mut arn_remotes_map = self.arn_remotes_map.write().await; let mut health_map = self.target_h_mutex.write().await; @@ -1093,6 +1132,11 @@ impl BucketTargetSys { /// Keeping persisted-config reads under the same mutex prevents a stale /// reload from overwriting a concurrent credential rotation. async fn update_all_targets_locked(&self, bucket: &str, targets: Option<&BucketTargets>) { + // Reaching here means the persisted configuration decoded, so the + // unreadable marker (if any) is stale. Cleared before the maps below + // so `unreadable_targets` stays the outermost of this module's locks. + self.unreadable_targets.write().await.remove(bucket); + let mut clients = Vec::new(); if let Some(new_targets) = targets { for target in &new_targets.targets { @@ -1100,9 +1144,9 @@ impl BucketTargetSys { } } - // Lock order: targets_map, then arn_remotes_map, then target_h_mutex, - // then ssec_passthrough_map (always last; also taken standalone by the - // capability accessors). + // Lock order: unreadable_targets (above), then targets_map, then + // arn_remotes_map, then target_h_mutex, then ssec_passthrough_map + // (always last; also taken standalone by the capability accessors). let mut targets_map = self.targets_map.write().await; let mut arn_remotes_map = self.arn_remotes_map.write().await; let mut health_map = self.target_h_mutex.write().await; @@ -1161,6 +1205,11 @@ impl BucketTargetSys { } pub async fn set(&self, bucket: &str, meta: &BucketMetadata) { + if meta.bucket_targets_unreadable() { + self.mark_targets_unreadable(bucket).await; + return; + } + let Some(config) = &meta.bucket_target_config else { return; }; @@ -2276,6 +2325,13 @@ pub enum BucketTargetError { BucketRemoteTargetNotFound { bucket: String, }, + /// The bucket's persisted targets configuration exists but cannot be + /// decoded. Distinct from `BucketRemoteTargetNotFound`, which means the + /// bucket genuinely has no targets: callers must not degrade this one to + /// an empty target set (rustfs/backlog#2282). + BucketRemoteTargetsUnreadable { + bucket: String, + }, BucketRemoteArnTypeInvalid { bucket: String, }, @@ -2309,6 +2365,9 @@ impl fmt::Display for BucketTargetError { BucketTargetError::BucketRemoteTargetNotFound { bucket } => { write!(f, "Remote target not found for bucket: {bucket}") } + BucketTargetError::BucketRemoteTargetsUnreadable { bucket } => { + write!(f, "Persisted replication target configuration is unreadable for bucket: {bucket}") + } BucketTargetError::BucketRemoteArnTypeInvalid { bucket } => { write!(f, "Invalid ARN type for bucket: {bucket}") } @@ -3256,7 +3315,7 @@ mod tests { }], ); - let targets = sys.list_targets("", "").await; + let targets = sys.list_targets("", "").await.expect("listing every bucket's targets"); assert_eq!(targets.len(), 1); assert!(!targets[0].online); diff --git a/crates/ecstore/src/bucket/metadata.rs b/crates/ecstore/src/bucket/metadata.rs index bd10f0f63..41b2afdc5 100644 --- a/crates/ecstore/src/bucket/metadata.rs +++ b/crates/ecstore/src/bucket/metadata.rs @@ -477,6 +477,18 @@ impl BucketMetadata { !self.table_bucket_config_json.is_empty() } + /// `bucket-targets.json` is stored for this bucket but this build cannot + /// decode it. + /// + /// Keeps "no replication targets configured" and "the target + /// configuration cannot be read" apart, the same distinction the + /// `fabricated` marker draws for the bucket metadata as a whole. Only + /// meaningful after [`Self::parse_all_configs`] has run; readers must fail + /// closed on `true` instead of serving an empty target set. + pub fn bucket_targets_unreadable(&self) -> bool { + !self.bucket_targets_config_json.is_empty() && self.bucket_target_config.is_none() + } + /// Parsed per-bucket durability override, if a valid one is stored. /// /// Absent/empty/unparsable payloads all mean "no override" (the bucket @@ -964,7 +976,32 @@ impl BucketMetadata { Ok(()) } - fn parse_all_configs(&mut self) -> Result<()> { + /// Decode every stored sub-configuration into its typed field. + /// + /// A decode failure never fails the whole load: this runs on every bucket + /// metadata read, including startup and peer reload, so one bucket's + /// corrupt sub-configuration must not make the bucket — or the node — + /// unloadable. Instead the failure is *retained*: the raw bytes stay + /// untouched and the typed field stays `None`, so `!raw.is_empty() && + /// typed.is_none()` is the durable "exists but cannot be read" signal that + /// each accessor keys off. Which accessors must fail closed on it: + /// + /// | Config | Verdict | + /// |---|---| + /// | policy | Fails closed: `get_bucket_policy` re-parses the raw JSON and propagates the error; `get_bucket_policy_raw` returns the stored bytes. | + /// | object lock | Fails closed in `object_lock_config_state_from_authoritative_metadata`; a retention decision may never be taken on a guess. | + /// | versioning | Fails closed in `get_versioning_config`; guessing Unversioned would make delete markers and version ids diverge from what is on disk. | + /// | replication | Fails closed in `get_replication_config`. | + /// | bucket targets | Fails closed in `get_bucket_targets_config`, and `sync_bucket_target_sys` marks the bucket unreadable in `BucketTargetSys` instead of publishing an empty target set (rustfs/backlog#2282). | + /// | encryption | Fails closed in `get_sse_config`: degrading to "no default encryption" stores plaintext objects the operator required to be encrypted. | + /// | public access block | Fails closed in `get_public_access_block_config`: degrading grants the anonymous access the operator asked to block. | + /// | quota | Fails closed in `get_quota_config`; the enforcement path in `quota::checker` already re-parses the raw JSON and refuses on error. | + /// | lifecycle | Safe to degrade: no rules means no expiration and no transition, so nothing is deleted or moved on the strength of an unreadable rule set. The bucket keeps serving reads and writes. | + /// | notification | Safe to degrade: events are an outbound side channel; no consumer draws a durability or authorization conclusion from their absence. | + /// | tagging | Safe to degrade: bucket tags are cost-allocation labels here; object-level tag conditions come from object metadata, not this blob. | + /// | CORS | Safe to degrade: an absent CORS configuration rejects cross-origin browser requests, which is already the restrictive direction. | + /// | logging, website, accelerate, request payment, bucket ACL | Safe to degrade: each only shapes an optional response or an optional side channel, and none of them authorizes an action or decides whether data is retained. | + pub(super) fn parse_all_configs(&mut self) -> Result<()> { if let Err(e) = self.parse_policy_config() { tracing::warn!( event = "bucket_metadata_parse_failed", @@ -1088,20 +1125,26 @@ impl BucketMetadata { "Failed to parse bucket metadata config" ); } + // A stored targets blob that cannot be decoded must not collapse into + // the empty target set: that is indistinguishable from "no replication + // configured", so replication stops and no caller ever sees an error + // (rustfs/backlog#2282). Leaving the typed field `None` while the raw + // bytes stay non-empty is the retained parse failure every targets + // reader keys off; the bytes are preserved so the configuration is + // still recoverable. + self.bucket_target_config = None; if !self.bucket_targets_config_json.is_empty() { - if let Err(e) = serde_json::from_slice::(&self.bucket_targets_config_json) - .map(|t| self.bucket_target_config = Some(t)) - { - tracing::warn!( + match serde_json::from_slice::(&self.bucket_targets_config_json) { + Ok(targets) => self.bucket_target_config = Some(targets), + Err(e) => tracing::error!( event = "bucket_metadata_parse_failed", component = "ecstore", subsystem = "bucket_metadata", bucket = %self.name, config = "bucket_targets", error = %e, - "Failed to parse bucket metadata config" - ); - self.bucket_target_config = Some(BucketTargets::default()); + "Bucket replication targets are unreadable; replication for this bucket fails closed" + ), } } else { self.bucket_target_config = Some(BucketTargets::default()); @@ -1535,6 +1578,117 @@ mod test { assert_eq!(bucket_targets.targets[0].target_bucket, "target-bucket"); } + /// rustfs/backlog#2282: a stored targets blob this build cannot decode + /// must not become the empty target set, and must stay distinguishable + /// from a bucket that never configured a target. + #[test] + fn unreadable_bucket_targets_never_degrade_to_an_empty_target_set() { + let truncated = br#"{"targets":[{"endpoint":"s3.example.com","#.to_vec(); + let mut corrupt = BucketMetadata::new("corrupt-targets"); + corrupt.bucket_targets_config_json = truncated.clone(); + + corrupt + .parse_all_configs() + .expect("one unreadable sub-config must not fail the whole metadata load"); + + assert!( + corrupt.bucket_target_config.is_none(), + "an undecodable targets blob must not produce a target set at all" + ); + assert!(corrupt.bucket_targets_unreadable()); + assert_eq!( + corrupt.bucket_targets_config_json, truncated, + "the raw bytes must survive so the configuration stays recoverable" + ); + + // The genuinely-absent case is unchanged, and the two now diverge. + let mut absent = BucketMetadata::new("no-targets"); + absent.parse_all_configs().expect("absent targets parse"); + assert!( + absent.bucket_target_config.as_ref().is_some_and(BucketTargets::is_empty), + "a bucket that configured no target still reads as an empty target set" + ); + assert!(!absent.bucket_targets_unreadable()); + } + + /// `Credentials` carries no struct-level `serde(default)`, so one target + /// missing `secretKey` is a hard parse error for the whole document. That + /// must surface as "unreadable", never as "no targets configured". + #[test] + fn bucket_targets_missing_secret_key_are_unreadable_not_empty() { + let mut bm = BucketMetadata::new("missing-secret-key"); + bm.bucket_targets_config_json = br#"{"targets":[{"endpoint":"s3.example.com","targetbucket":"remote","arn":"arn:rustfs:replication:us-east-1:src:1","credentials":{"accessKey":"AKIAEXAMPLE"}}]}"#.to_vec(); + + bm.parse_all_configs() + .expect("a rejected targets document must not fail the whole metadata load"); + + assert!( + bm.bucket_targets_unreadable(), + "a targets document rejected for a missing secretKey is unreadable, not empty" + ); + assert!(bm.bucket_target_config.is_none()); + } + + /// The invariant every branch of `parse_all_configs` shares: a stored but + /// undecodable payload keeps its raw bytes and leaves the typed field + /// `None`, so no branch fabricates a value. What a reader may then do with + /// that state is decided per config; see the table on `parse_all_configs`. + #[test] + fn every_config_branch_retains_its_parse_failure_instead_of_defaulting() { + let malformed_xml = b">) { } async fn sync_bucket_target_sys(bucket: &str, bm: &BucketMetadata) { + if bm.bucket_targets_unreadable() { + // "The configuration cannot be read" is not "no targets configured". + // Publishing an empty snapshot here is what silently stopped + // replication (rustfs/backlog#2282): mark the bucket instead, so every + // targets reader gets a typed error, and leave any snapshot from an + // earlier readable load in place rather than withdrawing it. + BucketTargetSys::get().mark_targets_unreadable(bucket).await; + return; + } + BucketTargetSys::get() .update_all_targets(bucket, bm.bucket_target_config.as_ref()) .await; @@ -2118,7 +2128,9 @@ impl BucketMetadataSys { pub async fn get_public_access_block_config(&self, bucket: &str) -> Result<(PublicAccessBlockConfiguration, OffsetDateTime)> { let (bm, _) = self.get_config(bucket).await?; - if let Some(config) = &bm.public_access_block_config { + if !bm.public_access_block_config_xml.is_empty() && bm.public_access_block_config.is_none() { + Err(Error::other("persisted bucket public access block configuration is invalid")) + } else if let Some(config) = &bm.public_access_block_config { Ok((config.clone(), bm.public_access_block_config_updated_at)) } else { Err(Error::ConfigNotFound) @@ -2429,7 +2441,9 @@ impl BucketMetadataSys { pub async fn get_sse_config(&self, bucket: &str) -> Result<(ServerSideEncryptionConfiguration, OffsetDateTime)> { let (bm, _) = self.get_config(bucket).await?; - if let Some(config) = &bm.sse_config { + if !bm.encryption_config_xml.is_empty() && bm.sse_config.is_none() { + Err(Error::other("persisted bucket encryption configuration is invalid")) + } else if let Some(config) = &bm.sse_config { Ok((config.clone(), bm.encryption_config_updated_at)) } else { Err(Error::ConfigNotFound) @@ -2500,7 +2514,9 @@ impl BucketMetadataSys { pub async fn get_quota_config(&self, bucket: &str) -> Result<(BucketQuota, OffsetDateTime)> { let (bm, _) = self.get_config(bucket).await?; - if let Some(config) = &bm.quota_config { + if !bm.quota_config_json.is_empty() && bm.quota_config.is_none() { + Err(Error::other("persisted bucket quota configuration is invalid")) + } else if let Some(config) = &bm.quota_config { Ok((config.clone(), bm.quota_config_updated_at)) } else { Err(Error::ConfigNotFound) @@ -2522,7 +2538,9 @@ impl BucketMetadataSys { pub async fn get_bucket_targets_config(&self, bucket: &str) -> Result { let (bm, _) = self.get_config(bucket).await?; - if let Some(config) = &bm.bucket_target_config { + if bm.bucket_targets_unreadable() { + Err(Error::other("persisted bucket replication target configuration is invalid")) + } else if let Some(config) = &bm.bucket_target_config { Ok(config.clone()) } else { Err(Error::ConfigNotFound) @@ -2593,6 +2611,7 @@ pub(crate) mod test_support { mod tests { use super::test_support::isolated_store_over_temp_disks; use super::*; + use crate::bucket::bucket_target_sys::BucketTargetError; use crate::bucket::metadata::{ BUCKET_ACCELERATE_CONFIG, BUCKET_CORS_CONFIG, BUCKET_LIFECYCLE_CONFIG, BUCKET_LOGGING_CONFIG, BUCKET_NOTIFICATION_CONFIG, BUCKET_POLICY_CONFIG, BUCKET_PUBLIC_ACCESS_BLOCK_CONFIG, BUCKET_REPLICATION_CONFIG, BUCKET_REQUEST_PAYMENT_CONFIG, @@ -2788,6 +2807,36 @@ mod tests { ); } + /// The `parse_all_configs` audit (rustfs/backlog#2282): every accessor + /// whose configuration grants something — plaintext storage, anonymous + /// access, capacity, replication targets — reports a corrupt payload as + /// invalid rather than as absent, because "absent" is what grants it. + #[tokio::test] + async fn malformed_permissive_configs_are_not_reported_as_absent() { + let (_dirs, ecstore) = isolated_store_over_temp_disks().await; + let sys = BucketMetadataSys::new(ecstore); + let bucket = "malformed-permissive-config"; + let mut metadata = BucketMetadata::new(bucket); + metadata.encryption_config_xml = b" Some(targets), + // A bucket whose persisted target configuration cannot be decoded has + // an unknown target set, not an empty one: scheduling against `None` + // here would drop every heal for it without a trace + // (rustfs/backlog#2282). Report it missed so the object is retried + // once the configuration is readable again. + Err(BucketTargetError::BucketRemoteTargetsUnreadable { .. }) => { + warn!( + event = EVENT_REPLICATION_CONFIG_LOOKUP_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + bucket, + reason = "target_config_unreadable", + "Bucket replication targets are unreadable; replication heal queue fails closed" + ); + + return ReplicationQueueAdmission::Missed; + } Err(err) => { debug!( event = EVENT_REPLICATION_CONFIG_LOOKUP_SKIPPED, diff --git a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs index a5cbf6b34..f7531319e 100644 --- a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs @@ -15,7 +15,8 @@ use std::collections::HashMap; use std::sync::Arc; -use crate::bucket::bucket_target_sys::{BucketTargetError, BucketTargetSys}; +pub(crate) use crate::bucket::bucket_target_sys::BucketTargetError; +use crate::bucket::bucket_target_sys::BucketTargetSys; use aws_sdk_s3::operation::head_object::HeadObjectOutput; use aws_sdk_s3::types::{ObjectLockLegalHoldStatus, ObjectLockRetentionMode}; use http::HeaderMap; diff --git a/rustfs/src/admin/handlers/replication.rs b/rustfs/src/admin/handlers/replication.rs index b2b2b80dc..b7bf5ef78 100644 --- a/rustfs/src/admin/handlers/replication.rs +++ b/rustfs/src/admin/handlers/replication.rs @@ -121,6 +121,11 @@ fn map_bucket_target_error(err: BucketTargetError) -> S3Error { | BucketTargetError::BucketRemoteRemoveDisallowed { .. } => { S3Error::with_message(S3ErrorCode::InvalidRequest, err.to_string()) } + // A stored target configuration this node cannot decode is a + // server-side data fault, not a bad request (rustfs/backlog#2282). + BucketTargetError::BucketRemoteTargetsUnreadable { .. } => { + S3Error::with_message(S3ErrorCode::InternalError, err.to_string()) + } BucketTargetError::Io(io_err) => S3Error::with_message(S3ErrorCode::InternalError, io_err.to_string()), } } @@ -753,7 +758,12 @@ impl Operation for ListRemoteTargetHandler { .map_err(ApiError::from)?; let sys = BucketTargetSys::get(); - let targets = sys.list_targets(bucket, "").await; + // An unreadable targets configuration must not be reported as an + // empty target list (rustfs/backlog#2282). + let targets = sys.list_targets(bucket, "").await.map_err(|e| { + error!("list remote targets failed: {}", e); + map_bucket_target_error(e) + })?; let targets: Vec<_> = targets .iter()