mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-27 23:47:28 +00:00
fix(notify): emit delete webhooks for prefix deletes and align replication headers (#2383)
Co-authored-by: 安正超 <anzhengchao@gmail.com>
This commit is contained in:
@@ -336,9 +336,21 @@ pub struct EventArgs {
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl EventArgs {
|
impl EventArgs {
|
||||||
// Helper function to check if it is a copy request
|
/// True when the RustFS replication header is explicitly enabled (`true` or `1`).
|
||||||
|
///
|
||||||
|
/// Only `x-rustfs-source-replication-request` is considered here. Many clients (including the
|
||||||
|
/// console) send `x-minio-source-replication-request` for MinIO compatibility; treating that
|
||||||
|
/// as replication would suppress webhooks on normal browser deletes. Storage still honors both
|
||||||
|
/// prefixes when parsing the typed HTTP headers for `ObjectOptions`.
|
||||||
pub fn is_replication_request(&self) -> bool {
|
pub fn is_replication_request(&self) -> bool {
|
||||||
self.req_params.contains_key("x-rustfs-source-replication-request")
|
self.replication_header_value_true("x-rustfs-source-replication-request")
|
||||||
|
}
|
||||||
|
|
||||||
|
fn replication_header_value_true(&self, key: &str) -> bool {
|
||||||
|
self.req_params
|
||||||
|
.get(key)
|
||||||
|
.map(|v| v.eq_ignore_ascii_case("true") || v == "1")
|
||||||
|
.unwrap_or(false)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -524,3 +536,38 @@ mod tests {
|
|||||||
assert_eq!(glacier.restore_event_data.lifecycle_restore_storage_class, "GLACIER");
|
assert_eq!(glacier.restore_event_data.lifecycle_restore_storage_class, "GLACIER");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod event_args_tests {
|
||||||
|
use super::EventArgs;
|
||||||
|
use hashbrown::HashMap;
|
||||||
|
use rustfs_ecstore::store_api::ObjectInfo;
|
||||||
|
use rustfs_s3_common::EventName;
|
||||||
|
|
||||||
|
fn args_with_headers(pairs: &[(&str, &str)]) -> EventArgs {
|
||||||
|
let mut req_params = HashMap::new();
|
||||||
|
for (k, v) in pairs {
|
||||||
|
req_params.insert((*k).to_string(), (*v).to_string());
|
||||||
|
}
|
||||||
|
EventArgs {
|
||||||
|
event_name: EventName::ObjectRemovedDelete,
|
||||||
|
bucket_name: "b".to_string(),
|
||||||
|
object: ObjectInfo::default(),
|
||||||
|
req_params,
|
||||||
|
resp_elements: HashMap::new(),
|
||||||
|
version_id: String::new(),
|
||||||
|
host: String::new(),
|
||||||
|
port: 0,
|
||||||
|
user_agent: String::new(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn replication_request_requires_true_value() {
|
||||||
|
assert!(!args_with_headers(&[("x-rustfs-source-replication-request", "")]).is_replication_request());
|
||||||
|
assert!(!args_with_headers(&[("x-rustfs-source-replication-request", "false")]).is_replication_request());
|
||||||
|
assert!(args_with_headers(&[("x-rustfs-source-replication-request", "true")]).is_replication_request());
|
||||||
|
assert!(args_with_headers(&[("x-rustfs-source-replication-request", "True")]).is_replication_request());
|
||||||
|
assert!(!args_with_headers(&[("x-minio-source-replication-request", "true")]).is_replication_request());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -24,7 +24,7 @@ use crate::storage::concurrency::{
|
|||||||
};
|
};
|
||||||
use crate::storage::ecfs::*;
|
use crate::storage::ecfs::*;
|
||||||
use crate::storage::head_prefix::{head_prefix_not_found_message, probe_prefix_has_children};
|
use crate::storage::head_prefix::{head_prefix_not_found_message, probe_prefix_has_children};
|
||||||
use crate::storage::helper::OperationHelper;
|
use crate::storage::helper::{OperationHelper, spawn_background};
|
||||||
use crate::storage::options::{
|
use crate::storage::options::{
|
||||||
copy_dst_opts, copy_src_opts, del_opts, extract_metadata, extract_metadata_from_mime_with_object_name,
|
copy_dst_opts, copy_src_opts, del_opts, extract_metadata, extract_metadata_from_mime_with_object_name,
|
||||||
filter_object_metadata, get_content_sha256_with_query, get_opts, normalize_content_encoding_for_storage, put_opts,
|
filter_object_metadata, get_content_sha256_with_query, get_opts, normalize_content_encoding_for_storage, put_opts,
|
||||||
@@ -3824,7 +3824,7 @@ impl DefaultObjectUsecase {
|
|||||||
.as_ref()
|
.as_ref()
|
||||||
.map(|context| context.notify())
|
.map(|context| context.notify())
|
||||||
.unwrap_or_else(default_notify_interface);
|
.unwrap_or_else(default_notify_interface);
|
||||||
tokio::spawn(async move {
|
spawn_background(async move {
|
||||||
for res in delete_results {
|
for res in delete_results {
|
||||||
if let Some(dobj) = res.delete_object {
|
if let Some(dobj) = res.delete_object {
|
||||||
let event_name = if dobj.delete_marker {
|
let event_name = if dobj.delete_marker {
|
||||||
@@ -3996,7 +3996,21 @@ impl DefaultObjectUsecase {
|
|||||||
})
|
})
|
||||||
.await;
|
.await;
|
||||||
}
|
}
|
||||||
return Ok(S3Response::with_status(DeleteObjectOutput::default(), StatusCode::NO_CONTENT));
|
// Prefix/force-delete returns empty ObjectInfo; still emit bucket notification so webhooks match S3 DELETE.
|
||||||
|
helper = helper
|
||||||
|
.event_name(EventName::ObjectRemovedDelete)
|
||||||
|
.object(ObjectInfo {
|
||||||
|
name: key.clone(),
|
||||||
|
bucket: bucket.clone(),
|
||||||
|
..Default::default()
|
||||||
|
})
|
||||||
|
.version_id(String::new());
|
||||||
|
let result = Ok(S3Response::with_status(DeleteObjectOutput::default(), StatusCode::NO_CONTENT));
|
||||||
|
// Match non-empty delete path: capacity manager write-op telemetry.
|
||||||
|
let manager = get_capacity_manager();
|
||||||
|
manager.record_write_operation().await;
|
||||||
|
let _ = helper.complete(&result);
|
||||||
|
return result;
|
||||||
}
|
}
|
||||||
|
|
||||||
if obj_info.replication_status == ReplicationStatusType::Replica
|
if obj_info.replication_status == ReplicationStatusType::Replica
|
||||||
|
|||||||
@@ -31,7 +31,7 @@ use tokio::runtime::{Builder, Handle};
|
|||||||
|
|
||||||
/// Schedules an asynchronous task on the current runtime;
|
/// Schedules an asynchronous task on the current runtime;
|
||||||
/// if there is no runtime, creates a minimal runtime execution on a new thread.
|
/// if there is no runtime, creates a minimal runtime execution on a new thread.
|
||||||
fn spawn_background<F>(fut: F)
|
pub(crate) fn spawn_background<F>(fut: F)
|
||||||
where
|
where
|
||||||
F: Future<Output = ()> + Send + 'static,
|
F: Future<Output = ()> + Send + 'static,
|
||||||
{
|
{
|
||||||
|
|||||||
Reference in New Issue
Block a user