fix(ecstore): refuse writes on unreadable bucket configs with 503 (#7768)

Close the remaining paths where a stored bucket sub-configuration whose
bytes cannot be parsed still read as absent (rustfs/backlog#1734, slice 2):

- Object writes resolve versioning through BucketVersioningSys::get_for_write;
  RUSTFS_BUCKET_CONFIG_PARSE_MODE=strict refuses them, the default permissive
  mode keeps the historical unversioned write and records it.
- Versioning, Object Lock, encryption, public access block and notification
  getters and the delete-time versioning check return the typed
  UnreadableBucketConfig refusal, which ApiError maps to 503 naming the
  bucket, config and stored length. StorageError::clone keeps it typed.
- Object Lock checks stay on when the lock config is unreadable.
- Notification setup isolates an unreadable config to its bucket instead of
  clearing its rules; GetBucketNotificationConfiguration reports it.
- Add rustfs_bucket_metadata_parse_failed_total and
  rustfs_bucket_metadata_unparsable_current, an invalid-mode startup check,
  and the s3gate-parse-strict compat register entry.
This commit is contained in:
Chris
2026-09-14 01:40:29 +08:00
committed by GitHub
parent 7111f8b44d
commit 1066a08aca
25 changed files with 739 additions and 91 deletions
+10
View File
@@ -116,6 +116,16 @@ pub const ENV_OBJECT_GET_SKIP_BITROT_VERIFY: &str = "RUSTFS_OBJECT_GET_SKIP_BITR
/// Default: bitrot verification is enabled on GetObject reads (do not skip).
pub const DEFAULT_OBJECT_GET_SKIP_BITROT_VERIFY: bool = false;
/// How object writes treat a bucket whose stored versioning configuration
/// cannot be parsed: `permissive` writes as if unversioned (the historical
/// behavior, recorded by metrics and an error log) and `strict` refuses the
/// write with 503. Paths that already refuse an unreadable configuration do
/// so in both modes. Any other value fails startup.
pub const ENV_BUCKET_CONFIG_PARSE_MODE: &str = "RUSTFS_BUCKET_CONFIG_PARSE_MODE";
/// Default bucket config parse mode.
pub const DEFAULT_BUCKET_CONFIG_PARSE_MODE: &str = "permissive";
/// Request writing the complete remote-tier version state into object metadata.
///
/// This remains ineffective until
+7 -1
View File
@@ -170,11 +170,17 @@ pub mod bucket {
BUCKET_TABLE_CATALOG_TABLE_BUCKETS_PREFIX, BUCKET_TABLE_CONFIG, BUCKET_TABLE_RESERVED_PREFIX, BUCKET_TAGGING_CONFIG,
BUCKET_TARGETS_FILE, BUCKET_VERSIONING_CONFIG, BUCKET_WEBSITE_CONFIG, BucketMetadata, ConfigState,
OBJECT_LOCK_CONFIG, UnreadableBucketConfig, is_unreadable_config_error, load_bucket_metadata,
table_catalog_path_hash,
table_catalog_path_hash, unreadable_config_refusal,
};
pub use crate::bucket::metadata::{BUCKET_DURABILITY_CONFIG, BUCKET_ON_DEMAND_MIGRATION_CONFIG};
}
pub mod config_parse_mode {
pub use crate::bucket::config_parse_mode::{
BucketConfigParseMode, bucket_config_parse_mode, validate_bucket_config_parse_mode_env,
};
}
pub mod durability {
pub use crate::bucket::durability::{
BUCKET_DURABILITY_MODE_NONE, BUCKET_DURABILITY_MODE_RELAXED, BUCKET_DURABILITY_MODE_STRICT, BucketDurabilityConfig,
@@ -0,0 +1,241 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//! Handling of stored bucket sub-configurations whose bytes cannot be parsed
//! (rustfs/backlog#1734): the rollout mode for paths that historically read
//! them as absent, and the metrics that make them visible.
//!
//! The mode only governs those historical degrade paths. Paths that already
//! refuse an unreadable config (the typed getters, the read-modify-write
//! guard, Object Lock and default-encryption decisions, delete-time
//! versioning) refuse in every mode.
use rustfs_config::{DEFAULT_BUCKET_CONFIG_PARSE_MODE, ENV_BUCKET_CONFIG_PARSE_MODE};
use std::collections::HashMap;
use std::sync::{LazyLock, Mutex, OnceLock};
use tracing::error;
/// Counter of stored XML sub-configurations that failed to parse, labeled
/// `bucket`, `config` and `mode`. Increments on every metadata parse, so it
/// must read zero fleet-wide before the default mode flips to strict.
pub const METRIC_BUCKET_METADATA_PARSE_FAILED_TOTAL: &str = "rustfs_bucket_metadata_parse_failed_total";
/// Gauge of buckets whose most recently parsed metadata holds an unreadable
/// XML sub-configuration, labeled `config`. The counter only moves when a
/// parse runs; this reflects the current state between parses.
pub const METRIC_BUCKET_METADATA_UNPARSABLE_CURRENT: &str = "rustfs_bucket_metadata_unparsable_current";
const LOG_COMPONENT: &str = "ecstore";
const LOG_SUBSYSTEM: &str = "bucket_metadata";
const EVENT_CONFIG_UNREADABLE: &str = "bucket_metadata_config_unreadable";
const EVENT_PARSE_MODE_INVALID: &str = "bucket_config_parse_mode_invalid";
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BucketConfigParseMode {
/// Historical degrade paths keep reading an unreadable config as absent;
/// the metrics and an error log record every occurrence.
Permissive,
/// Historical degrade paths refuse instead.
Strict,
}
impl BucketConfigParseMode {
pub fn as_str(self) -> &'static str {
match self {
Self::Permissive => "permissive",
Self::Strict => "strict",
}
}
/// Parse a configured value; unset or blank selects the default. An
/// unknown value is an error rather than a silent fallback.
pub fn parse(value: Option<&str>) -> Result<Self, String> {
let value = value
.map(str::trim)
.filter(|value| !value.is_empty())
.unwrap_or(DEFAULT_BUCKET_CONFIG_PARSE_MODE);
match value.to_ascii_lowercase().as_str() {
"permissive" => Ok(Self::Permissive),
"strict" => Ok(Self::Strict),
_ => Err(format!(
"invalid {ENV_BUCKET_CONFIG_PARSE_MODE} value {value:?}; expected permissive or strict"
)),
}
}
}
/// Validate the configured mode; startup calls this so an invalid value fails
/// the node instead of being guessed at.
pub fn validate_bucket_config_parse_mode_env() -> Result<BucketConfigParseMode, String> {
BucketConfigParseMode::parse(rustfs_utils::get_env_opt_str(ENV_BUCKET_CONFIG_PARSE_MODE).as_deref())
}
/// The process-wide mode. Startup has already rejected an invalid value; if
/// this is reached without that validation, an invalid value selects strict,
/// never permissive.
pub fn bucket_config_parse_mode() -> BucketConfigParseMode {
static MODE: OnceLock<BucketConfigParseMode> = OnceLock::new();
*MODE.get_or_init(|| {
validate_bucket_config_parse_mode_env().unwrap_or_else(|err| {
error!(
event = EVENT_PARSE_MODE_INVALID,
component = LOG_COMPONENT,
subsystem = LOG_SUBSYSTEM,
error = %err,
"Invalid bucket config parse mode; using strict"
);
BucketConfigParseMode::Strict
})
})
}
/// Which buckets currently hold which unreadable XML configs, so the gauge
/// reports buckets rather than parse events.
#[derive(Debug, Default)]
struct UnreadableConfigTracker {
by_bucket: HashMap<String, Vec<&'static str>>,
}
impl UnreadableConfigTracker {
/// Replace `bucket`'s unreadable set and return the current bucket count
/// of every config whose count may have changed.
fn update(&mut self, bucket: &str, configs: Vec<&'static str>) -> Vec<(&'static str, usize)> {
let previous = if configs.is_empty() {
self.by_bucket.remove(bucket).unwrap_or_default()
} else {
self.by_bucket.insert(bucket.to_string(), configs.clone()).unwrap_or_default()
};
let mut touched = previous;
touched.extend(configs);
touched.sort_unstable();
touched.dedup();
touched
.into_iter()
.map(|config| (config, self.by_bucket.values().filter(|set| set.contains(&config)).count()))
.collect()
}
}
static TRACKER: LazyLock<Mutex<UnreadableConfigTracker>> = LazyLock::new(Default::default);
fn publish_gauges(bucket: &str, configs: Vec<&'static str>) {
let changed = TRACKER
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.update(bucket, configs);
for (config, count) in changed {
metrics::gauge!(METRIC_BUCKET_METADATA_UNPARSABLE_CURRENT, "config" => config).set(count as f64);
}
}
/// Record the outcome of one metadata parse: every unreadable XML config
/// (config file, stored byte length) counts once, logs at error level, and
/// replaces the bucket's entry in the current-state gauge.
pub(crate) fn record_bucket_config_parse_state(bucket: &str, unreadable: &[(&'static str, usize)]) {
if !unreadable.is_empty() {
let mode = bucket_config_parse_mode();
for (config, _) in unreadable {
metrics::counter!(
METRIC_BUCKET_METADATA_PARSE_FAILED_TOTAL,
"bucket" => bucket.to_string(),
"config" => *config,
"mode" => mode.as_str()
)
.increment(1);
}
error!(
event = EVENT_CONFIG_UNREADABLE,
component = LOG_COMPONENT,
subsystem = LOG_SUBSYSTEM,
bucket = %bucket,
configs = ?unreadable,
mode = mode.as_str(),
"Stored bucket configuration cannot be parsed"
);
}
publish_gauges(bucket, unreadable.iter().map(|(config, _)| *config).collect());
}
/// Drop a deleted bucket from the current-state gauge.
pub(crate) fn forget_bucket_config_parse_state(bucket: &str) {
publish_gauges(bucket, Vec::new());
}
#[cfg(test)]
mod tests {
use super::*;
use crate::bucket::metadata::{BUCKET_TAGGING_CONFIG, BUCKET_VERSIONING_CONFIG};
#[test]
fn parse_mode_defaults_to_permissive_and_rejects_unknown_values() {
assert_eq!(BucketConfigParseMode::parse(None), Ok(BucketConfigParseMode::Permissive));
assert_eq!(BucketConfigParseMode::parse(Some(" ")), Ok(BucketConfigParseMode::Permissive));
assert_eq!(BucketConfigParseMode::parse(Some("permissive")), Ok(BucketConfigParseMode::Permissive));
assert_eq!(BucketConfigParseMode::parse(Some(" Strict ")), Ok(BucketConfigParseMode::Strict));
let err = BucketConfigParseMode::parse(Some("lenient")).expect_err("an unknown mode must not be guessed");
assert!(err.contains(ENV_BUCKET_CONFIG_PARSE_MODE), "{err}");
assert!(err.contains("permissive") && err.contains("strict"), "must list the valid values: {err}");
}
#[test]
fn tracker_counts_buckets_not_parse_events() {
let mut tracker = UnreadableConfigTracker::default();
assert_eq!(tracker.update("a", vec![BUCKET_VERSIONING_CONFIG]), vec![(BUCKET_VERSIONING_CONFIG, 1)]);
// Re-parsing the same state does not double count.
assert_eq!(tracker.update("a", vec![BUCKET_VERSIONING_CONFIG]), vec![(BUCKET_VERSIONING_CONFIG, 1)]);
assert_eq!(tracker.update("b", vec![BUCKET_VERSIONING_CONFIG]), vec![(BUCKET_VERSIONING_CONFIG, 2)]);
// A repaired bucket releases its configs; the changed one is reported.
let mut changed = tracker.update("a", vec![BUCKET_TAGGING_CONFIG]);
changed.sort_unstable();
let mut expected = vec![(BUCKET_TAGGING_CONFIG, 1), (BUCKET_VERSIONING_CONFIG, 1)];
expected.sort_unstable();
assert_eq!(changed, expected);
assert_eq!(tracker.update("b", Vec::new()), vec![(BUCKET_VERSIONING_CONFIG, 0)]);
assert_eq!(tracker.update("never-unreadable", Vec::new()), Vec::new());
}
#[test]
fn every_unreadable_config_increments_the_labeled_counter() {
let recorder = metrics_util::debugging::DebuggingRecorder::new();
let snapshotter = recorder.snapshotter();
metrics::with_local_recorder(&recorder, || {
record_bucket_config_parse_state("metric-bucket", &[(BUCKET_VERSIONING_CONFIG, 12), (BUCKET_TAGGING_CONFIG, 3)]);
record_bucket_config_parse_state("metric-bucket", &[(BUCKET_VERSIONING_CONFIG, 12)]);
record_bucket_config_parse_state("metric-clean-bucket", &[]);
});
let mut versioning = 0;
let mut tagging = 0;
for (composite, _, _, value) in snapshotter.snapshot().into_vec() {
if composite.key().name() != METRIC_BUCKET_METADATA_PARSE_FAILED_TOTAL {
continue;
}
let labels: HashMap<_, _> = composite.key().labels().map(|l| (l.key(), l.value())).collect();
assert_eq!(labels.get("bucket"), Some(&"metric-bucket"));
assert_eq!(labels.get("mode"), Some(&bucket_config_parse_mode().as_str()));
let metrics_util::debugging::DebugValue::Counter(count) = value else {
panic!("parse failures must be a counter");
};
match labels.get("config") {
Some(&config) if config == BUCKET_VERSIONING_CONFIG => versioning += count,
Some(&config) if config == BUCKET_TAGGING_CONFIG => tagging += count,
other => panic!("unexpected config label {other:?}"),
}
}
assert_eq!((versioning, tagging), (2, 1));
}
}
@@ -12621,9 +12621,8 @@ mod tests {
.await
.expect_err("malformed Object Lock metadata must reject lifecycle config resolution");
assert!(
exact_error
.to_string()
.contains("persisted bucket Object Lock configuration is invalid")
crate::bucket::metadata::is_unreadable_config_error(&exact_error),
"malformed Object Lock metadata must surface as the typed unreadable-config refusal: {exact_error}"
);
let runtime_state = install_unconsumed_runtime_expiry_worker(&ecstore, 1).await;
+82 -13
View File
@@ -278,18 +278,20 @@ pub const BUCKET_TABLE_CATALOG_TABLE_BUCKETS_PREFIX: &str = "table-buckets";
/// Refusal to act on a stored sub-configuration whose bytes exist but cannot
/// be parsed. Carried inside [`Error::other`] so callers can tell it apart
/// from a storage fault with [`is_unreadable_config_error`].
#[derive(Debug)]
#[derive(Debug, Clone)]
pub struct UnreadableBucketConfig {
pub bucket: String,
pub config_file: String,
/// Length of the stored bytes that failed to parse.
pub raw_len: usize,
}
impl std::fmt::Display for UnreadableBucketConfig {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
"persisted bucket configuration {} for bucket {} cannot be parsed; replace or delete it before changing it",
self.config_file, self.bucket
"persisted bucket configuration {} ({} bytes) for bucket {} cannot be parsed; back up the stored bytes (rustfs inspect bucket-meta), then replace or delete the configuration",
self.config_file, self.raw_len, self.bucket
)
}
}
@@ -297,13 +299,21 @@ impl std::fmt::Display for UnreadableBucketConfig {
impl std::error::Error for UnreadableBucketConfig {}
pub fn is_unreadable_config_error(err: &Error) -> bool {
matches!(err, Error::Io(io) if io.get_ref().is_some_and(|inner| inner.is::<UnreadableBucketConfig>()))
unreadable_config_refusal(err).is_some()
}
pub(crate) fn unreadable_config_error(bucket: &str, config_file: &str) -> Error {
pub fn unreadable_config_refusal(err: &Error) -> Option<&UnreadableBucketConfig> {
match err {
Error::Io(io) => io.get_ref().and_then(|inner| inner.downcast_ref::<UnreadableBucketConfig>()),
_ => None,
}
}
pub(crate) fn unreadable_config_error(bucket: &str, config_file: &str, raw_len: usize) -> Error {
Error::other(UnreadableBucketConfig {
bucket: bucket.to_string(),
config_file: config_file.to_string(),
raw_len,
})
}
@@ -320,7 +330,7 @@ pub enum ConfigState<'a, T> {
/// The stored bytes parsed.
Valid(&'a T),
/// Bytes are stored but could not be parsed.
Unreadable,
Unreadable { raw_len: usize },
}
impl<'a, T> ConfigState<'a, T> {
@@ -328,7 +338,7 @@ impl<'a, T> ConfigState<'a, T> {
match parsed {
Some(config) => Self::Valid(config),
None if raw.is_empty() => Self::Absent,
None => Self::Unreadable,
None => Self::Unreadable { raw_len: raw.len() },
}
}
@@ -338,18 +348,46 @@ impl<'a, T> ConfigState<'a, T> {
match self {
Self::Absent => Ok(None),
Self::Valid(config) => Ok(Some(config)),
Self::Unreadable => Err(unreadable_config_error(bucket, config_file)),
Self::Unreadable { raw_len } => Err(unreadable_config_error(bucket, config_file, raw_len)),
}
}
}
/// The XML sub-configurations whose parse failure is retained as
/// [`ConfigState::Unreadable`].
pub const XML_BUCKET_CONFIG_FILES: [&str; 13] = [
BUCKET_NOTIFICATION_CONFIG,
BUCKET_LIFECYCLE_CONFIG,
OBJECT_LOCK_CONFIG,
BUCKET_VERSIONING_CONFIG,
BUCKET_SSECONFIG,
BUCKET_TAGGING_CONFIG,
BUCKET_REPLICATION_CONFIG,
BUCKET_CORS_CONFIG,
BUCKET_LOGGING_CONFIG,
BUCKET_WEBSITE_CONFIG,
BUCKET_ACCELERATE_CONFIG,
BUCKET_REQUEST_PAYMENT_CONFIG,
BUCKET_PUBLIC_ACCESS_BLOCK_CONFIG,
];
impl BucketMetadata {
/// Whether the stored XML sub-configuration `config_file` has bytes that
/// cannot be parsed. Non-XML config files report `false`: they carry their
/// own unreadable handling (policy, quota, bucket targets).
pub fn xml_config_unreadable(&self, config_file: &str) -> bool {
fn unreadable<T>(raw: &[u8], parsed: &Option<T>) -> bool {
matches!(ConfigState::of(raw, parsed), ConfigState::Unreadable)
self.xml_config_unreadable_len(config_file).is_some()
}
/// Length of the stored bytes of XML sub-configuration `config_file` when
/// they cannot be parsed; `None` when it is absent, readable, or not an
/// XML config.
pub fn xml_config_unreadable_len(&self, config_file: &str) -> Option<usize> {
fn unreadable<T>(raw: &[u8], parsed: &Option<T>) -> Option<usize> {
match ConfigState::of(raw, parsed) {
ConfigState::Unreadable { raw_len } => Some(raw_len),
ConfigState::Absent | ConfigState::Valid(_) => None,
}
}
match config_file {
BUCKET_NOTIFICATION_CONFIG => unreadable(&self.notification_config_xml, &self.notification_config),
@@ -367,7 +405,7 @@ impl BucketMetadata {
BUCKET_PUBLIC_ACCESS_BLOCK_CONFIG => {
unreadable(&self.public_access_block_config_xml, &self.public_access_block_config)
}
_ => false,
_ => None,
}
}
}
@@ -570,6 +608,13 @@ impl BucketMetadata {
self.lock_enabled || self.object_lock_config.as_ref().is_some_and(|v| v.enabled())
}
/// Whether an operation that may skip Object Lock checks must keep them.
/// Stored lock bytes that cannot be parsed leave the lock state unknown,
/// which must not be read as "no Object Lock".
pub fn object_lock_checks_required(&self) -> bool {
self.object_locking() || self.xml_config_unreadable(OBJECT_LOCK_CONFIG)
}
pub fn table_bucket_enabled(&self) -> bool {
!self.table_bucket_config_json.is_empty()
}
@@ -1082,14 +1127,14 @@ impl BucketMetadata {
/// |---|---|
/// | 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. |
/// | versioning | Fails closed in `get_versioning_config` and the delete-time snapshot; guessing Unversioned would make delete markers and version ids diverge from what is on disk. Object writes lay out versions through `BucketVersioningSys::get_for_write`, which refuses in strict mode and, in the default permissive mode, keeps the historical unversioned write (see `config_parse_mode`). |
/// | 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 | `get_lifecycle_config` fails closed, so GetBucketLifecycle reports the fault instead of NoSuchLifecycleConfiguration. ILM, scanner and expiry-header consumers still degrade to "no rules": nothing is deleted or moved on the strength of an unreadable rule set, and 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. |
/// | notification | `get_notification_config` fails closed, so GetBucketNotificationConfiguration reports the fault and startup leaves that one bucket's rules unchanged instead of clearing them; other buckets are unaffected. |
/// | tagging, CORS, logging, website, accelerate, request payment | The getter fails closed so the matching GET API reports the fault instead of "not configured". Per-request consumers (CORS response headers) still degrade to "not configured", which is the restrictive direction. |
/// | bucket ACL | Safe to degrade: it only shapes an optional response. |
///
@@ -1097,6 +1142,10 @@ impl BucketMetadata {
/// read-modify-write of an unreadable XML config before its mutate step
/// runs, so the unreadable bytes are never replaced by a rewrite that saw
/// them as absent. Other configs of the same bucket stay writable.
///
/// Every unreadable XML config found here is counted in
/// `rustfs_bucket_metadata_parse_failed_total`, reflected in
/// `rustfs_bucket_metadata_unparsable_current`, and logged at error level.
pub(super) fn parse_all_configs(&mut self) -> Result<()> {
if let Err(e) = self.parse_policy_config() {
tracing::warn!(
@@ -1340,6 +1389,14 @@ impl BucketMetadata {
);
}
if !self.name.is_empty() {
let unreadable: Vec<(&'static str, usize)> = XML_BUCKET_CONFIG_FILES
.iter()
.filter_map(|config| self.xml_config_unreadable_len(config).map(|raw_len| (*config, raw_len)))
.collect();
super::config_parse_mode::record_bucket_config_parse_state(&self.name, &unreadable);
}
Ok(())
}
}
@@ -1457,6 +1514,18 @@ where
mod test {
use super::*;
/// rustfs/backlog#1734: `StorageError::clone` rebuilds I/O errors from
/// their text. The unreadable-config refusal must stay typed across that
/// clone, or a cloned error stops mapping to its retryable S3 response.
#[test]
fn unreadable_config_refusal_survives_storage_error_clone() {
let err = unreadable_config_error("b", BUCKET_TAGGING_CONFIG, 7);
let cloned = err.clone();
assert!(is_unreadable_config_error(&cloned), "clone lost the typed refusal: {cloned}");
assert_eq!(unreadable_config_refusal(&cloned).map(|r| r.raw_len), Some(7));
assert!(is_unreadable_config_error(&err));
}
/// Decode a whitespace-tolerant hex fixture into bytes.
fn decode_hex(s: &str) -> Vec<u8> {
let s: String = s.chars().filter(|c| !c.is_whitespace()).collect();
+108 -34
View File
@@ -152,8 +152,8 @@ enum BucketMetadataAuthority {
}
pub(crate) fn object_lock_config_state_from_authoritative_metadata(bm: &BucketMetadata) -> Result<ObjectLockConfigState> {
if bm.object_lock_config.is_none() && !bm.object_lock_config_xml.is_empty() {
return Err(Error::other("persisted bucket Object Lock configuration is invalid"));
if let Some(raw_len) = bm.xml_config_unreadable_len(super::metadata::OBJECT_LOCK_CONFIG) {
return Err(unreadable_config_error(&bm.name, super::metadata::OBJECT_LOCK_CONFIG, raw_len));
}
if let Some(config) = bm.object_lock_config.clone() {
@@ -1865,6 +1865,7 @@ impl BucketMetadataSys {
drop(map);
let removed_fabricated = self.fabricated_metadata.write().await.remove(bucket);
self.missing_buckets.insert(bucket.to_string(), ()).await;
super::config_parse_mode::forget_bucket_config_parse_state(bucket);
if removed {
BucketTargetSys::get().delete(bucket).await;
clear_bucket_durability(bucket);
@@ -1962,8 +1963,8 @@ impl BucketMetadataSys {
// from nothing; persisting that destroys the only copy of the stored
// bytes. Only the rewritten config is checked: `update_config` carries
// every other raw config through unchanged.
if bm.xml_config_unreadable(config_file) {
return Err(unreadable_config_error(bucket, config_file));
if let Some(raw_len) = bm.xml_config_unreadable_len(config_file) {
return Err(unreadable_config_error(bucket, config_file, raw_len));
}
let data = mutate(&bm)?;
@@ -2207,12 +2208,11 @@ impl BucketMetadataSys {
}
};
if !bm.versioning_config_xml.is_empty() && bm.versioning_config.is_none() {
Err(Error::other("persisted bucket versioning configuration is invalid"))
} else if let Some(config) = &bm.versioning_config {
Ok((config.clone(), bm.versioning_config_updated_at))
} else {
Ok((VersioningConfiguration::default(), bm.versioning_config_updated_at))
match ConfigState::of(&bm.versioning_config_xml, &bm.versioning_config)
.require(bucket, super::metadata::BUCKET_VERSIONING_CONFIG)?
{
Some(config) => Ok((config.clone(), bm.versioning_config_updated_at)),
None => Ok((VersioningConfiguration::default(), bm.versioning_config_updated_at)),
}
}
@@ -2221,8 +2221,8 @@ impl BucketMetadataSys {
return Ok(false);
};
if metadata.versioning_config.is_none() && !metadata.versioning_config_xml.is_empty() {
return Err(Error::other("persisted bucket versioning configuration is invalid"));
if let Some(raw_len) = metadata.xml_config_unreadable_len(super::metadata::BUCKET_VERSIONING_CONFIG) {
return Err(unreadable_config_error(bucket, super::metadata::BUCKET_VERSIONING_CONFIG, raw_len));
}
Ok(metadata.versioning_config.is_none() && metadata.versioning_config_xml.is_empty())
@@ -2292,12 +2292,11 @@ impl BucketMetadataSys {
pub async fn get_public_access_block_config(&self, bucket: &str) -> Result<(PublicAccessBlockConfiguration, OffsetDateTime)> {
let (bm, _) = self.get_config(bucket).await?;
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)
match ConfigState::of(&bm.public_access_block_config_xml, &bm.public_access_block_config)
.require(bucket, super::metadata::BUCKET_PUBLIC_ACCESS_BLOCK_CONFIG)?
{
Some(config) => Ok((config.clone(), bm.public_access_block_config_updated_at)),
None => Err(Error::ConfigNotFound),
}
}
@@ -2587,28 +2586,24 @@ impl BucketMetadataSys {
pub async fn get_notification_config(&self, bucket: &str) -> Result<Option<NotificationConfiguration>> {
let bm = match self.get_config(bucket).await {
Ok((bm, _)) => bm.notification_config.clone(),
Err(err) => {
if err == Error::ConfigNotFound {
None
} else {
return Err(err);
}
}
Ok((bm, _)) => bm,
Err(Error::ConfigNotFound) => return Ok(None),
Err(err) => return Err(err),
};
Ok(bm)
// Unreadable must not read as "no notification configured": that
// would silently drop the bucket's event rules.
Ok(ConfigState::of(&bm.notification_config_xml, &bm.notification_config)
.require(bucket, super::metadata::BUCKET_NOTIFICATION_CONFIG)?
.cloned())
}
pub async fn get_sse_config(&self, bucket: &str) -> Result<(ServerSideEncryptionConfiguration, OffsetDateTime)> {
let (bm, _) = self.get_config(bucket).await?;
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)
match ConfigState::of(&bm.encryption_config_xml, &bm.sse_config).require(bucket, super::metadata::BUCKET_SSECONFIG)? {
Some(config) => Ok((config.clone(), bm.encryption_config_updated_at)),
None => Err(Error::ConfigNotFound),
}
}
@@ -3938,11 +3933,90 @@ mod tests {
assert_ne!(err, Error::ConfigNotFound, "unreadable lifecycle must not read as absent");
// The genuinely absent case still reads as absent.
let absent_bucket = "absent-tagging-read";
let absent_bucket = "absent-tagging-read-control";
persist_bucket_with_raw_config(&sys, &dirs, absent_bucket, BUCKET_TAGGING_CONFIG, b"").await;
assert_eq!(sys.get_tagging_config(absent_bucket).await.expect_err("absent"), Error::ConfigNotFound);
}
/// rustfs/backlog#1734: the configs that gate object writes and deletes
/// (versioning, Object Lock, default encryption) and the notification
/// config must refuse with the typed unreadable-config error, so the S3
/// layer can answer 503 with the bucket and config named instead of a
/// generic 500, and notification setup can isolate the one bucket.
#[tokio::test]
async fn unreadable_gating_configs_refuse_with_the_typed_error() {
use crate::bucket::metadata::{BUCKET_VERSIONING_CONFIG, unreadable_config_refusal};
let (dirs, ecstore) = isolated_store_over_temp_disks().await;
let sys = BucketMetadataSys::new(ecstore);
// `update_config` validates some configs on write, so the corrupt
// bytes go straight into the raw fields, as a damaged object would.
let persist_corrupt = |bucket: &'static str, corrupt: fn(&mut BucketMetadata)| {
for dir in &dirs {
std::fs::create_dir_all(dir.path().join(bucket)).expect("bucket volume should be created");
}
let mut bm = BucketMetadata::new(bucket);
corrupt(&mut bm);
sys.persist_new_and_set(bm)
};
persist_corrupt("unreadable-versioning", |bm| {
bm.versioning_config_xml = b"<VersioningConfiguration><Status>".to_vec();
})
.await
.expect("corrupt versioning should persist");
let err = sys
.get_versioning_config("unreadable-versioning")
.await
.expect_err("unreadable versioning must not read as a value");
let refusal = unreadable_config_refusal(&err).unwrap_or_else(|| panic!("expected a typed refusal, got {err}"));
assert_eq!(refusal.config_file, BUCKET_VERSIONING_CONFIG);
assert_eq!(refusal.raw_len, b"<VersioningConfiguration><Status>".len());
persist_corrupt("unreadable-lock", |bm| bm.object_lock_config_xml = b"<ObjectLockConfiguration>".to_vec())
.await
.expect("corrupt Object Lock should persist");
let err = sys
.get_object_lock_config_state("unreadable-lock")
.await
.expect_err("unreadable Object Lock must not read as a value");
assert!(unreadable_config_refusal(&err).is_some(), "{err}");
persist_corrupt("unreadable-sse", |bm| {
bm.encryption_config_xml = b"<ServerSideEncryptionConfiguration>".to_vec();
})
.await
.expect("corrupt encryption should persist");
let err = sys
.get_sse_config("unreadable-sse")
.await
.expect_err("unreadable encryption must not read as a value");
assert!(unreadable_config_refusal(&err).is_some(), "{err}");
persist_corrupt("unreadable-notify", |bm| {
bm.notification_config_xml = b"<NotificationConfiguration>".to_vec();
})
.await
.expect("corrupt notification should persist");
let err = sys
.get_notification_config("unreadable-notify")
.await
.expect_err("unreadable notification must not read as \"no notification configured\"");
assert!(unreadable_config_refusal(&err).is_some(), "{err}");
// Absent stays absent.
persist_corrupt("absent-notification-read", |_| {})
.await
.expect("plain bucket should persist");
assert!(
sys.get_notification_config("absent-notification-read")
.await
.expect("absent")
.is_none()
);
}
/// A tagging rewrite through `update_config_with` (the Swift metadata
/// POST path) is persisted: it survives a metadata reload from disk, and
/// an emptied rewrite clears the config in the cached copy too instead of
+1
View File
@@ -16,6 +16,7 @@
pub mod bandwidth;
pub mod bucket_target_sys;
pub mod config_parse_mode;
pub mod durability;
pub mod error;
pub mod lifecycle;
@@ -179,9 +179,11 @@ fn replication_config_from_metadata(metadata: &BucketMetadata) -> Result<Option<
}
fn delete_request_snapshot_from_metadata(metadata: Arc<BucketMetadata>) -> Result<DeleteReplicationConfigSnapshot> {
if !metadata.versioning_config_xml.is_empty() && metadata.versioning_config.is_none() {
return Err(super::replication_error_boundary::Error::other(
"persisted bucket versioning configuration is invalid",
if let Some(raw_len) = metadata.xml_config_unreadable_len(crate::bucket::metadata::BUCKET_VERSIONING_CONFIG) {
return Err(crate::bucket::metadata::unreadable_config_error(
&metadata.name,
crate::bucket::metadata::BUCKET_VERSIONING_CONFIG,
raw_len,
));
}
+104 -1
View File
@@ -12,11 +12,13 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use super::config_parse_mode::{BucketConfigParseMode, bucket_config_parse_mode};
use super::metadata::unreadable_config_refusal;
use super::{metadata_sys::get_bucket_metadata_sys, versioning::VersioningApi};
use crate::disk::RUSTFS_META_BUCKET;
use crate::error::Result;
use s3s::dto::VersioningConfiguration;
use tracing::warn;
use tracing::{error, warn};
pub struct BucketVersioningSys {}
@@ -86,6 +88,23 @@ impl BucketVersioningSys {
Ok(cfg)
}
/// Versioning configuration for laying out an object write.
///
/// An unreadable stored configuration is refused in strict mode: writing
/// as if unversioned would overwrite versions of a bucket that may have
/// versioning enabled. Any other lookup failure keeps the historical
/// fallback to the default configuration.
pub async fn get_for_write(bucket: &str) -> Result<VersioningConfiguration> {
resolve_versioning_for_write(bucket, Self::get(bucket).await, bucket_config_parse_mode())
}
/// `(versioned, version_suspended)` for an object write under `prefix`,
/// from one [`Self::get_for_write`] lookup.
pub async fn write_state(bucket: &str, prefix: &str) -> Result<(bool, bool)> {
let config = Self::get_for_write(bucket).await?;
Ok((config.prefix_enabled(prefix), config.prefix_suspended(prefix)))
}
/// Instance-scoped variant of [`Self::get`] (backlog#1052): resolves the
/// caller's own instance context so a second in-process store never
/// answers with the first instance's versioning state; falls back to the
@@ -103,3 +122,87 @@ impl BucketVersioningSys {
Ok(cfg)
}
}
fn resolve_versioning_for_write(
bucket: &str,
lookup: Result<VersioningConfiguration>,
mode: BucketConfigParseMode,
) -> Result<VersioningConfiguration> {
match lookup {
Ok(config) => Ok(config),
Err(err) => match (unreadable_config_refusal(&err), mode) {
(Some(_), BucketConfigParseMode::Strict) => Err(err),
(Some(refusal), BucketConfigParseMode::Permissive) => {
// RUSTFS_COMPAT_TODO(s3gate-parse-strict): permissive mode keeps the historical unversioned write for an unreadable versioning config so a rollout can measure affected buckets first. Remove after the parse-failure metric has read zero fleet-wide for two releases and one further release has shipped with strict as the default.
error!(
event = "bucket_versioning_config_unreadable",
component = "ecstore",
subsystem = "bucket_versioning",
bucket = %bucket,
config = %refusal.config_file,
raw_len = refusal.raw_len,
mode = mode.as_str(),
result = "write_unversioned",
"Bucket versioning configuration is unreadable; writing as unversioned"
);
Ok(VersioningConfiguration::default())
}
(None, _) => {
warn!(bucket = %bucket, error = ?err, "failed to load bucket versioning configuration; using default configuration");
Ok(VersioningConfiguration::default())
}
},
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::bucket::metadata::{BUCKET_VERSIONING_CONFIG, is_unreadable_config_error, unreadable_config_error};
use crate::error::Error;
fn enabled() -> VersioningConfiguration {
VersioningConfiguration {
status: Some(s3s::dto::BucketVersioningStatus::from_static(s3s::dto::BucketVersioningStatus::ENABLED)),
..Default::default()
}
}
/// rustfs/backlog#1734: strict mode refuses a write whose versioning
/// state is unknown instead of laying it out as unversioned.
#[test]
fn strict_mode_refuses_a_write_against_unreadable_versioning() {
let err = resolve_versioning_for_write(
"b",
Err(unreadable_config_error("b", BUCKET_VERSIONING_CONFIG, 9)),
BucketConfigParseMode::Strict,
)
.expect_err("strict mode must refuse");
assert!(is_unreadable_config_error(&err), "{err}");
}
#[test]
fn permissive_mode_keeps_the_historical_unversioned_write() {
let config = resolve_versioning_for_write(
"b",
Err(unreadable_config_error("b", BUCKET_VERSIONING_CONFIG, 9)),
BucketConfigParseMode::Permissive,
)
.expect("permissive mode keeps writing");
assert!(!config.enabled());
}
#[test]
fn readable_config_and_lookup_faults_behave_as_before_in_every_mode() {
for mode in [BucketConfigParseMode::Permissive, BucketConfigParseMode::Strict] {
assert!(
resolve_versioning_for_write("b", Ok(enabled()), mode)
.expect("readable")
.enabled()
);
let fallback =
resolve_versioning_for_write("b", Err(Error::other("metadata read failed")), mode).expect("historical fallback");
assert!(!fallback.enabled());
}
}
}
+4
View File
@@ -616,6 +616,10 @@ impl Clone for StorageError {
source: Box::new(context.clone()),
},
))
} else if let Some(refusal) = crate::bucket::metadata::unreadable_config_refusal(self) {
// Keep the refusal typed so a cloned error still maps to
// its retryable S3 response (rustfs/backlog#1734).
StorageError::Io(std::io::Error::new(e.kind(), refusal.clone()))
} else {
StorageError::Io(std::io::Error::new(e.kind(), e.to_string()))
}
+10 -1
View File
@@ -5594,7 +5594,7 @@ fn check_object_lock_retention_update(bucket: &str, object: &str, obj_info: &Obj
/// object-lock protection is never skipped because of a metadata lookup miss.
#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
pub(crate) fn object_lock_delete_check_required(bucket_meta: Option<&crate::bucket::metadata::BucketMetadata>) -> bool {
bucket_meta.is_none_or(|meta| meta.object_locking())
bucket_meta.is_none_or(|meta| meta.object_lock_checks_required())
}
fn restore_expiry_snapshot_matches(obj_info: &ObjectInfo, opts: &ObjectOptions) -> bool {
@@ -11996,6 +11996,15 @@ mod tests {
assert!(object_lock_delete_check_required(None));
}
/// rustfs/backlog#1734: stored Object Lock bytes that cannot be parsed
/// mean the lock state is unknown, not absent; the check must stay on.
#[test]
fn test_object_lock_delete_check_required_fails_closed_on_unreadable_lock_config() {
let mut bm = crate::bucket::metadata::BucketMetadata::new("unreadable-lock-bucket");
bm.object_lock_config_xml = b"<ObjectLockConfiguration>".to_vec();
assert!(object_lock_delete_check_required(Some(&bm)));
}
#[test]
fn test_should_persist_encryption_original_size_rejects_plain_metadata() {
let metadata = HashMap::from([("content-type".to_string(), "application/octet-stream".to_string())]);
@@ -19,6 +19,7 @@
- `backlog-2102` rc.2/rc.3 empty scanner usage floor recovery: old DeleteBucket cleanup could synthesize an empty incomplete v2 usage primary/backup before leadership added an epoch, while newer scanners require a durable authoritative baseline identity. New scanners recognize only that exact serialized empty-fence shape, preserve its epoch through a CAS-protected recovery marker, and rebuild namespace coverage without treating zero usage as authoritative. Remove this recovery path and marker after rc.2 and rc.3 are no longer supported direct-upgrade sources.
- `backlog-2122` rc.1-rc.3 non-empty scanner usage floor recovery: leadership fencing in those releases can stamp scanner_epoch onto a real bucket-usage snapshot before any scanner cycle completed, leaving a non-empty floor with no scanner_cycle and no authoritative baseline identity. New scanners recognize only this consistent incomplete fenced shape, preserve the epoch through the CAS-protected recovery marker, and rebuild namespace coverage without treating the old usage data as authoritative. Remove this recovery path after rc.1, rc.2, and rc.3 are no longer supported direct-upgrade sources.
- `s3gate-metadata-xml` persisted bucket XML migration: mixed-version site-replication peers, retained `.metadata.bin` objects, and backup archives can all carry XML written by the s3s codec, so the gateway migration must keep the legacy codec available until every stored form has crossed a verified rewrite boundary. Remove the legacy s3s parser and serializer only after the minimum supported direct-upgrade release reads and writes every persisted XML configuration family through the gateway codec, every supported mixed-version site-replication topology has completed its writer upgrade, and migration tooling has verified or rewritten every retained bucket metadata object and restorable backup archive.
- `s3gate-parse-strict` unreadable bucket versioning on object writes: a stored versioning configuration that cannot be parsed was historically read as unversioned, and existing clusters may already hold such buckets unnoticed. RUSTFS_BUCKET_CONFIG_PARSE_MODE defaults to permissive, which keeps that unversioned write and records it through rustfs_bucket_metadata_parse_failed_total, rustfs_bucket_metadata_unparsable_current, and an error log; strict refuses the write with 503. Paths that already refuse an unreadable configuration (typed getters, the read-modify-write guard, Object Lock, default encryption, delete-time versioning) refuse in both modes. Mixed-version risk window: older nodes lack the per-config read-modify-write guard and still write unversioned, so the metric must be driven to zero before relying on strict. Flip the default to strict after the parse-failure counter has read zero fleet-wide for two consecutive releases, and remove the permissive branch and the switch one release after that.
- `rustfs-6339` legacy bucket policy ID casing: earlier RustFS releases persisted the top-level policy identifier as "ID", while current writes use the S3-compatible "Id" spelling. Readers accept both spellings so retained bucket metadata remains usable after upgrade. Remove the legacy alias after migration tooling has rewritten every retained bucket policy using "ID".
- `table-publication-fence-v1` table publication fencing: nodes that predate table and table-bucket publication fences can mutate live files while a new node is publishing a catalog pointer. New nodes retain exact object guards until the operator confirms that every serving node uses the new fences. Fleet confirmation also requires non-overlapping active warehouse prefixes and lifecycle workers that exclude table buckets. Remove the exact live-file fallback and the fleet-confirmation gate after the minimum supported RustFS release acquires table fences for registered-table mutations and table-bucket fences for unresolved-prefix mutations.
- `table-catalog-strong-snapshot-v1` durable strong catalog snapshot compatibility: version 1 writes continue during mixed-version rollout until operators confirm that every serving node reads version 2, and version 1 table/view identifier collisions remain available only for cleanup. Remove version 1 writes and collision cleanup after the minimum supported RustFS release reads version 2 and every retained durable strong snapshot is collision-free and has been upgraded to version 2.
+18 -11
View File
@@ -1942,17 +1942,24 @@ impl DefaultBucketUsecase {
.await
.map_err(ApiError::from)?;
let has_notification_config = metadata_sys::get_notification_config(&bucket).await.unwrap_or_else(|err| {
warn!(
component = LOG_COMPONENT_APP,
subsystem = LOG_SUBSYSTEM_BUCKET,
event = "bucket_notification_config_load_failed",
bucket = %bucket,
error = ?err,
"Failed to load bucket notification configuration"
);
None
});
let has_notification_config = match metadata_sys::get_notification_config(&bucket).await {
Ok(config) => config,
// An unreadable config is not "no notifications configured".
Err(err) if crate::storage_api::error::is_unreadable_config_error(&err) => {
return Err(ApiError::from(err).into());
}
Err(err) => {
warn!(
component = LOG_COMPONENT_APP,
subsystem = LOG_SUBSYSTEM_BUCKET,
event = "bucket_notification_config_load_failed",
bucket = %bucket,
error = ?err,
"Failed to load bucket notification configuration"
);
None
}
};
if let Some(NotificationConfiguration {
event_bridge_configuration,
@@ -3155,7 +3155,7 @@ async fn delete_object_versioning_config_failure_leaves_latest_object_intact() {
.await
.expect_err("versioning config failure must reject DeleteObject");
assert_eq!(err.code(), &s3s::S3ErrorCode::InternalError);
assert_eq!(err.code(), &s3s::S3ErrorCode::ServiceUnavailable);
assert_eq!(read_object_bytes(&ecstore, &bucket, object).await, payload);
}
+3 -2
View File
@@ -630,8 +630,9 @@ impl DefaultMultipartUsecase {
let mut opts = get_complete_multipart_upload_opts_with_replication_authorization(&req.headers, replication_authorized)
.map_err(ApiError::from)?;
apply_bucket_generation_guard(&req, &bucket, &mut opts)?;
let versioned = BucketVersioningSys::prefix_enabled(&bucket, &key).await;
let version_suspended = BucketVersioningSys::prefix_suspended(&bucket, &key).await;
let (versioned, version_suspended) = BucketVersioningSys::write_state(&bucket, &key)
.await
.map_err(ApiError::from)?;
opts.versioned = versioned;
opts.version_suspended = version_suspended;
let capacity_scope_token = Uuid::new_v4();
+1 -1
View File
@@ -1429,7 +1429,7 @@ mod tests {
.await
.expect_err("an unreadable bucket encryption configuration must refuse the copy");
assert_eq!(err.code(), &S3ErrorCode::InternalError);
assert_eq!(err.code(), &S3ErrorCode::ServiceUnavailable);
let lookup_err = store
.get_object_info(&bucket, destination, &ObjectOptions::default())
.await
+4 -2
View File
@@ -554,9 +554,11 @@ impl DefaultObjectUsecase {
opts.expected_bucket_incarnation_id = ctx.expected_bucket_incarnation_id;
opts.preserve_etag = ctx.preserve_etag.clone();
opts.preserve_delete_marker = ctx.preserve_delete_marker;
let versioned = BucketVersioningSys::prefix_enabled(&bucket, &key).await;
let (versioned, version_suspended) = BucketVersioningSys::write_state(&bucket, &key)
.await
.map_err(ApiError::from)?;
opts.versioned = versioned;
opts.version_suspended = BucketVersioningSys::prefix_suspended(&bucket, &key).await;
opts.version_suspended = version_suspended;
let capacity_scope_token = Uuid::new_v4();
opts.capacity_scope_token = Some(capacity_scope_token);
+2 -2
View File
@@ -4188,7 +4188,7 @@ mod tests {
.await
.expect_err("an unreadable bucket encryption configuration must refuse the write");
assert_eq!(err.code(), &S3ErrorCode::InternalError);
assert_eq!(err.code(), &S3ErrorCode::ServiceUnavailable);
let lookup_err = store
.get_object_info(&bucket, object, &ObjectOptions::default())
.await
@@ -4269,7 +4269,7 @@ mod tests {
.await
.expect_err("an unreadable bucket encryption configuration must refuse the extract upload");
assert_eq!(err.code(), &S3ErrorCode::InternalError);
assert_eq!(err.code(), &S3ErrorCode::ServiceUnavailable);
let lookup_err = store
.get_object_info(&bucket, "archive.tar", &ObjectOptions::default())
.await
+4 -3
View File
@@ -372,8 +372,9 @@ pub(super) fn resolve_bucket_default_sse(
/// bucket whose metadata document is absent — `ConfigNotFound`, so a cold
/// cache and a missing bucket are never turned into a refusal, and the write
/// still fails later with its own `NoSuchBucket`;
/// * blob present but unparseable — deterministic, so retrying cannot help;
/// surfaces as `InternalError` until an operator repairs or removes it;
/// * blob present but unparseable — the typed unreadable-config refusal,
/// surfaced as `ServiceUnavailable` naming the bucket and config until an
/// operator repairs or removes it (rustfs/backlog#1734);
/// * the metadata read itself failed (namespace lock, quorum, disk, an
/// uninitialized metadata system) — transient, and the typed error maps to
/// the retryable `ServiceUnavailable`.
@@ -943,7 +944,7 @@ impl DefaultObjectUsecase {
pub(crate) async fn object_lock_checks_required(bucket: &str) -> bool {
get_bucket_metadata(bucket)
.await
.map_or(true, |metadata| metadata.object_locking())
.map_or(true, |metadata| metadata.object_lock_checks_required())
}
pub(super) fn object_lock_checks_required_for_state(state: &metadata_sys::ObjectLockConfigState) -> bool {
+30 -1
View File
@@ -13,7 +13,7 @@
// limitations under the License.
use crate::storage_api::error::contract::{StorageErrorCode, range::HTTPRangeError};
use crate::storage_api::error::{PoolMetadataError, QuotaError, StorageError};
use crate::storage_api::error::{PoolMetadataError, QuotaError, StorageError, unreadable_config_refusal};
use http::StatusCode;
use rustfs_kms::KmsUnavailableError;
use s3s::{S3Error, S3ErrorCode};
@@ -520,6 +520,16 @@ impl From<StorageError> for ApiError {
source: Some(Box::new(err)),
};
}
// A stored bucket config that cannot be parsed stays refused until an
// operator repairs it, so it is a recoverable 503 that names what to
// repair rather than a generic 500 (rustfs/backlog#1734).
if let Some(message) = unreadable_config_refusal(&err).map(ToString::to_string) {
return ApiError {
code: S3ErrorCode::ServiceUnavailable,
message,
source: Some(Box::new(err)),
};
}
if let StorageError::Io(ref io_err) = err
&& let Some(inner) = io_err.get_ref()
{
@@ -805,6 +815,25 @@ mod tests {
use s3s::{S3Error, S3ErrorCode};
use std::io::{Error as IoError, ErrorKind};
/// rustfs/backlog#1734: a refusal to act on a stored bucket config that
/// cannot be parsed is recoverable once an operator repairs the bytes, so
/// it answers 503 naming the bucket, config and stored length, not a
/// generic 500. It must survive the error being cloned on the way.
#[test]
fn unreadable_bucket_config_maps_to_service_unavailable_naming_the_config() {
let err = StorageError::other(crate::storage_api::error::UnreadableBucketConfig {
bucket: "photos".to_string(),
config_file: "versioning.xml".to_string(),
raw_len: 42,
});
for api_error in [ApiError::from(err.clone()), ApiError::from(err)] {
assert_eq!(api_error.code, S3ErrorCode::ServiceUnavailable);
for needle in ["photos", "versioning.xml", "42"] {
assert!(api_error.message.contains(needle), "{needle} missing from {:?}", api_error.message);
}
}
}
#[test]
fn api_error_diagnostic_preserves_typed_cause_without_sensitive_payload() {
let error = ApiError::from(StorageError::Io(IoError::new(ErrorKind::TimedOut, "secret=do-not-log")));
+71 -9
View File
@@ -15,6 +15,7 @@
use crate::runtime_sources::current_region;
use crate::server::ShutdownHandle;
use crate::server::runtime_sources::current_notify_interface;
use crate::storage_api::error::{StorageError, is_unreadable_config_error};
use crate::storage_api::startup::bucket_metadata::contract::bucket::{BucketOperations, BucketOptions};
use crate::storage_api::startup::init::{
get_bucket_notification_config, process_lambda_configurations, process_queue_configurations, process_topic_configurations,
@@ -169,13 +170,48 @@ fn notification_config_to_event_rules(
Ok(event_rules)
}
async fn apply_bucket_notification_configuration(bucket: &str, region: &str) -> Result<bool, NotificationError> {
let has_notification_config = get_bucket_notification_config(bucket)
.await
.map_err(|err| NotificationError::StorageNotAvailable(format!("load bucket notification config for {bucket}: {err}")))?;
/// One bucket's persisted notification configuration as startup sees it.
#[derive(Debug)]
enum BucketNotificationLookup {
Configured(s3s::dto::NotificationConfiguration),
Missing,
/// Stored bytes cannot be parsed. Deterministic, so a retry cannot help,
/// and reading it as missing would clear the bucket's rules.
Unreadable,
}
match has_notification_config {
Some(cfg) => {
fn classify_bucket_notification_lookup(
bucket: &str,
lookup: Result<Option<s3s::dto::NotificationConfiguration>, StorageError>,
) -> Result<BucketNotificationLookup, NotificationError> {
match lookup {
Ok(Some(cfg)) => Ok(BucketNotificationLookup::Configured(cfg)),
Ok(None) => Ok(BucketNotificationLookup::Missing),
Err(err) if is_unreadable_config_error(&err) => {
error!(
target: "rustfs::init",
event = "notification_config_unreadable",
component = LOG_COMPONENT_INIT,
subsystem = LOG_SUBSYSTEM_NOTIFICATION,
bucket = %bucket,
error = %err,
"Bucket notification configuration is unreadable; leaving its rules unchanged"
);
Ok(BucketNotificationLookup::Unreadable)
}
Err(err) => Err(NotificationError::StorageNotAvailable(format!(
"load bucket notification config for {bucket}: {err}"
))),
}
}
/// Apply one bucket's persisted notification rules. An unreadable config is
/// isolated to its bucket: it neither aborts setup for the other buckets nor
/// clears the bucket's rules as if it had none.
async fn apply_bucket_notification_configuration(bucket: &str, region: &str) -> Result<bool, NotificationError> {
match classify_bucket_notification_lookup(bucket, get_bucket_notification_config(bucket).await)? {
BucketNotificationLookup::Unreadable => Ok(false),
BucketNotificationLookup::Configured(cfg) => {
info!(
target: "rustfs::init",
event = "notification_config_loaded",
@@ -195,7 +231,7 @@ async fn apply_bucket_notification_configuration(bucket: &str, region: &str) ->
.await?;
Ok(true)
}
None => {
BucketNotificationLookup::Missing => {
info!(
target: "rustfs::init",
event = "notification_config_missing",
@@ -1375,16 +1411,42 @@ pub async fn init_sftp_system() -> Result<Option<ShutdownHandle>, Box<dyn std::e
#[cfg(test)]
mod tests {
use super::{
build_aws_kms_config, build_vault_kms_config, build_vault_transit_kms_config, notification_config_to_event_rules,
resolve_buffer_profile_config,
BucketNotificationLookup, build_aws_kms_config, build_vault_kms_config, build_vault_transit_kms_config,
classify_bucket_notification_lookup, notification_config_to_event_rules, resolve_buffer_profile_config,
};
use crate::config::{BufferConfig, WorkloadProfile};
use crate::storage_api::error::{StorageError, UnreadableBucketConfig};
use rustfs_config::KI_B;
use rustfs_s3_types::EventName;
use s3s::dto::{
FilterRule, FilterRuleName, NotificationConfiguration, NotificationConfigurationFilter, QueueConfiguration, S3KeyFilter,
};
/// rustfs/backlog#1734: an unreadable notification config is isolated to
/// its bucket. It must not abort setup for every other bucket (a storage
/// fault still does, so reconciliation retries) and must not read as
/// missing, which would clear the bucket's rules.
#[test]
fn unreadable_notification_config_is_isolated_to_its_bucket() {
let unreadable = StorageError::other(UnreadableBucketConfig {
bucket: "b".to_string(),
config_file: "notification.xml".to_string(),
raw_len: 5,
});
assert!(matches!(
classify_bucket_notification_lookup("b", Err(unreadable)),
Ok(BucketNotificationLookup::Unreadable)
));
assert!(matches!(
classify_bucket_notification_lookup("b", Ok(None)),
Ok(BucketNotificationLookup::Missing)
));
assert!(matches!(
classify_bucket_notification_lookup("b", Err(StorageError::ErasureReadQuorum)),
Err(rustfs_notify::NotificationError::StorageNotAvailable(_))
));
}
#[test]
fn resolve_buffer_profile_config_returns_fallback_when_primary_is_invalid() {
let invalid_primary = WorkloadProfile::Custom(BufferConfig {
+10
View File
@@ -26,6 +26,7 @@ const EVENT_EXTERNAL_ENV_COMPAT_CONFLICT: &str = "external_env_compat_conflict";
const EVENT_EXTERNAL_ENV_COMPAT_APPLIED: &str = "external_env_compat_applied";
const EVENT_OBSERVABILITY_GUARD_SET: &str = "observability_guard_set";
const EVENT_OBSERVABILITY_GUARD_SET_FAILED: &str = "observability_guard_set_failed";
const EVENT_BUCKET_CONFIG_PARSE_MODE_SELECTED: &str = "bucket_config_parse_mode_selected";
#[derive(Debug)]
pub(crate) enum StartupServerPreflightError {
@@ -66,6 +67,15 @@ pub(crate) async fn init_startup_server_preflight(
init_license(config.license.clone());
init_startup_observability(config.obs_endpoint.clone()).await?;
log_external_prefix_compat_report(env_compat_report);
let parse_mode = crate::storage_api::startup::init::validate_bucket_config_parse_mode_env()
.map_err(|err| StartupServerPreflightError::Other(Error::other(err)))?;
info!(
event = EVENT_BUCKET_CONFIG_PARSE_MODE_SELECTED,
component = LOG_COMPONENT_MAIN,
subsystem = LOG_SUBSYSTEM_STARTUP,
mode = parse_mode.as_str(),
"Bucket config parse mode selected"
);
init_startup_runtime_foundation(config)
.await
.map_err(StartupServerPreflightError::Other)
+13 -2
View File
@@ -145,6 +145,17 @@ fn get_skip_verify_bitrot() -> bool {
}
}
/// Like [`bucket_versioning_config`], for laying out an object write or
/// delete: an unreadable stored configuration is refused in strict mode
/// instead of being read as unversioned (rustfs/backlog#1734).
async fn bucket_versioning_config_for_write(bucket: &str) -> Result<VersioningConfiguration> {
#[cfg(test)]
VERSIONING_CONFIG_LOOKUPS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
#[cfg(test)]
wait_for_versioning_config_test_hook(bucket).await;
BucketVersioningSys::get_for_write(bucket).await
}
/// Creates options for deleting an object in a bucket.
pub async fn del_opts(
bucket: &str,
@@ -153,7 +164,7 @@ pub async fn del_opts(
headers: &HeaderMap<HeaderValue>,
metadata: HashMap<String, String>,
) -> Result<ObjectOptions> {
let versioning_cfg = bucket_versioning_config(bucket).await;
let versioning_cfg = bucket_versioning_config_for_write(bucket).await?;
del_opts_with_versioning(bucket, object, vid, headers, metadata, &versioning_cfg, false)
}
@@ -350,7 +361,7 @@ pub async fn put_opts_with_replication_authorization(
metadata: HashMap<String, String>,
replication_request_authorized: bool,
) -> Result<ObjectOptions> {
let versioning_cfg = bucket_versioning_config(bucket).await;
let versioning_cfg = bucket_versioning_config_for_write(bucket).await?;
let versioned = versioning_cfg.prefix_enabled(object);
let version_suspended = versioning_cfg.prefix_suspended(object);
+1 -1
View File
@@ -414,7 +414,7 @@ pub(crate) mod ecstore_bucket {
bandwidth, bucket_target_sys, durability, lifecycle, metadata, metadata_sys, migration, object_lock, policy_sys,
remote_s3_client, replication, tagging, target, utils,
};
pub(crate) use rustfs_ecstore::api::bucket::{quota, versioning, versioning_sys};
pub(crate) use rustfs_ecstore::api::bucket::{config_parse_mode, quota, versioning, versioning_sys};
}
pub(crate) mod ecstore_capacity {
+6
View File
@@ -87,6 +87,11 @@ pub(crate) mod error {
}
}
#[cfg(test)]
pub(crate) use crate::storage::storage_api::ecstore_bucket::metadata::UnreadableBucketConfig;
pub(crate) use crate::storage::storage_api::ecstore_bucket::metadata::{
is_unreadable_config_error, unreadable_config_refusal,
};
pub(crate) use crate::storage::storage_api::ecstore_error::PoolMetadataError;
#[cfg(test)]
pub(crate) use crate::storage::storage_api::ecstore_error::PoolMetadataFailure;
@@ -330,6 +335,7 @@ pub(crate) mod startup {
}
pub(crate) mod init {
pub(crate) use crate::storage::storage_api::ecstore_bucket::config_parse_mode::validate_bucket_config_parse_mode_env;
pub(crate) use crate::storage::storage_api::{
get_bucket_notification_config, process_lambda_configurations, process_queue_configurations,
process_topic_configurations,