gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-06 20:19:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(admin): preserve metadata during export and target repair (#7258)

Browse Source 
* fix(admin): reject incomplete metadata backups

* fix(admin): repair remote targets from locked disk state

* test(admin): fence target repair against source changes

* fix(admin): report unreadable XML in metadata exports

* test(admin): enable loopback in target repair fixtures

* refactor(admin): remove unused metadata getter forwards

* test(admin): box direct target repair futures

* test(admin): box target repair scenarios at env boundary
This commit is contained in:
Zhengchao An
2026-09-06 12:18:00 +08:00
committed by GitHub
parent d3884ed3ea
commit d44244f60f
6 changed files with 870 additions and 421 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/admin/handlers/bucket_meta.rs
+227 -261
View File
@@ -67,15 +67,8 @@ use zip::{ZipArchive, ZipWriter, write::SimpleFileOptions};
const DIAGNOSTIC_EXPORT_PREFIX: &str = "_diagnostic";
const DIAGNOSTIC_EXPORT_MANIFEST: &str = "_diagnostic-manifest.json";
/// Archive entry naming the configurations a bucket stores but this build
/// could not read (rustfs/backlog#2309).
///
/// The name deliberately sits outside the configuration-file namespace the
/// importers switch on — `ImportBucketMetadata` matches known configuration
/// names and ignores everything else — so no importer can mistake the marker
/// for a configuration. Ordinary exports carry it; diagnostic exports report
/// the same failures through [`DIAGNOSTIC_EXPORT_MANIFEST`] instead, which
/// deliberately withholds the parser detail this marker records.
/// Earlier partial exports omitted configurations and carried this marker.
/// Reject those archives before import can create an incomplete replacement.
const EXPORT_UNREADABLE_MANIFEST: &str = "rustfs-unreadable-configs.json";
const LOG_COMPONENT_ADMIN: &str = "admin";
@@ -87,32 +80,15 @@ fn export_internal_error(message: impl Into<String>) -> s3s::S3Error {
s3_error!(InternalError, "{message}")
}
/// One configuration that is stored for a bucket but could not be exported.
#[derive(serde::Serialize)]
struct UnreadableExportEntry {
config: &'static str,
error: String,
}
#[derive(serde::Serialize)]
struct UnreadableExportManifest<'a> {
bucket: &'a str,
unreadable: &'a [UnreadableExportEntry],
}
/// Why one of a bucket's configurations could not be exported.
///
/// The two variants are what an ordinary export dispatches on: a bucket whose
/// stored bytes this build cannot turn into a configuration is named and
/// skipped, while a failure of our own output machinery still fails the whole
/// export closed.
/// Ordinary backups fail on either variant. Explicit diagnostics may report
/// unreadable configurations, but output failures still abort the archive.
#[derive(Debug)]
enum ExportConfigError {
/// The configuration is stored but this build cannot read it: a
/// MinIO-origin or otherwise undecodable blob, or a revision that moved
/// underneath the export. No retry of ours turns those bytes into a
/// configuration, so one such bucket must not abort a whole-cluster export
/// (rustfs/backlog#2309).
/// underneath the export. Only an explicit diagnostic export may omit it.
Unreadable(String),
/// This build failed to produce its own output for a configuration it had
/// already decoded. Nothing about the stored bytes is in doubt, so the
@@ -179,62 +155,50 @@ async fn exported_bucket_config(bucket: &str, conf: &str) -> Result<Option<Vec<u
Ok(Some(config_json))
}
BUCKET_NOTIFICATION_CONFIG => {
let config: s3s::dto::NotificationConfiguration = match metadata_sys::get_notification_config(bucket).await {
Ok(Some(res)) => res,
Err(e) => {
if e == StorageError::ConfigNotFound {
return Ok(None);
}
return Err(ExportConfigError::unreadable(format!("get bucket metadata failed: {e}")));
}
Ok(None) => return Ok(None),
};
let raw_config = metadata_sys::get(bucket)
let metadata = metadata_sys::get(bucket)
.await
.map_err(|e| ExportConfigError::unreadable(format!("get bucket metadata failed: {e}")))?
.notification_config_xml
.clone();
let config_xml = checked_raw_xml(&config, raw_config, deserialize::<s3s::dto::NotificationConfiguration>)?;
.map_err(|e| ExportConfigError::unreadable(format!("get bucket metadata failed: {e}")))?;
if metadata.notification_config_xml.is_empty() {
return Ok(None);
}
let config = metadata
.notification_config
.as_ref()
.ok_or_else(|| ExportConfigError::unreadable("persisted bucket notification configuration is invalid"))?;
let config_xml = checked_raw_xml(
config,
metadata.notification_config_xml.clone(),
deserialize::<s3s::dto::NotificationConfiguration>,
)?;
Ok(Some(config_xml))
}
BUCKET_LIFECYCLE_CONFIG => {
let config: BucketLifecycleConfiguration = match metadata_sys::get_lifecycle_config(bucket).await {
Ok((res, _)) => res,
Err(e) => {
if e == StorageError::ConfigNotFound {
return Ok(None);
}
return Err(ExportConfigError::unreadable(format!("failed to load bucket metadata: {e}")));
}
};
let raw_config = metadata_sys::get(bucket)
let metadata = metadata_sys::get(bucket)
.await
.map_err(|e| ExportConfigError::unreadable(format!("failed to load bucket metadata: {e}")))?
.lifecycle_config_xml
.clone();
let config_xml = checked_raw_xml(&config, raw_config, deserialize::<BucketLifecycleConfiguration>)?;
.map_err(|e| ExportConfigError::unreadable(format!("get bucket metadata failed: {e}")))?;
if metadata.lifecycle_config_xml.is_empty() {
return Ok(None);
}
let config = metadata
.lifecycle_config
.as_ref()
.ok_or_else(|| ExportConfigError::unreadable("persisted bucket lifecycle configuration is invalid"))?;
let config_xml =
checked_raw_xml(config, metadata.lifecycle_config_xml.clone(), deserialize::<BucketLifecycleConfiguration>)?;
Ok(Some(config_xml))
}
BUCKET_TAGGING_CONFIG => {
let config: Tagging = match metadata_sys::get_tagging_config(bucket).await {
Ok((res, _)) => res,
Err(e) => {
if e == StorageError::ConfigNotFound {
return Ok(None);
}
return Err(ExportConfigError::unreadable(format!("failed to load bucket metadata: {e}")));
}
};
let raw_config = metadata_sys::get(bucket)
let metadata = metadata_sys::get(bucket)
.await
.map_err(|e| ExportConfigError::unreadable(format!("failed to load bucket metadata: {e}")))?
.tagging_config_xml
.clone();
let config_xml = checked_raw_xml(&config, raw_config, deserialize::<Tagging>)?;
.map_err(|e| ExportConfigError::unreadable(format!("get bucket metadata failed: {e}")))?;
if metadata.tagging_config_xml.is_empty() {
return Ok(None);
}
let config = metadata
.tagging_config
.as_ref()
.ok_or_else(|| ExportConfigError::unreadable("persisted bucket tagging configuration is invalid"))?;
let config_xml = checked_raw_xml(config, metadata.tagging_config_xml.clone(), deserialize::<Tagging>)?;
Ok(Some(config_xml))
}
BUCKET_QUOTA_CONFIG_FILE => {
@@ -439,53 +403,25 @@ impl Operation for ExportBucketMetadata {
];
for bucket in buckets {
let mut unreadable: Vec<UnreadableExportEntry> = Vec::new();
for &conf in confs.iter() {
let conf_path = path_join_buf(&[bucket.name.as_str(), conf]);
let config = match exported_bucket_config(&bucket.name, conf).await {
Ok(Some(config)) => config,
Ok(None) => continue,
Err(error) => {
if query.diagnostic {
// A diagnostic archive names every configuration it
// could not export under one fixed code, carrying
// neither the payload nor the parser detail, so it
// stays shareable (rustfs/rustfs#7225).
errors.push(serde_json::json!({
"bucket": bucket.name,
"config": conf,
"code": "configuration_unavailable",
}));
continue;
}
match error {
// One bucket's undecodable blob must not abort the
// whole-cluster export: record which configuration
// could not be read and keep going, so an operator
// migrating away still gets every readable
// configuration (rustfs/backlog#2309).
ExportConfigError::Unreadable(error) => {
warn!(
event = EVENT_ADMIN_BUCKET_META_STATE,
component = LOG_COMPONENT_ADMIN,
subsystem = LOG_SUBSYSTEM_BUCKET_META,
action = "export_bucket_metadata",
result = "config_unreadable",
bucket = %bucket.name,
config_name = %conf,
error = %error,
"admin bucket meta state"
);
unreadable.push(UnreadableExportEntry { config: conf, error });
continue;
}
// Our own encoder failed on a configuration this
// build had already decoded. The stored bytes are
// not in question, so fail the export instead of
// reporting readable metadata as unreadable.
ExportConfigError::Internal(error) => return Err(error),
Err(ExportConfigError::Unreadable(error)) => {
if !query.diagnostic {
return Err(export_internal_error(format!("failed to export {conf_path}: {error}")));
}
// Diagnostics identify omitted configurations without
// exposing payloads or parser details.
errors.push(serde_json::json!({
"bucket": bucket.name,
"config": conf,
"code": "configuration_unavailable",
}));
continue;
}
Err(ExportConfigError::Internal(error)) => return Err(error),
};
let conf_path = if query.diagnostic {
path_join_buf(&[DIAGNOSTIC_EXPORT_PREFIX, &conf_path])
@@ -499,23 +435,6 @@ impl Operation for ExportBucketMetadata {
.write_all(&config)
.map_err(|e| s3_error!(InternalError, "failed to write archive entry: {e}"))?;
}
// Only reachable outside diagnostic mode, which reports the same
// failures through the archive-wide manifest instead.
if !unreadable.is_empty() {
let manifest = serde_json::to_vec(&UnreadableExportManifest {
bucket: bucket.name.as_str(),
unreadable: &unreadable,
})
.map_err(|e| export_internal_error(format!("failed to serialize unreadable manifest: {e}")))?;
let manifest_path = path_join_buf(&[bucket.name.as_str(), EXPORT_UNREADABLE_MANIFEST]);
zip_writer
.start_file(manifest_path, SimpleFileOptions::default())
.map_err(|e| s3_error!(InternalError, "failed to start archive entry: {e}"))?;
zip_writer
.write_all(&manifest)
.map_err(|e| s3_error!(InternalError, "failed to write archive entry: {e}"))?;
}
}
if query.diagnostic {
@@ -626,8 +545,12 @@ impl Operation for ImportBucketMetadata {
|| path
.strip_prefix(DIAGNOSTIC_EXPORT_PREFIX)
.is_some_and(|suffix| suffix.starts_with('/'))
|| path.rsplit('/').next() == Some(EXPORT_UNREADABLE_MANIFEST)
}) {
return Err(s3_error!(InvalidRequest, "diagnostic bucket metadata archives cannot be imported"));
return Err(s3_error!(
InvalidRequest,
"diagnostic or incomplete bucket metadata archives cannot be imported"
));
}
let durable_quota_import = imported_quota_requires_fleet_proof(&file_contents)?;
@@ -1554,6 +1477,137 @@ mod backup_zip_compatibility_tests {
assert_eq!(response.output.0, StatusCode::OK);
}
async fn assert_unreadable_xml_export_is_explicit(config_file: &str) {
const RAW_SECRET: &[u8] = b"unreadable-xml-with-private-config";
let _ = rustfs_credentials::init_global_action_credentials(
Some(ROOT_ACCESS_KEY.to_string()),
Some(ROOT_SECRET_KEY.to_string()),
);
let temp = tempfile::tempdir().expect("create unreadable XML export test root");
let env = rustfs_test_utils::TestECStoreEnv::builder()
.base_dir(temp.path())
.disk_count(1)
.build()
.await;
env.make_bucket(BUCKET, false).await;
rustfs_iam::store::object::ObjectStore::new(Arc::clone(&env.ecstore))
.save_iam_config(serde_json::json!({"version": 1}), format!("{}/format.json", *IAM_CONFIG_PREFIX))
.await
.expect("seed IAM format");
let iam = rustfs_iam::build_iam_sys(Arc::clone(&env.ecstore))
.await
.expect("build test IAM");
publish_test_app_context(Arc::new(AppContext::with_default_interfaces(
Arc::clone(&env.ecstore),
iam,
Arc::new(rustfs_kms::KmsServiceManager::new()),
)));
metadata_sys::update(BUCKET, BUCKET_VERSIONING_CONFIG, VERSIONING_XML.to_vec())
.await
.expect("persist readable companion config");
assert!(
exported_bucket_config(BUCKET, config_file)
.await
.expect("absent configuration")
.is_none()
);
let mut metadata = metadata_sys::get_config_from_disk(BUCKET)
.await
.expect("load source metadata");
match config_file {
BUCKET_NOTIFICATION_CONFIG => metadata.notification_config_xml = RAW_SECRET.to_vec(),
BUCKET_LIFECYCLE_CONFIG => metadata.lifecycle_config_xml = RAW_SECRET.to_vec(),
BUCKET_TAGGING_CONFIG => metadata.tagging_config_xml = RAW_SECRET.to_vec(),
_ => panic!("unexpected unreadable XML fixture"),
}
metadata
.save_with_store(Arc::clone(&env.ecstore))
.await
.expect("persist raw configuration with a failed parse");
crate::storage::storage_api::set_bucket_metadata(BUCKET.to_string(), metadata)
.await
.expect("publish unreadable XML fixture");
let ordinary = ExportBucketMetadata {}
.call(
admin_request(Method::GET, Uri::from_static("/rustfs/admin/v3/export-bucket-metadata"), Vec::new()),
Params::new(),
)
.await
.expect_err("persisted invalid XML must not disappear from an ordinary backup");
assert_eq!(ordinary.code(), &s3s::S3ErrorCode::InternalError);
let diagnostic = ExportBucketMetadata {}
.call(
admin_request(
Method::GET,
Uri::from_static("/rustfs/admin/v3/export-bucket-metadata?diagnostic=true"),
Vec::new(),
),
Params::new(),
)
.await
.expect("diagnostic export must identify the omitted configuration");
assert_eq!(diagnostic.output.0, StatusCode::OK);
let bytes = diagnostic
.output
.1
.collect()
.await
.expect("read diagnostic archive")
.to_bytes();
let mut archive = ZipArchive::new(Cursor::new(&bytes)).expect("open diagnostic archive");
let mut files = HashMap::new();
for index in 0..archive.len() {
let mut file = archive.by_index(index).expect("read diagnostic entry");
let mut content = Vec::new();
file.read_to_end(&mut content).expect("read diagnostic config");
assert!(!content.windows(RAW_SECRET.len()).any(|window| window == RAW_SECRET));
files.insert(file.name().to_string(), content);
}
assert!(!files.contains_key(&format!("_diagnostic/{BUCKET}/{config_file}")));
assert_eq!(files[&format!("_diagnostic/{BUCKET}/{BUCKET_VERSIONING_CONFIG}")], VERSIONING_XML);
let manifest: serde_json::Value =
serde_json::from_slice(&files[DIAGNOSTIC_EXPORT_MANIFEST]).expect("decode diagnostic manifest");
assert_eq!(
manifest,
serde_json::json!({
"version": 1,
"mode": "diagnostic",
"complete": false,
"errors": [{ "bucket": BUCKET, "config": config_file, "code": "configuration_unavailable" }],
})
);
assert_eq!(
persisted_xml(
&metadata_sys::get_config_from_disk(BUCKET)
.await
.expect("read unchanged source"),
config_file
),
RAW_SECRET
);
}
#[tokio::test]
#[serial_test::serial]
async fn unreadable_notification_xml_fails_backup_and_is_named_in_diagnostics() {
assert_unreadable_xml_export_is_explicit(BUCKET_NOTIFICATION_CONFIG).await;
}
#[tokio::test]
#[serial_test::serial]
async fn unreadable_lifecycle_xml_fails_backup_and_is_named_in_diagnostics() {
assert_unreadable_xml_export_is_explicit(BUCKET_LIFECYCLE_CONFIG).await;
}
#[tokio::test]
#[serial_test::serial]
async fn unreadable_tagging_xml_fails_backup_and_is_named_in_diagnostics() {
assert_unreadable_xml_export_is_explicit(BUCKET_TAGGING_CONFIG).await;
}
#[tokio::test]
#[serial_test::serial]
async fn diagnostic_export_isolated_errors_and_import_recovers_unreadable_targets() {
@@ -1611,62 +1665,18 @@ mod backup_zip_compatibility_tests {
.expect("publish unreadable targets fixture");
assert!(metadata_sys::get_bucket_targets_config(UNREADABLE).await.is_err());
// rustfs/backlog#2309: an ordinary export no longer fails closed on
// a configuration that is stored but unreadable. It names that one
// configuration in the bucket's own marker entry and keeps every
// readable configuration of every bucket, so one MinIO-origin blob
// cannot cost an operator the whole-cluster backup.
let ordinary = ExportBucketMetadata {}
.call(
admin_request(Method::GET, Uri::from_static("/rustfs/admin/v3/export-bucket-metadata"), Vec::new()),
Params::new(),
)
.await
.expect("one unreadable configuration must not abort the ordinary export");
assert_eq!(ordinary.output.0, StatusCode::OK);
assert!(!ordinary.headers.contains_key("x-rustfs-bucket-metadata-export"));
let ordinary_bytes = ordinary.output.1.collect().await.expect("read ordinary archive").to_bytes();
let mut ordinary_archive = ZipArchive::new(Cursor::new(&ordinary_bytes)).expect("open ordinary archive");
.expect_err("an ordinary backup must fail rather than omit an unreadable configuration");
assert_eq!(*ordinary.code(), s3s::S3ErrorCode::InternalError);
assert!(
ordinary_archive
.by_name(&format!("{HEALTHY}/{BUCKET_VERSIONING_CONFIG}"))
.is_ok(),
"a healthy bucket must still export while another bucket is unreadable"
);
assert!(
ordinary_archive
.by_name(&format!("{HEALTHY}/{EXPORT_UNREADABLE_MANIFEST}"))
.is_err(),
"a bucket whose configurations all read must carry no unreadable marker"
);
assert!(
ordinary_archive
.by_name(&format!("{UNREADABLE}/{BUCKET_TARGETS_FILE}"))
.is_err(),
"an unreadable targets blob must never be exported as a configuration"
);
let mut ordinary_marker = Vec::new();
ordinary_archive
.by_name(&format!("{UNREADABLE}/{EXPORT_UNREADABLE_MANIFEST}"))
.expect("the ordinary export must name the configuration it could not read")
.read_to_end(&mut ordinary_marker)
.expect("read unreadable marker");
assert!(
!ordinary_marker
.windows(SECRET.len())
.any(|window| window == SECRET.as_bytes())
);
let ordinary_marker: serde_json::Value = serde_json::from_slice(&ordinary_marker).expect("the marker must be JSON");
assert_eq!(ordinary_marker["bucket"], UNREADABLE);
assert_eq!(
ordinary_marker["unreadable"].as_array().map(Vec::len),
Some(1),
"only the configuration that could not be read may be marked: {ordinary_marker}"
);
assert_eq!(ordinary_marker["unreadable"][0]["config"], BUCKET_TARGETS_FILE);
assert!(
ordinary_marker["unreadable"][0]["error"].is_string(),
"the marker must carry the reason an operator needs to repair the bucket"
ordinary
.message()
.is_some_and(|message| message.contains(BUCKET_TARGETS_FILE))
);
let response = ExportBucketMetadata {}
@@ -1739,31 +1749,44 @@ mod backup_zip_compatibility_tests {
);
}
// The marker may be malformed, come last, or be removed while the
// reserved directory remains. None may allow an earlier config write.
// Diagnostic and historical partial-export markers may come last.
// Neither may allow an earlier configuration write or bucket creation.
for marker in [
DIAGNOSTIC_EXPORT_MANIFEST.to_string(),
DIAGNOSTIC_EXPORT_PREFIX.to_string(),
format!("{DIAGNOSTIC_EXPORT_PREFIX}/bucket/config"),
format!("diagnostic-never-created/{EXPORT_UNREADABLE_MANIFEST}"),
] {
let mut writer = ZipWriter::new(Cursor::new(Vec::new()));
writer
.start_file(format!("{HEALTHY}/{BUCKET_VERSIONING_CONFIG}"), SimpleFileOptions::default())
.expect("start ordinary config before diagnostic marker");
.expect("start ordinary config before rejected marker");
writer
.write_all(b"<VersioningConfiguration><Status>Suspended</Status></VersioningConfiguration>")
.expect("write ordinary config before diagnostic marker");
.expect("write ordinary config before rejected marker");
writer
.start_file(
format!("diagnostic-never-created/{BUCKET_VERSIONING_CONFIG}"),
SimpleFileOptions::default(),
)
.expect("start a nonexistent bucket config before diagnostic marker");
.expect("start a nonexistent bucket config before rejected marker");
writer.write_all(VERSIONING_XML).expect("write nonexistent bucket config");
let marker_content = if marker.ends_with(EXPORT_UNREADABLE_MANIFEST) {
serde_json::to_vec(&serde_json::json!({
"bucket": "diagnostic-never-created",
"unreadable": [
{ "config": BUCKET_TARGETS_FILE, "error": "targets could not be read" },
{ "config": OBJECT_LOCK_CONFIG, "error": "object lock could not be read" },
],
}))
.expect("encode historical partial-export marker")
} else {
b"not json".to_vec()
};
writer
.start_file(marker, SimpleFileOptions::default())
.expect("start diagnostic marker");
writer.write_all(b"not json").expect("write malformed diagnostic marker");
.expect("start rejected marker");
writer.write_all(&marker_content).expect("write rejected marker");
let error = ImportBucketMetadata {}
.call(
admin_request(
@@ -1774,7 +1797,7 @@ mod backup_zip_compatibility_tests {
Params::new(),
)
.await
.expect_err("diagnostic preflight must reject before any config write");
.expect_err("non-restorable archive preflight must reject before any config write");
assert_eq!(*error.code(), s3s::S3ErrorCode::InvalidRequest);
assert_eq!(
metadata_sys::get_config_from_disk(HEALTHY)
@@ -1788,7 +1811,7 @@ mod backup_zip_compatibility_tests {
.get_bucket_info("diagnostic-never-created", &BucketOptions::default())
.await
.is_err(),
"diagnostic preflight must reject before bucket creation"
"non-restorable archive preflight must reject before bucket creation"
);
}
@@ -1845,7 +1868,7 @@ mod backup_zip_compatibility_tests {
archive
.by_name(&format!("{UNREADABLE}/{EXPORT_UNREADABLE_MANIFEST}"))
.is_err(),
"the marker must disappear once the configuration reads again"
"ordinary backups must never contain partial-export markers"
);
}
@@ -1990,18 +2013,15 @@ mod backup_zip_compatibility_tests {
let _: ReplicationConfiguration =
deserialize(&restored.replication_config_xml).expect("old parser must read the newly exported archive payload");
// rustfs/backlog#2309: a MinIO-origin `.metadata.bin` stores its
// targets as a bare JSON array, which `BucketTargets` cannot decode.
// Since rustfs/rustfs#7172 that reads as "stored but unreadable" —
// which must mark one bucket's one configuration, not abort the
// whole-cluster export an operator needs to migrate away.
// The array-shaped compatibility fixture is unreadable as targets.
// A backup must not claim success after silently omitting that setting.
env.make_bucket(UNREADABLE_BUCKET, false).await;
metadata_sys::update(UNREADABLE_BUCKET, BUCKET_TARGETS_FILE, MINIO_ARRAY_TARGETS.to_vec())
.await
.expect("persist the MinIO-shaped targets blob");
.expect("persist the array-shaped targets blob");
metadata_sys::get_bucket_targets_config(UNREADABLE_BUCKET)
.await
.expect_err("a MinIO array-shaped targets blob must read as unreadable, not as an empty set");
.expect_err("an array-shaped targets blob must read as unreadable, not as an empty set");
let cluster_export = ExportBucketMetadata {}
.call(
@@ -2009,68 +2029,14 @@ mod backup_zip_compatibility_tests {
Params::new(),
)
.await
.expect("one bucket's unreadable configuration must not abort the whole-cluster export");
assert_eq!(cluster_export.output.0, StatusCode::OK);
let cluster_archive = cluster_export
.output
.1
.collect()
.await
.expect("read cluster archive body")
.to_bytes()
.to_vec();
let mut archive = ZipArchive::new(Cursor::new(&cluster_archive)).expect("open cluster archive");
// Every readable configuration of every other bucket still exports.
for (config_file, payload) in persisted_xml_fixtures() {
let mut exported_payload = Vec::new();
archive
.by_name(&format!("{BUCKET}/{config_file}"))
.unwrap_or_else(|_| panic!("cluster export must still contain {config_file}"))
.read_to_end(&mut exported_payload)
.unwrap_or_else(|_| panic!("read exported {config_file}"));
assert_eq!(exported_payload, payload, "one bad bucket must not change another bucket's export");
}
assert!(
archive.by_name(&format!("{BUCKET}/{EXPORT_UNREADABLE_MANIFEST}")).is_err(),
"a bucket whose configurations all read must carry no unreadable marker"
);
// The unreadable configuration is named rather than fabricated: no
// targets entry is exported for it at all.
assert!(
archive
.by_name(&format!("{UNREADABLE_BUCKET}/{BUCKET_TARGETS_FILE}"))
.is_err(),
"an unreadable targets blob must never be exported as a configuration"
);
let mut marker = Vec::new();
archive
.by_name(&format!("{UNREADABLE_BUCKET}/{EXPORT_UNREADABLE_MANIFEST}"))
.expect("the export must name the configuration it could not read")
.read_to_end(&mut marker)
.expect("read unreadable marker");
let marker: serde_json::Value = serde_json::from_slice(&marker).expect("the marker must be JSON");
assert_eq!(marker["bucket"], UNREADABLE_BUCKET);
.expect_err("one unreadable configuration must fail the ordinary whole-cluster backup");
assert_eq!(*cluster_export.code(), s3s::S3ErrorCode::InternalError);
assert_eq!(
marker["unreadable"].as_array().map(Vec::len),
Some(1),
"only the configuration that could not be read may be marked: {marker}"
);
assert_eq!(marker["unreadable"][0]["config"], BUCKET_TARGETS_FILE);
drop(archive);
// The marker cannot be misread as a configuration on the way back in:
// the importer switches on configuration names and ignores everything
// else, so the bucket's stored bytes come through untouched and the
// operator still has to repair them explicitly.
import_archive(cluster_archive).await;
let after_round_trip = metadata_sys::get_config_from_disk(UNREADABLE_BUCKET)
.await
.expect("the marked bucket must still load after the round trip");
assert_eq!(
after_round_trip.bucket_targets_config_json, MINIO_ARRAY_TARGETS,
"importing the marker must not overwrite or fabricate the bucket's targets configuration"
metadata_sys::get_config_from_disk(UNREADABLE_BUCKET)
.await
.expect("failed export must leave targets untouched")
.bucket_targets_config_json,
MINIO_ARRAY_TARGETS
);
}
}
rustfs/src/admin/handlers/replication.rs
+587 -31
View File
@@ -19,6 +19,7 @@ use crate::admin::runtime_sources::{
AppContext, app_context_from_req, current_notification_system_for_context, current_replication_pool_handle,
current_replication_stats_handle_for_context, current_runtime_port, object_store_from_req,
};
use crate::admin::storage_api::AdminVersioningConfigExt as _;
use crate::admin::storage_api::bucket::metadata::BUCKET_TARGETS_FILE;
use crate::admin::storage_api::bucket::metadata_sys;
use crate::admin::storage_api::bucket::metadata_sys::get_replication_config;
@@ -29,7 +30,7 @@ use crate::admin::storage_api::bucket::replication::{REMOTE_TARGET_READ_ONLY_HIS
use crate::admin::storage_api::bucket::target::{
BucketTarget, BucketTargetType, Credentials as TargetCredentials, LatencyStat, duration_from_secs_or_nanos,
};
use crate::admin::storage_api::bucket::target_sys::{BucketTargetError, BucketTargetSys, UnreadableTargetsPolicy};
use crate::admin::storage_api::bucket::target_sys::{BucketTargetError, BucketTargetSys};
use crate::admin::storage_api::contract::bucket::{BucketOperations, BucketOptions};
use crate::admin::storage_api::contract::list::ListOperations as _;
use crate::admin::storage_api::error::StorageError;
@@ -71,6 +72,90 @@ const EVENT_ADMIN_REMOTE_TARGET_STATE: &str = "admin_remote_target_state";
/// repaired first, then the rule is set.
const REPLACE_UNREADABLE_TARGETS_PARAM: &str = "replace-unreadable";
fn parse_remote_target_write_modes(uri: &http::Uri) -> S3Result<(bool, bool)> {
let mut update = None;
let mut replace_unreadable = None;
for (key, value) in url::form_urlencoded::parse(uri.query().unwrap_or_default().as_bytes()) {
let mode = match key.as_ref() {
"update" => &mut update,
REPLACE_UNREADABLE_TARGETS_PARAM => &mut replace_unreadable,
_ => continue,
};
if mode.is_some() {
return Err(s3_error!(InvalidRequest, "duplicate remote target write mode"));
}
*mode = Some(match value.as_ref() {
"true" => true,
"false" => false,
_ => return Err(s3_error!(InvalidRequest, "remote target write modes must be true or false")),
});
}
let update = update.unwrap_or(false);
let replace_unreadable = replace_unreadable.unwrap_or(false);
if update && replace_unreadable {
return Err(s3_error!(InvalidRequest, "replace-unreadable requires a complete target create request"));
}
Ok((update, replace_unreadable))
}
/// Repair decisions use the disk snapshot protected by the metadata transaction,
/// never the stale target cache retained after an unreadable configuration load.
async fn persist_remote_target_repair(bucket: &str, mut target: BucketTarget, incarnation: uuid::Uuid) -> S3Result<String> {
let mut discarded_unreadable = false;
let mut target_error = None;
let updated = metadata_sys::update_config_with(bucket, BUCKET_TARGETS_FILE, |metadata| {
if metadata.bucket_incarnation_id != incarnation {
return Err(StorageError::BucketNotFound(bucket.to_string()));
}
if target.target_type == BucketTargetType::ReplicationService
&& !metadata.versioning_config.as_ref().is_some_and(|config| config.enabled())
{
target_error = Some(BucketTargetError::BucketReplicationSourceNotVersioned {
bucket: bucket.to_string(),
});
return Err(StorageError::other("source bucket versioning changed before target repair"));
}
discarded_unreadable = metadata.bucket_targets_unreadable();
let mut targets = metadata.bucket_target_config.clone().unwrap_or_default();
let (arn, exists) = BucketTargetSys::remote_arn_for_targets(&targets.targets, &target, &target.deployment_id);
target.arn = arn;
if target.arn.is_empty() {
target_error = Some(BucketTargetError::BucketRemoteArnInvalid {
bucket: bucket.to_string(),
});
return Err(StorageError::other("remote target ARN is empty"));
}
if !exists {
BucketTargetSys::upsert_target_entry(&mut targets.targets, &target, false).map_err(|error| {
target_error = Some(error);
StorageError::other("remote target merge failed")
})?;
}
serde_json::to_vec(&targets).map_err(StorageError::other)
})
.await;
if let Some(error) = target_error {
return Err(map_bucket_target_error(error));
}
updated.map_err(ApiError::from)?;
// Persistence also publishes the fresh target set under the transaction
// guard. Publishing another snapshot here could undo a concurrent repair.
if discarded_unreadable {
warn!(
event = EVENT_ADMIN_REMOTE_TARGET_STATE,
component = LOG_COMPONENT_ADMIN,
subsystem = LOG_SUBSYSTEM_REPLICATION,
action = "set_remote_target",
result = "unreadable_targets_replaced",
bucket = %bucket,
arn = %target.arn,
"admin remote target state"
);
}
Ok(target.arn)
}
/// Field groups a `set-remote-target?update=true` request may modify, mirroring
/// MinIO's `TargetUpdateType` / `GetTargetUpdateOps` query contract: the update
/// overlays only the requested groups onto the stored target, so a client can
@@ -554,8 +639,7 @@ impl Operation for SetRemoteTargetHandler {
return Err(s3_error!(InvalidRequest, "bucket is required"));
};
let update = queries.get("update").is_some_and(|v| v == "true");
let replace_unreadable = queries.get(REPLACE_UNREADABLE_TARGETS_PARAM).is_some_and(|v| v == "true");
let (update, replace_unreadable) = parse_remote_target_write_modes(&req.uri)?;
warn!("set remote target, bucket: {}, update: {}", bucket, update);
@@ -634,6 +718,24 @@ impl Operation for SetRemoteTargetHandler {
let bucket_target_sys = BucketTargetSys::get();
if replace_unreadable {
// Validate the complete replacement before acquiring the metadata
// transaction; remote I/O must not extend the cluster-wide lock.
let incarnation = metadata_sys::capture_bucket_metadata_incarnation(bucket)
.await
.map_err(ApiError::from)?;
bucket_target_sys
.validate_target(bucket, &remote_target)
.await
.map_err(map_bucket_target_error)?;
// Match ordinary writers: local targets lock, then lifecycle and
// cluster metadata transaction guards acquired by the repair.
let _targets_guard = lock_bucket_targets_metadata(bucket).await;
let arn = persist_remote_target_repair(bucket, remote_target, incarnation).await?;
let arn_str = serde_json::to_string(&arn).map_err(|_| s3_error!(InternalError, "Failed to serialize target ARN"))?;
return Ok(S3Response::new((StatusCode::OK, Body::from(arn_str))));
}
if !update {
let (arn, exist) = bucket_target_sys
.get_remote_arn(bucket, Some(&remote_target), remote_target.deployment_id.as_str())
@@ -723,37 +825,10 @@ impl Operation for SetRemoteTargetHandler {
let arn = remote_target.arn.clone();
let unreadable_policy = if replace_unreadable {
UnreadableTargetsPolicy::Replace
} else {
UnreadableTargetsPolicy::FailClosed
};
let discarding_unreadable_targets = replace_unreadable
&& matches!(
bucket_target_sys.list_bucket_targets(bucket).await,
Err(BucketTargetError::BucketRemoteTargetsUnreadable { .. })
);
let targets = bucket_target_sys
.set_target(bucket, &remote_target, update, unreadable_policy)
.set_target(bucket, &remote_target, update)
.await
.map_err(map_bucket_target_error)?;
// Audited only where the discard actually happened: the flag alone is
// not an event, on a readable set it changes nothing, and a refused
// write must not leave a record claiming the set was replaced.
if discarding_unreadable_targets {
warn!(
event = EVENT_ADMIN_REMOTE_TARGET_STATE,
component = LOG_COMPONENT_ADMIN,
subsystem = LOG_SUBSYSTEM_REPLICATION,
action = "set_remote_target",
result = "unreadable_targets_replaced",
bucket = %bucket,
arn = %remote_target.arn,
"admin remote target state"
);
}
let json_targets = serde_json::to_vec(&targets).map_err(|e| {
error!("Serialization error: {}", e);
S3Error::with_message(S3ErrorCode::InternalError, "Failed to serialize targets".to_string())
@@ -2712,3 +2787,484 @@ mod tests {
assert_eq!(payload["ScannedVersions"], 7);
}
}
#[cfg(test)]
mod target_repair_tests {
use super::*;
use crate::admin::runtime_sources::publish_test_app_context;
use crate::admin::storage_api::bucket::metadata::BUCKET_VERSIONING_CONFIG;
use crate::admin::storage_api::bucket::target::BucketTargets;
use http::Extensions;
use http_body_util::BodyExt as _;
use rustfs_iam::store::{Store as _, object::IAM_CONFIG_PREFIX};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;
use tracing::instrument::WithSubscriber as _;
const ACCESS_KEY: &str = "TARGETREPAIRROOT";
const SECRET_KEY: &str = "targetRepairRootSecret123";
const BUCKET: &str = "target-repair";
const BAD_TARGETS: &[u8] = b"unreadable-targets";
const TARGET_REPAIR_ENV: [(&str, Option<&str>); 3] = [
("NO_PROXY", Some("*")),
("no_proxy", Some("*")),
("RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET", Some("true")),
];
struct RemoteTargetServer {
endpoint: String,
task: tokio::task::JoinHandle<()>,
}
impl Drop for RemoteTargetServer {
fn drop(&mut self) {
self.task.abort();
}
}
impl RemoteTargetServer {
async fn start() -> Self {
let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind remote target");
let endpoint = listener.local_addr().expect("remote target address").to_string();
let task = tokio::spawn(async move {
loop {
let (mut socket, _) = listener.accept().await.expect("accept remote target request");
let mut request = Vec::new();
let mut chunk = [0; 4096];
while !request.windows(4).any(|window| window == b"\r\n\r\n") {
let read = socket.read(&mut chunk).await.expect("read remote target request");
if read == 0 {
break;
}
request.extend_from_slice(&chunk[..read]);
}
let head = String::from_utf8_lossy(&request);
let first_line = head.lines().next().unwrap_or_default();
let body = if first_line.starts_with("HEAD /") {
""
} else if first_line.starts_with("GET /") && first_line.contains("versioning") {
"<VersioningConfiguration xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\"><Status>Enabled</Status></VersioningConfiguration>"
} else {
panic!("unexpected remote target request: {first_line}");
};
let response = format!(
"HTTP/1.1 200 OK\r\ncontent-type: application/xml\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",
body.len()
);
socket
.write_all(response.as_bytes())
.await
.expect("reply to remote target request");
}
});
Self { endpoint, task }
}
fn target(&self) -> BucketTarget {
BucketTarget {
source_bucket: BUCKET.to_string(),
endpoint: self.endpoint.clone(),
target_bucket: "remote".to_string(),
target_type: BucketTargetType::ReplicationService,
region: "us-east-1".to_string(),
credentials: Some(TargetCredentials {
access_key: "remote-access".to_string(),
secret_key: "remote-secret".to_string(),
..Default::default()
}),
..Default::default()
}
}
}
async fn test_env() -> (tempfile::TempDir, rustfs_test_utils::TestECStoreEnv) {
let _ = rustfs_credentials::init_global_action_credentials(Some(ACCESS_KEY.to_string()), Some(SECRET_KEY.to_string()));
let temp = tempfile::tempdir().expect("create repair test root");
let env = rustfs_test_utils::TestECStoreEnv::builder()
.base_dir(temp.path())
.disk_count(1)
.build()
.await;
env.make_bucket(BUCKET, true).await;
rustfs_iam::store::object::ObjectStore::new(Arc::clone(&env.ecstore))
.save_iam_config(serde_json::json!({"version": 1}), format!("{}/format.json", *IAM_CONFIG_PREFIX))
.await
.expect("seed IAM format");
let iam = rustfs_iam::build_iam_sys(Arc::clone(&env.ecstore))
.await
.expect("build test IAM");
publish_test_app_context(Arc::new(AppContext::with_default_interfaces(
Arc::clone(&env.ecstore),
iam,
Arc::new(rustfs_kms::KmsServiceManager::new()),
)));
metadata_sys::update(
BUCKET,
BUCKET_VERSIONING_CONFIG,
b"<VersioningConfiguration><Status>Enabled</Status></VersioningConfiguration>".to_vec(),
)
.await
.expect("persist source versioning");
(temp, env)
}
fn request(method: Method, query: &str, body: Vec<u8>) -> S3Request<Body> {
let operation = if method == Method::GET {
"list-remote-targets"
} else {
"set-remote-target"
};
S3Request {
input: Body::from(body),
method,
uri: format!("/rustfs/admin/v3/{operation}?bucket={BUCKET}&{query}")
.parse()
.expect("admin URI"),
headers: HeaderMap::new(),
extensions: Extensions::new(),
credentials: Some(s3s::auth::Credentials {
access_key: ACCESS_KEY.to_string(),
secret_key: s3s::auth::SecretKey::from(SECRET_KEY.to_string()),
}),
region: None,
service: None,
trailing_headers: None,
}
}
async fn seed_unreadable(env: &rustfs_test_utils::TestECStoreEnv) {
let mut metadata = metadata_sys::get_config_from_disk(BUCKET)
.await
.expect("read source metadata");
metadata.bucket_targets_config_json = BAD_TARGETS.to_vec();
metadata
.save_with_store(Arc::clone(&env.ecstore))
.await
.expect("persist unreadable targets");
crate::storage::storage_api::set_bucket_metadata(BUCKET.to_string(), metadata)
.await
.expect("publish unreadable targets");
assert!(BucketTargetSys::get().list_bucket_targets(BUCKET).await.is_err());
}
async fn repair(target: &BucketTarget, query: &str) -> S3Result<String> {
let body = serde_json::to_vec(&remote_target_admin_json(target).expect("serialize target request"))
.expect("encode target request");
let response = SetRemoteTargetHandler {}
.call(request(Method::PUT, query, body), Params::new())
.await?;
assert_eq!(response.output.0, StatusCode::OK);
let body = response.output.1.collect().await.expect("collect target ARN").to_bytes();
Ok(serde_json::from_slice(&body).expect("plain JSON ARN"))
}
#[tokio::test]
#[serial_test::serial]
async fn repair_existing_cached_target_persists_readable_targets_and_lists() {
temp_env::async_with_vars(TARGET_REPAIR_ENV, async {
let (_temp, env) = test_env().await;
let server = RemoteTargetServer::start().await;
let mut target = server.target();
target.arn = "arn:rustfs:replication:us-east-1:cached:remote".to_string();
let targets = BucketTargets {
targets: vec![target.clone()],
};
metadata_sys::update(BUCKET, BUCKET_TARGETS_FILE, serde_json::to_vec(&targets).expect("encode cached target"))
.await
.expect("seed cached target");
seed_unreadable(&env).await;
assert_eq!(
BucketTargetSys::get().get_remote_arn(BUCKET, Some(&target), "").await,
(target.arn.clone(), true)
);
assert!(
BucketTargetSys::get()
.get_remote_target_client(BUCKET, &target.arn)
.await
.is_some()
);
let arn = repair(&target, "replace-unreadable=true").await.expect("repair must commit");
assert_ne!(arn, target.arn);
assert!(
BucketTargetSys::get()
.get_remote_target_client(BUCKET, &target.arn)
.await
.is_none()
);
assert!(BucketTargetSys::get().get_remote_target_client(BUCKET, &arn).await.is_some());
let persisted = metadata_sys::get_config_from_disk(BUCKET)
.await
.expect("read repaired metadata");
assert!(!persisted.bucket_targets_unreadable());
let targets = persisted.bucket_target_config.expect("decode persisted repair");
assert_eq!(targets.targets.len(), 1);
assert_eq!(targets.targets[0].arn, arn);
assert_eq!(
targets.targets[0]
.credentials
.as_ref()
.expect("persist credentials")
.secret_key,
"remote-secret"
);
let list = ListRemoteTargetHandler {}
.call(request(Method::GET, "", Vec::new()), Params::new())
.await
.expect("list repaired targets");
assert_eq!(list.output.0, StatusCode::OK);
let listed: serde_json::Value =
serde_json::from_slice(&list.output.1.collect().await.expect("collect target list").to_bytes())
.expect("decode list");
assert_eq!(listed.as_array().expect("targets list").len(), 1);
assert_eq!(listed[0]["arn"], arn);
})
.await;
}
#[tokio::test]
#[serial_test::serial]
async fn repair_partial_update_and_invalid_flags_preserve_persisted_bytes() {
temp_env::async_with_vars(TARGET_REPAIR_ENV, async {
let (_temp, env) = test_env().await;
let server = RemoteTargetServer::start().await;
let mut target = server.target();
target.arn = "arn:rustfs:replication:us-east-1:cached:remote".to_string();
metadata_sys::update(
BUCKET,
BUCKET_TARGETS_FILE,
serde_json::to_vec(&BucketTargets {
targets: vec![target.clone()],
})
.expect("encode cached target"),
)
.await
.expect("seed cached target");
seed_unreadable(&env).await;
let file = metadata_sys::get_config_from_disk(BUCKET)
.await
.expect("read source metadata")
.save_file_path();
let before = crate::admin::storage_api::read_admin_config(Arc::clone(&env.ecstore), &file)
.await
.expect("read original bytes");
for query in [
"update=true&replace-unreadable=true",
"replace-unreadable=false&replace-unreadable=true",
"replace-unreadable=true&replace-unreadable=false",
"replace-unreadable=true&replace%2dunreadable=true",
"replace-unreadable=TRUE",
"replace-unreadable=1",
"replace-unreadable=",
"update=true&update=false",
"update=invalid",
] {
assert_eq!(
repair(&target, query).await.expect_err("invalid repair must fail").code(),
&S3ErrorCode::InvalidRequest
);
let after = crate::admin::storage_api::read_admin_config(Arc::clone(&env.ecstore), &file)
.await
.expect("read unchanged bytes");
assert_eq!(after, before, "rejected opt-in must not rewrite metadata");
}
})
.await;
}
#[tokio::test]
#[serial_test::serial]
async fn repair_with_stale_unreadable_cache_preserves_another_committed_repair() {
temp_env::async_with_vars(TARGET_REPAIR_ENV, async {
let (_temp, env) = test_env().await;
let first = RemoteTargetServer::start().await;
let second = RemoteTargetServer::start().await;
seed_unreadable(&env).await;
let first_arn = repair(&first.target(), "replace-unreadable=true")
.await
.expect("commit first repair");
// Model another node which still retains the original unreadable
// verdict when it begins its repair after this commit.
BucketTargetSys::get().mark_targets_unreadable(BUCKET).await;
let second_arn = repair(&second.target(), "replace-unreadable=true")
.await
.expect("merge second repair");
let persisted = metadata_sys::get_config_from_disk(BUCKET).await.expect("read both repairs");
let targets = persisted.bucket_target_config.expect("decode both repairs");
assert_eq!(targets.targets.len(), 2);
assert!(targets.targets.iter().any(|target| target.arn == first_arn));
assert!(targets.targets.iter().any(|target| target.arn == second_arn));
assert_eq!(
BucketTargetSys::get()
.list_bucket_targets(BUCKET)
.await
.expect("published repair")
.targets
.len(),
2
);
assert_eq!(
repair(&first.target(), "replace-unreadable=true")
.await
.expect("idempotent repair"),
first_arn
);
assert_eq!(
metadata_sys::get_config_from_disk(BUCKET)
.await
.expect("read repeated repair")
.bucket_target_config
.expect("decode repeated repair")
.targets
.len(),
2
);
})
.await;
}
#[tokio::test]
#[serial_test::serial]
async fn repair_transaction_rejects_a_bucket_recreated_after_target_validation() {
temp_env::async_with_vars(
TARGET_REPAIR_ENV,
Box::pin(async {
let (_temp, env) = test_env().await;
let server = RemoteTargetServer::start().await;
let target = server.target();
let incarnation = metadata_sys::capture_bucket_metadata_incarnation(BUCKET)
.await
.expect("capture original bucket");
BucketTargetSys::get()
.validate_target(BUCKET, &target)
.await
.expect("validate original source and remote target");
env.ecstore
.delete_bucket(BUCKET, &Default::default())
.await
.expect("delete original bucket");
env.make_bucket(BUCKET, true).await;
seed_unreadable(&env).await;
let recreated = metadata_sys::get_config_from_disk(BUCKET)
.await
.expect("load recreated bucket");
assert_ne!(recreated.bucket_incarnation_id, incarnation);
let file = recreated.save_file_path();
let before = crate::admin::storage_api::read_admin_config(Arc::clone(&env.ecstore), &file)
.await
.expect("read recreated bucket bytes");
let error = persist_remote_target_repair(BUCKET, target, incarnation)
.await
.expect_err("validation of a deleted bucket must not authorize repair of its replacement");
assert_eq!(error.code(), &S3ErrorCode::NoSuchBucket);
assert_eq!(
crate::admin::storage_api::read_admin_config(Arc::clone(&env.ecstore), &file)
.await
.expect("read rejected incarnation repair bytes"),
before
);
}),
)
.await;
}
#[tokio::test]
#[serial_test::serial]
async fn repair_transaction_rejects_versioning_suspended_after_target_validation() {
temp_env::async_with_vars(
TARGET_REPAIR_ENV,
Box::pin(async {
let (_temp, env) = test_env().await;
let server = RemoteTargetServer::start().await;
let target = server.target();
seed_unreadable(&env).await;
let incarnation = metadata_sys::capture_bucket_metadata_incarnation(BUCKET)
.await
.expect("capture source bucket");
BucketTargetSys::get()
.validate_target(BUCKET, &target)
.await
.expect("validate versioned source and remote target");
metadata_sys::update(
BUCKET,
BUCKET_VERSIONING_CONFIG,
b"<VersioningConfiguration><Status>Suspended</Status></VersioningConfiguration>".to_vec(),
)
.await
.expect("suspend source versioning after validation");
let suspended = metadata_sys::get_config_from_disk(BUCKET)
.await
.expect("load suspended source bucket");
assert_eq!(suspended.bucket_incarnation_id, incarnation);
assert!(!suspended.versioning_config.as_ref().expect("persisted versioning").enabled());
let file = suspended.save_file_path();
let before = crate::admin::storage_api::read_admin_config(Arc::clone(&env.ecstore), &file)
.await
.expect("read suspended bucket bytes");
let error = persist_remote_target_repair(BUCKET, target, incarnation)
.await
.expect_err("a target validated before suspension must not be committed");
assert_eq!(error.code(), &S3ErrorCode::InvalidRequest);
assert_eq!(
crate::admin::storage_api::read_admin_config(Arc::clone(&env.ecstore), &file)
.await
.expect("read rejected versioning repair bytes"),
before
);
}),
)
.await;
}
#[tokio::test]
#[serial_test::serial]
async fn failed_repair_transaction_never_reports_a_successful_replacement() {
temp_env::async_with_vars(TARGET_REPAIR_ENV, async {
let (_temp, env) = test_env().await;
let server = RemoteTargetServer::start().await;
seed_unreadable(&env).await;
let target = server.target();
BucketTargetSys::get()
.validate_target(BUCKET, &target)
.await
.expect("remote validation must succeed before injecting the metadata failure");
let file = metadata_sys::get_config_from_disk(BUCKET)
.await
.expect("read source metadata")
.save_file_path();
// Keep the source versioning and unreadable-target caches intact,
// but make the transaction's fresh disk load fail.
let corrupt = b"invalid metadata envelope".to_vec();
env.put_object_bytes(".rustfs.sys", &file, corrupt.clone()).await;
assert!(metadata_sys::get_config_from_disk(BUCKET).await.is_err());
let log = tempfile::NamedTempFile::new().expect("create captured log");
let writer = log.reopen().expect("open captured log writer");
let subscriber = tracing_subscriber::fmt()
.with_ansi(false)
.without_time()
.with_writer(writer)
.finish();
let error = repair(&target, "replace-unreadable=true")
.with_subscriber(subscriber)
.await
.expect_err("repair must fail on the unreadable metadata envelope");
assert_eq!(error.code(), &S3ErrorCode::InternalError);
let lines = std::fs::read_to_string(log.path()).expect("read captured log");
assert!(
!lines.contains("unreadable_targets_replaced"),
"a failed transaction must not claim success: {lines}"
);
assert_eq!(
crate::admin::storage_api::read_admin_config(Arc::clone(&env.ecstore), &file)
.await
.expect("read failed repair bytes"),
corrupt
);
})
.await;
}
}
rustfs/src/admin/storage_api.rs
+8 -17
View File
@@ -212,7 +212,6 @@ pub(crate) mod bucket_target_sys {
pub(crate) type S3ClientError = super::ecstore_bucket::bucket_target_sys::S3ClientError;
pub(crate) type SsecPassthroughCapability = super::ecstore_bucket::bucket_target_sys::SsecPassthroughCapability;
pub(crate) type TargetClient = super::ecstore_bucket::bucket_target_sys::TargetClient;
pub(crate) type UnreadableTargetsPolicy = super::ecstore_bucket::bucket_target_sys::UnreadableTargetsPolicy;
}
pub(crate) mod lifecycle {
@@ -299,10 +298,7 @@ pub(crate) mod metadata_sys {
use std::sync::Arc;
use rustfs_policy::policy::BucketPolicy;
use s3s::dto::{
BucketLifecycleConfiguration, NotificationConfiguration, ObjectLockConfiguration, ServerSideEncryptionConfiguration,
Tagging, VersioningConfiguration,
};
use s3s::dto::{ObjectLockConfiguration, ServerSideEncryptionConfiguration, VersioningConfiguration};
use time::OffsetDateTime;
use super::Result;
@@ -320,6 +316,13 @@ pub(crate) mod metadata_sys {
crate::storage::storage_api::update_bucket_metadata_config(bucket, config_file, data).await
}
pub(crate) async fn update_config_with<F>(bucket: &str, config_file: &str, mutate: F) -> Result<OffsetDateTime>
where
F: FnOnce(&BucketMetadata) -> Result<Vec<u8>> + Send,
{
super::ecstore_bucket::metadata_sys::update_config_with(bucket, config_file, mutate).await
}
pub(crate) async fn update_if_incarnation(
bucket: &str,
config_file: &str,
@@ -412,14 +415,6 @@ pub(crate) mod metadata_sys {
serde_json::from_slice(&metadata.bucket_targets_config_json).map_err(super::Error::other)
}
pub(crate) async fn get_lifecycle_config(bucket: &str) -> Result<(BucketLifecycleConfiguration, OffsetDateTime)> {
super::ecstore_bucket::metadata_sys::get_lifecycle_config(bucket).await
}
pub(crate) async fn get_notification_config(bucket: &str) -> Result<Option<NotificationConfiguration>> {
super::ecstore_bucket::metadata_sys::get_notification_config(bucket).await
}
pub(crate) async fn get_object_lock_config(bucket: &str) -> Result<(ObjectLockConfiguration, OffsetDateTime)> {
super::ecstore_bucket::metadata_sys::get_object_lock_config(bucket).await
}
@@ -442,10 +437,6 @@ pub(crate) mod metadata_sys {
super::ecstore_bucket::metadata_sys::get_sse_config(bucket).await
}
pub(crate) async fn get_tagging_config(bucket: &str) -> Result<(Tagging, OffsetDateTime)> {
super::ecstore_bucket::metadata_sys::get_tagging_config(bucket).await
}
pub(crate) async fn get_versioning_config(bucket: &str) -> Result<(VersioningConfiguration, OffsetDateTime)> {
super::ecstore_bucket::metadata_sys::get_versioning_config(bucket).await
}