mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-29 01:29:00 +00:00
fix(bucket): propagate bucket deletion to peer metadata caches (backlog#646) (#4326)
The peer `DeleteBucketMetadata` RPC handler was a stub that returned success without doing anything, and the delete-bucket flow never sent the notification in the first place. As a result, after a bucket was deleted other nodes kept serving its stale cached metadata. Wire the whole path end to end: - ecstore: add an in-memory `remove_bucket_metadata` (free fn) and `BucketMetadataSys::remove`, the counterpart to `set_bucket_metadata`, and export it through the `api::bucket::metadata_sys` facade. - node_service: `handle_delete_bucket_metadata` now validates the bucket name and actually drops the cached metadata for it. - bucket_usecase: after a successful delete_bucket, notify peers via `notification_sys.delete_bucket_metadata` in the background, symmetric to the existing `notify_bucket_metadata_reload` path. Also update the delete-bucket-metadata unit test to assert the empty-bucket rejection instead of the old always-success stub, and drop an unused `tracing::debug` test import left over from #4322. Verified: cargo fmt; cargo check -p rustfs-ecstore; cargo test -p rustfs --lib --features rio-v2 test_delete_bucket_metadata_empty_bucket; arch guardrail scripts pass.
This commit is contained in:
@@ -99,7 +99,7 @@ pub mod bucket {
|
||||
get_lifecycle_config, get_logging_config, get_notification_config, get_object_lock_config,
|
||||
get_public_access_block_config, get_quota_config, get_replication_config, get_request_payment_config, get_sse_config,
|
||||
get_tagging_config, get_versioning_config, get_website_config, init_bucket_metadata_sys, list_bucket_targets,
|
||||
set_bucket_metadata, update,
|
||||
remove_bucket_metadata, set_bucket_metadata, update,
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -74,6 +74,17 @@ pub async fn set_bucket_metadata(bucket: String, bm: BucketMetadata) -> Result<(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Drop a bucket's cached metadata from the in-memory map.
|
||||
///
|
||||
/// This is the counterpart to [`set_bucket_metadata`] and is invoked when a
|
||||
/// bucket is deleted so peers stop serving stale cached configuration for it.
|
||||
/// Returns `true` if an entry was present.
|
||||
pub async fn remove_bucket_metadata(bucket: &str) -> Result<bool> {
|
||||
let sys = get_bucket_metadata_sys()?;
|
||||
let lock = sys.read().await;
|
||||
Ok(lock.remove(bucket).await)
|
||||
}
|
||||
|
||||
pub async fn get(bucket: &str) -> Result<Arc<BucketMetadata>> {
|
||||
let sys = get_bucket_metadata_sys()?;
|
||||
let lock = sys.read().await;
|
||||
@@ -360,6 +371,17 @@ impl BucketMetadataSys {
|
||||
}
|
||||
}
|
||||
|
||||
/// Remove a bucket's cached metadata from the in-memory map.
|
||||
///
|
||||
/// Returns `true` if an entry was present. Reserved meta buckets are ignored.
|
||||
pub async fn remove(&self, bucket: &str) -> bool {
|
||||
if is_meta_bucketname(bucket) {
|
||||
return false;
|
||||
}
|
||||
let mut map = self.metadata_map.write().await;
|
||||
map.remove(bucket).is_some()
|
||||
}
|
||||
|
||||
async fn _reset(&mut self) {
|
||||
let mut map = self.metadata_map.write().await;
|
||||
map.clear();
|
||||
|
||||
@@ -791,7 +791,6 @@ mod tests {
|
||||
use serde_json::json;
|
||||
use time::{OffsetDateTime, format_description::well_known::Rfc3339};
|
||||
use tokio::sync::mpsc;
|
||||
use tracing::debug;
|
||||
|
||||
#[test]
|
||||
fn test_heal_opts_serialization() {
|
||||
|
||||
@@ -452,6 +452,29 @@ fn notify_bucket_metadata_reload(
|
||||
});
|
||||
}
|
||||
|
||||
/// Notify peers to drop their cached metadata for a bucket that was just deleted,
|
||||
/// so they stop serving stale bucket configuration. Runs in the background to
|
||||
/// avoid blocking the delete response.
|
||||
fn notify_bucket_metadata_delete(bucket: String, request_context: Option<request_context::RequestContext>) {
|
||||
spawn_background_with_context(request_context, async move {
|
||||
if let Some(notification_sys) = current_notification_system() {
|
||||
for peer_err in notification_sys
|
||||
.delete_bucket_metadata(&bucket)
|
||||
.await
|
||||
.into_iter()
|
||||
.filter(|e| e.err.is_some())
|
||||
{
|
||||
warn!(
|
||||
bucket = %bucket,
|
||||
host = %peer_err.host,
|
||||
error = ?peer_err.err,
|
||||
"failed to notify peer to delete bucket metadata"
|
||||
);
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
fn validate_replication_config_targets(targets: &BucketTargets, config: &ReplicationConfiguration) -> S3Result<()> {
|
||||
let configured_arns = targets
|
||||
.targets
|
||||
@@ -1133,6 +1156,10 @@ impl DefaultBucketUsecase {
|
||||
warn!(bucket = %input.bucket, error = ?err, "site replication delete bucket hook failed");
|
||||
}
|
||||
|
||||
// Notify peers to drop their cached metadata for the now-deleted bucket.
|
||||
let request_context = req.extensions.get::<request_context::RequestContext>().cloned();
|
||||
notify_bucket_metadata_delete(input.bucket.clone(), request_context);
|
||||
|
||||
let result = Ok(S3Response::new(DeleteBucketOutput {}));
|
||||
let _ = helper.complete(&result);
|
||||
result
|
||||
|
||||
@@ -2601,18 +2601,18 @@ mod tests {
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_delete_bucket_metadata() {
|
||||
async fn test_delete_bucket_metadata_empty_bucket() {
|
||||
let service = create_test_node_service();
|
||||
|
||||
let request = Request::new(DeleteBucketMetadataRequest {
|
||||
bucket: "test-bucket".to_string(),
|
||||
});
|
||||
let request = Request::new(DeleteBucketMetadataRequest { bucket: String::new() });
|
||||
|
||||
let response = service.delete_bucket_metadata(request).await;
|
||||
assert!(response.is_ok());
|
||||
|
||||
// An empty bucket name is rejected before touching the metadata system.
|
||||
let delete_response = response.unwrap().into_inner();
|
||||
assert!(delete_response.success); // Currently returns success (todo implementation)
|
||||
assert!(!delete_response.success);
|
||||
assert!(delete_response.error_info.unwrap().contains("bucket name is missing"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
|
||||
@@ -17,7 +17,7 @@ use crate::storage::storage_api::rpc_consumer::node_service::contract::bucket::{
|
||||
BucketOptions, DeleteBucketOptions, MakeBucketOptions,
|
||||
};
|
||||
use crate::storage::storage_api::rpc_consumer::node_service::{
|
||||
DiskError, StoragePeerS3ClientExt as _, load_bucket_metadata, set_bucket_metadata,
|
||||
DiskError, StoragePeerS3ClientExt as _, load_bucket_metadata, remove_bucket_metadata, set_bucket_metadata,
|
||||
};
|
||||
use rustfs_common::heal_channel::HealOpts;
|
||||
use rustfs_protos::proto_gen::node_service::*;
|
||||
@@ -30,13 +30,26 @@ impl NodeService {
|
||||
request: Request<DeleteBucketMetadataRequest>,
|
||||
) -> Result<Response<DeleteBucketMetadataResponse>, Status> {
|
||||
let request = request.into_inner();
|
||||
let _bucket = request.bucket;
|
||||
let bucket = request.bucket;
|
||||
if bucket.is_empty() {
|
||||
return Ok(Response::new(DeleteBucketMetadataResponse {
|
||||
success: false,
|
||||
error_info: Some("bucket name is missing".to_string()),
|
||||
}));
|
||||
}
|
||||
|
||||
//todo
|
||||
Ok(Response::new(DeleteBucketMetadataResponse {
|
||||
success: true,
|
||||
error_info: None,
|
||||
}))
|
||||
// Drop the bucket's cached metadata on this peer so it stops serving
|
||||
// stale configuration after the bucket has been deleted.
|
||||
match remove_bucket_metadata(&bucket).await {
|
||||
Ok(_) => Ok(Response::new(DeleteBucketMetadataResponse {
|
||||
success: true,
|
||||
error_info: None,
|
||||
})),
|
||||
Err(err) => Ok(Response::new(DeleteBucketMetadataResponse {
|
||||
success: false,
|
||||
error_info: Some(err.to_string()),
|
||||
})),
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) async fn handle_load_bucket_metadata(
|
||||
|
||||
@@ -197,8 +197,8 @@ pub(crate) mod rpc_consumer {
|
||||
ECStore, Error, FileInfoVersions, LocalPeerS3Client, MetricType, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, ReadMultipleReq,
|
||||
ReadMultipleResp, ReadOptions, SERVICE_SIGNAL_REFRESH_CONFIG, SERVICE_SIGNAL_RELOAD_DYNAMIC, StorageDiskRpcExt,
|
||||
StoragePeerS3ClientExt, UpdateMetadataOpts, all_local_disk_path, collect_local_metrics, find_local_disk_by_ref,
|
||||
get_local_server_property, load_bucket_metadata, reload_transition_tier_config, set_bucket_metadata,
|
||||
validate_batch_read_version_item_count,
|
||||
get_local_server_property, load_bucket_metadata, reload_transition_tier_config, remove_bucket_metadata,
|
||||
set_bucket_metadata, validate_batch_read_version_item_count,
|
||||
};
|
||||
pub(crate) type StorageResult<T> = super::super::Result<T>;
|
||||
|
||||
@@ -1170,6 +1170,10 @@ pub(crate) async fn set_bucket_metadata(bucket: String, bm: BucketMetadata) -> R
|
||||
ecstore_bucket::metadata_sys::set_bucket_metadata(bucket, bm).await
|
||||
}
|
||||
|
||||
pub(crate) async fn remove_bucket_metadata(bucket: &str) -> Result<bool> {
|
||||
ecstore_bucket::metadata_sys::remove_bucket_metadata(bucket).await
|
||||
}
|
||||
|
||||
pub(crate) async fn update_bucket_metadata_config(
|
||||
bucket: &str,
|
||||
config_file: &str,
|
||||
|
||||
Reference in New Issue
Block a user