Compare commits

...

14 Commits

Author SHA1 Message Date
Zhengchao An f235e81755 fix(data-usage): make empty checks exhaustive (#5424) 2026-07-29 09:35:18 +00:00
Zhengchao An d42bc52f8b test(heal): avoid live disk removal race (#5421) 2026-07-29 17:29:18 +08:00
Zhengchao An d48870df97 fix(rpc): authenticate non-disk mutation bodies (#5425) 2026-07-29 09:17:30 +00:00
Zhengchao An 453e3d0faa ci(concurrency): inherit workspace lint policy (#5420) 2026-07-29 16:40:56 +08:00
cxymds b65210b1db fix(ilm): preserve restore source version mode (#5406)
* fix(ilm): preserve restore source version mode

* test(ilm): cover suspended null-version restore

* fix(ilm): reconcile worker results from journal
2026-07-29 16:16:40 +08:00
Zhengchao An a7f035a8c3 fix(ecstore): fail peer metadata reloads closed instead of caching fabricated defaults (#5396)
The LoadBucketMetadata peer-notification handler loaded bucket metadata
with the fabricating loader (ConfigNotFound -> BucketMetadata::new) and
unconditionally cached the result. On a transient read-quorum dip during
a reload notification, a peer cached an authoritative "no Object Lock"
default for a lock-enabled bucket, disabling the batch-delete retention
gate (object_lock_delete_check_required) on that node until the next
refresh, and wiping its bucket-target/durability sync state.

Production changes:
- New BucketMetadataSys::reload_from_store (metadata_sys::
  reload_bucket_metadata): the peer reload path uses the presence-aware
  loader and installs only metadata actually read from persisted
  storage. A load miss returns an error (surfaced to the notifying peer
  as success=false) and leaves the cache untouched; deletion still
  propagates only through the dedicated DeleteBucketMetadata
  notification.
- The reload runs under the outer metadata-sys write guard, load
  included, mirroring update(): every other cache installer holds that
  lock, so a reload snapshot can never land after - and roll back - a
  newer concurrent install (the stale-load lost-update from the
  review), and the install-plus-registry-sync sequence stays atomic
  against concurrent removes and reloads (previously only the set call
  was write-guarded, with the load outside any lock).
- The peer-visible miss error is a fixed string: the notifying peer
  substring-matches error text against network-failure needles
  (is_network_like_error), so interpolating a bucket name (e.g. a legal
  bucket literally named "unavailable") could mark a healthy peer
  offline.
- get_config's lazy insert routes through set(), picking up the
  negative-cache invalidation.

An earlier draft instead guarded set() with a per-config updated_at
freshness comparison. Adversarial validation rejected it (three roles
independently): update_config stamps with the handling node's wall
clock, so within the skew the cluster already tolerates (+/-300s RPC
auth window) a config rewritten with an earlier stamp - e.g. revoking a
public-read policy through a second node, or any same-field rewrite
after an NTP step-back - would be skipped by every peer forever,
silently pinning the revoked permissive config with no re-convergence
path (the 15-minute refresh also routed through the guard). Race
staleness is second-scale while skew is minute-scale, so no tolerance
bound can separate them; the write-guard serialization closes the same
race without clocks and preserves the refresh loop's unconditional
converge-to-disk property, which is the cluster's self-healing
mechanism.

Startup audit (BucketMetadataSys::init): concurrent_load's
insert-if-vacant still installs a fabricated default when a transient
miss hits at boot - indistinguishable from a legacy bucket without a
metadata file at this layer - bounded by the next successful persisted
load. Making the object-lock gate fail closed on such entries is filed
as a follow-up, alongside the bare "unavailable" needle in
is_network_like_error and the Swift cache-only metadata writes.

Tests:
- bucket::metadata_sys::tests::
  peer_reload_never_caches_fabricated_defaults_as_authoritative:
  miss installs nothing / miss keeps the existing entry intact
  (asserting the dedicated non-persisted error) / persisted reload
  converges the cache over a stale entry.
- node_service::tests::
  test_load_bucket_metadata_failure_skips_scanner_maintenance:
  a failed reload reports failure and does not advance scanner
  maintenance activity (previously recorded even on a miss).
- The handler success path stays uncovered at the RPC layer (needs an
  isolated global object layer, like the pre-existing ignored test);
  the composition is pinned at the sys level instead.

Verification:
- cargo fmt --check and cargo clippy --lib --tests clean on
  rustfs-ecstore and rustfs.
- Targeted suites green; full cargo test -p rustfs-ecstore --lib:
  3198/3200 with two parallelism-sensitive lock-test flakes from the
  known baseline (pass in isolation; a different pair flakes per run).
- Adversarial validation (high-risk tier, all seven roles as
  independent parallel reviewers) run per AGENTS.md; all findings
  fixed or rebutted with evidence, three out-of-scope findings filed
  as follow-up tasks.

Co-authored-by: houseme <housemecn@gmail.com>
2026-07-29 08:13:36 +00:00
Zhengchao An 87d97a5f48 fix(data-usage): preserve nonempty replication stats (#5422) 2026-07-29 15:59:04 +08:00
cxymds d9efd6b853 feat(ecstore): add remote snapshot lease RPCs (#5389)
* feat(ecstore): add local snapshot leases

* feat(ecstore): add remote snapshot lease RPCs

* fix(rpc): keep snapshot lease checks CI-compatible
2026-07-29 15:06:18 +08:00
cxymds 2801b2500d fix(tiering): gate remote version state writes (#5382)
* feat(tiering): model provider version capabilities

* feat(tiering): persist opaque remote versions

* fix(tiering): gate remote version state safely

* fix(tiering): gate remote version state writes

* fix(tiering): preserve remote version state on delete

* fix(tiering): accept unversioned transition responses

* fix(tiering): replay exact cleanup journals

* test(tiering): pin empty exact cleanup guard

* test(tiering): accept strict missing journal errors

* test(tiering): exercise free-version identity guard

* test(tiering): reach destination identity guard

* test(tiering): persist version identity drift

* test(tiering): bind version drift fixture

* test(config): keep fleet gate tests after constants
2026-07-29 15:06:08 +08:00
cxymds 02f4dbeb68 fix(heal): fail closed on ambiguous metadata rescue (#5361)
* fix(heal): fail closed on ambiguous metadata rescue

* fix(heal): preserve uncertain dangling state

* test(heal): satisfy strict clippy checks

* fix(heal): retain explicit version metadata recovery

* fix(heal): preserve metadata rescue diagnostics

---------

Co-authored-by: houseme <housemecn@gmail.com>
2026-07-29 15:06:01 +08:00
cxymds f7757e6437 test(ecstore): rendezvous concurrent resend commits (#5411)
* test(ecstore): rendezvous concurrent resend commits

* test(ecstore): satisfy multipart barrier lint
2026-07-29 15:05:53 +08:00
cxymds 1cb1b02b08 fix(ecstore): prevent deleted bucket recreation (#5380)
* fix(ecstore): prevent deleted bucket recreation

* fix(ecstore): reject empty metadata bucket names

* fix(ecstore): avoid recursive bucket metadata lookup

* fix(ecstore): bound lazy metadata future stack use

* test(ecstore): create bucket before metadata reload

---------

Co-authored-by: houseme <housemecn@gmail.com>
2026-07-29 14:49:18 +08:00
Zhengchao An 3fe74a5019 fix(swift): merge account and container metadata POSTs (#5414) 2026-07-29 06:36:07 +00:00
Zhengchao An 451cbc099b fix(tier): keep converging peers that reject a config reload (#5412) 2026-07-29 14:24:26 +08:00
45 changed files with 4188 additions and 833 deletions
Generated
+1
View File
@@ -11829,6 +11829,7 @@ dependencies = [
"futures-util",
"libc",
"pin-project-lite",
"slab",
"tokio",
]
+3
View File
@@ -10,6 +10,9 @@ description = "Shared concurrency contract types for RustFS - workload admission
keywords = ["rustfs", "concurrency", "admission", "backpressure", "workers"]
categories = ["concurrency", "filesystem"]
[lints]
workspace = true
[dependencies]
# Internal crates
rustfs-io-core = { workspace = true }
+33
View File
@@ -116,6 +116,27 @@ 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;
/// Request writing the complete remote-tier version state into object metadata.
///
/// This remains ineffective until
/// [`ENV_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED`] is also enabled.
pub const ENV_TIER_REMOTE_VERSION_STATE_WRITE: &str = "RUSTFS_TIER_REMOTE_VERSION_STATE_WRITE";
pub const DEFAULT_TIER_REMOTE_VERSION_STATE_WRITE: bool = false;
/// Operator-attested fleet-wide confirmation for
/// [`ENV_TIER_REMOTE_VERSION_STATE_WRITE`].
///
/// This flag is an operational contract, not automatic capability discovery.
/// Operators may enable it only after every node that can write or read
/// transitioned object metadata supports the remote version-state schema and
/// semantics. Keeping the confirmation separate makes a single-node request or
/// a writer whose local opt-in is removed fail closed.
pub const ENV_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED: &str = "RUSTFS_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED";
pub const DEFAULT_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED: bool = false;
const _: () = assert!(!DEFAULT_TIER_REMOTE_VERSION_STATE_WRITE);
const _: () = assert!(!DEFAULT_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED);
// =============================================================================
// Concurrent Request Fix - Timeout and Backpressure Configuration
// =============================================================================
@@ -617,3 +638,15 @@ pub const ENV_OBJECT_IO_RANDOM_READAHEAD_DISABLE_CONCURRENCY: &str = "RUSTFS_OBJ
/// Default read-ahead disable concurrency threshold: 4.
pub const DEFAULT_OBJECT_IO_RANDOM_READAHEAD_DISABLE_CONCURRENCY: usize = 4;
#[cfg(test)]
mod remote_version_state_tests {
#[test]
fn remote_version_state_gate_uses_stable_environment_names() {
assert_eq!(super::ENV_TIER_REMOTE_VERSION_STATE_WRITE, "RUSTFS_TIER_REMOTE_VERSION_STATE_WRITE");
assert_eq!(
super::ENV_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED,
"RUSTFS_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED"
);
}
}
+161 -11
View File
@@ -506,8 +506,35 @@ pub struct ReplicationStats {
}
impl ReplicationStats {
pub fn is_empty(&self) -> bool {
let Self {
pending_size,
replicated_size,
failed_size,
failed_count,
pending_count,
missed_threshold_size,
after_threshold_size,
missed_threshold_count,
after_threshold_count,
replicated_count,
} = self;
*pending_size == 0
&& *replicated_size == 0
&& *failed_size == 0
&& *failed_count == 0
&& *pending_count == 0
&& *missed_threshold_size == 0
&& *after_threshold_size == 0
&& *missed_threshold_count == 0
&& *after_threshold_count == 0
&& *replicated_count == 0
}
#[deprecated(note = "use is_empty instead")]
pub fn empty(&self) -> bool {
self.replicated_size == 0 && self.failed_size == 0 && self.failed_count == 0
self.is_empty()
}
}
@@ -520,16 +547,19 @@ pub struct ReplicationAllStats {
}
impl ReplicationAllStats {
pub fn is_empty(&self) -> bool {
let Self {
replica_size,
replica_count,
targets,
} = self;
*replica_size == 0 && *replica_count == 0 && targets.values().all(ReplicationStats::is_empty)
}
#[deprecated(note = "use is_empty instead")]
pub fn empty(&self) -> bool {
if self.replica_size != 0 && self.replica_count != 0 {
return false;
}
for v in self.targets.values() {
if !v.empty() {
return false;
}
}
true
self.is_empty()
}
}
@@ -783,7 +813,7 @@ impl DataUsageCache {
return Some(root);
}
let mut flat = self.flatten(&root);
if flat.replication_stats.as_ref().is_some_and(|stats| stats.empty()) {
if flat.replication_stats.as_ref().is_some_and(ReplicationAllStats::is_empty) {
flat.replication_stats = None;
}
Some(flat)
@@ -1582,6 +1612,126 @@ mod tests {
assert_eq!(map["BETWEEN_1024B_AND_1_MB"], u64::MAX);
}
#[test]
fn replication_stats_empty_checks_every_field() {
type SetField = fn(&mut ReplicationStats);
let cases: [(&str, SetField); 10] = [
("pending_size", |stats| stats.pending_size = 1),
("replicated_size", |stats| stats.replicated_size = 1),
("failed_size", |stats| stats.failed_size = 1),
("failed_count", |stats| stats.failed_count = 1),
("pending_count", |stats| stats.pending_count = 1),
("missed_threshold_size", |stats| stats.missed_threshold_size = 1),
("after_threshold_size", |stats| stats.after_threshold_size = 1),
("missed_threshold_count", |stats| stats.missed_threshold_count = 1),
("after_threshold_count", |stats| stats.after_threshold_count = 1),
("replicated_count", |stats| stats.replicated_count = 1),
];
assert!(ReplicationStats::default().is_empty());
for (field, set_nonzero) in cases {
let mut stats = ReplicationStats::default();
set_nonzero(&mut stats);
assert!(!stats.is_empty(), "{field} must make replication stats non-empty");
}
}
#[test]
fn replication_all_stats_empty_checks_aggregate_fields_independently() {
let cases = [
(
"replica_size",
ReplicationAllStats {
replica_size: 1,
..Default::default()
},
),
(
"replica_count",
ReplicationAllStats {
replica_count: 1,
..Default::default()
},
),
];
assert!(ReplicationAllStats::default().is_empty());
for (field, stats) in cases {
assert!(!stats.is_empty(), "{field} must make aggregate replication stats non-empty");
}
let empty_targets = ReplicationAllStats {
targets: HashMap::from([("arn:test:empty".to_string(), ReplicationStats::default())]),
..Default::default()
};
assert!(empty_targets.is_empty(), "all-empty targets must keep aggregate stats empty");
let stats = ReplicationAllStats {
targets: HashMap::from([
("arn:test:empty".to_string(), ReplicationStats::default()),
(
"arn:test:non-empty".to_string(),
ReplicationStats {
pending_count: 1,
..Default::default()
},
),
]),
..Default::default()
};
assert!(!stats.is_empty(), "a non-empty target must make aggregate replication stats non-empty");
}
#[test]
fn size_recursive_prunes_empty_and_preserves_pending_replication_stats() {
let root = hash_path("bucket");
let child = hash_path("bucket/child");
let mut cache = DataUsageCache::default();
cache.replace_hashed(&root, &None, &DataUsageEntry::default());
cache.replace_hashed(
&child,
&Some(root.clone()),
&DataUsageEntry {
replication_stats: Some(ReplicationAllStats::default()),
..Default::default()
},
);
assert!(
cache
.size_recursive("bucket")
.expect("bucket usage should flatten")
.replication_stats
.is_none()
);
cache.replace_hashed(
&child,
&Some(root.clone()),
&DataUsageEntry {
replication_stats: Some(ReplicationAllStats {
targets: HashMap::from([(
"arn:test:pending".to_string(),
ReplicationStats {
pending_count: 1,
..Default::default()
},
)]),
..Default::default()
}),
..Default::default()
},
);
let flattened = cache.size_recursive("bucket").expect("bucket usage should flatten");
let replication = flattened
.replication_stats
.expect("pending-only replication stats must survive pruning");
assert_eq!(replication.targets["arn:test:pending"].pending_count, 1);
}
#[test]
fn test_data_usage_cache_merge_adds_missing_child() {
let mut base = DataUsageCache::default();
@@ -23,7 +23,8 @@ use rustfs_protos::{
proto_gen::node_service::{
BatchGenerallyLockRequest, BatchGenerallyLockResponse, BatchReadVersionRequest, BatchReadVersionResponse,
GenerallyLockRequest, GenerallyLockResponse, GenerallyLockResult, PingRequest, PingResponse,
node_service_server::NodeService,
SnapshotLeaseMutationResponse, SnapshotLeaseReleaseRequest, SnapshotLeaseRenewRequest, SnapshotLeaseRequest,
SnapshotLeaseResponse, node_service_server::NodeService,
},
};
use std::pin::Pin;
@@ -104,6 +105,27 @@ impl NodeService for MinimalLockNodeService {
Err(Status::unimplemented("MinimalLockNodeService only supports lock RPCs"))
}
async fn acquire_snapshot_lease(
&self,
_request: Request<SnapshotLeaseRequest>,
) -> Result<Response<SnapshotLeaseResponse>, Status> {
Err(Status::unimplemented("MinimalLockNodeService only supports lock RPCs"))
}
async fn renew_snapshot_lease(
&self,
_request: Request<SnapshotLeaseRenewRequest>,
) -> Result<Response<SnapshotLeaseResponse>, Status> {
Err(Status::unimplemented("MinimalLockNodeService only supports lock RPCs"))
}
async fn release_snapshot_lease(
&self,
_request: Request<SnapshotLeaseReleaseRequest>,
) -> Result<Response<SnapshotLeaseMutationResponse>, Status> {
Err(Status::unimplemented("MinimalLockNodeService only supports lock RPCs"))
}
async fn lock(&self, request: Request<GenerallyLockRequest>) -> Result<Response<GenerallyLockResponse>, Status> {
let request = request.into_inner();
let args: LockRequest = match serde_json::from_str(&request.args) {
+4 -3
View File
@@ -127,8 +127,8 @@ pub mod bucket {
get_global_bucket_metadata_sys, 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, remove_bucket_metadata, set_bucket_metadata, update,
update_bucket_targets_under_transaction_lock, update_config_with,
init_bucket_metadata_sys, list_bucket_targets, reload_bucket_metadata, remove_bucket_metadata, set_bucket_metadata,
update, update_bucket_targets_under_transaction_lock, update_config_with,
};
}
@@ -305,7 +305,8 @@ pub mod disk {
CheckPartsResp, DeleteOptions, Disk, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskOption, DiskStore,
FileInfoVersions, FileReader, FileWriter, HEALING_MARKER_PATH, NsScannerOpenRequest, OldCurrentSize,
PartTransactionAction, RUSTFS_META_BUCKET, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp,
STORAGE_FORMAT_FILE, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, new_disk, validate_batch_read_version_item_count,
STORAGE_FORMAT_FILE, SnapshotLeaseToken, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, new_disk,
validate_batch_read_version_item_count,
};
pub use bytes::Bytes;
pub use endpoint::Endpoint;
@@ -8974,7 +8974,7 @@ mod tests {
.await
.expect("first worker result should persist");
assert_eq!(first.state, ManualTransitionJobState::Running);
assert_eq!(first.report.transition_completed, 1);
assert_eq!(first.report.transition_completed, 0);
assert_eq!(first.report.transition_failed, 0);
let duplicate = record_manual_transition_worker_result(
@@ -8987,11 +8987,11 @@ mod tests {
.await
.expect("duplicate worker result should be idempotent");
assert_eq!(duplicate.state, ManualTransitionJobState::Running);
assert_eq!(duplicate.report.transition_completed, 1);
assert_eq!(duplicate.report.transition_completed, 0);
assert_eq!(duplicate.report.transition_failed, 0);
let second_key = manual_transition_worker_result_task_key(&bucket, "logs/b", None);
let final_record = record_manual_transition_worker_result(
let pending_record = record_manual_transition_worker_result(
ecstore.clone(),
job_id,
&second_key,
@@ -9000,7 +9000,14 @@ mod tests {
)
.await
.expect("second distinct worker result should persist");
assert_eq!(pending_record.state, ManualTransitionJobState::Running);
assert_eq!(pending_record.report.transition_completed, 0);
assert_eq!(pending_record.report.transition_failed, 0);
let final_record =
reconcile_manual_transition_worker_results(ecstore.clone(), job_id, ManualTransitionQueueSnapshot::default())
.await
.expect("worker result journal should reconcile");
assert_eq!(final_record.state, ManualTransitionJobState::Partial);
assert_eq!(final_record.report.transition_completed, 1);
assert_eq!(final_record.report.transition_failed, 1);
@@ -9035,7 +9042,7 @@ mod tests {
.expect("worker result job record should save");
let task_key = manual_transition_worker_result_task_key(&bucket, "logs/fail", None);
let final_record = record_manual_transition_worker_result_with_reason(
let pending_record = record_manual_transition_worker_result_with_reason(
ecstore.clone(),
job_id,
&task_key,
@@ -9045,7 +9052,12 @@ mod tests {
)
.await
.expect("worker result with failure reason should persist");
assert!(pending_record.report.tier_failure_by_reason.is_empty());
assert_eq!(pending_record.report.transition_failed, 0);
let final_record = reconcile_manual_transition_worker_results(ecstore, job_id, ManualTransitionQueueSnapshot::default())
.await
.expect("worker failure reason should reconcile");
assert_eq!(
final_record
.report
@@ -11306,23 +11318,25 @@ mod tests {
// Distinct payloads with distinct sizes: a mixed-generation reassembly
// would produce bytes matching none of them (or fail the read outright).
let candidates: Vec<Vec<u8>> = (0..3)
let candidates: Vec<Vec<u8>> = (0..2)
.map(|g| {
let len = 4096 + g * 512;
vec![b'a' + g as u8; len]
})
.collect();
let commit_barrier = MultipartCommitBarrier::install(&bucket, object, MultipartCommitPause::PutPartBeforeLockLost);
let start = Arc::new(tokio::sync::Barrier::new(candidates.len() + 1));
let commit_barrier = MultipartCommitBarrier::install_for_arrivals(
&bucket,
object,
MultipartCommitPause::PutPartBeforeLockAcquire,
candidates.len(),
);
let mut tasks = tokio::task::JoinSet::new();
for payload in candidates.iter().cloned() {
let store = ecstore.clone();
let bucket = bucket.clone();
let upload_id = upload.upload_id.clone();
let start = Arc::clone(&start);
tasks.spawn(async move {
start.wait().await;
let mut data = PutObjReader::from_vec(payload.clone());
store
.put_object_part(&bucket, object, &upload_id, 1, &mut data, &ObjectOptions::default())
@@ -11330,11 +11344,10 @@ mod tests {
.map(|info| (info, payload))
});
}
start.wait().await;
// The first writer holds the uploadId commit lock while the other
// resends reach the same critical section. Releasing it proves the
// handoff without depending on saturated CI disk latency.
// Both writers finish streaming before racing for the uploadId commit
// lock. Two generations are sufficient to exercise the mixed-shard
// hazard, while each waiter sits behind at most one cross-disk rename.
commit_barrier.wait_until_paused().await;
commit_barrier.release();
@@ -1646,40 +1646,14 @@ pub async fn record_manual_transition_worker_result_with_reason(
job_id: Uuid,
task_key: &str,
result: ManualTransitionWorkerResult,
queue_snapshot: ManualTransitionQueueSnapshot,
_queue_snapshot: ManualTransitionQueueSnapshot,
failure_reason: Option<ManualTransitionWorkerFailureReason>,
) -> EcstoreResult<ManualTransitionJobRecord> {
let result_record = ManualTransitionWorkerResultRecord::new_with_reason(job_id, task_key, result, failure_reason);
if !save_manual_transition_worker_result_if_absent(api.clone(), &result_record).await? {
return load_manual_transition_job_record(api, job_id).await;
}
for _ in 0..4 {
let (mut record, etag) = load_manual_transition_job_record_with_etag(api.clone(), job_id).await?;
if record.is_terminal() {
return Ok(record);
}
record.record_worker_result_with_reason(result, queue_snapshot, failure_reason);
match save_manual_transition_job_record_if_current(api.clone(), &record, &etag).await {
Ok(()) => {
if record.is_terminal() {
delete_manual_transition_scope_admission_if_current(
api.clone(),
&record.scope_key,
record.job_id,
record.lease_id,
)
.await?;
} else {
renew_manual_transition_scope_admission_from_job(api, &record).await?;
}
return Ok(record);
}
Err(Error::PreconditionFailed) => continue,
Err(err) => return Err(err),
}
}
Err(Error::PreconditionFailed)
load_manual_transition_job_record(api, job_id).await
}
pub async fn renew_manual_transition_job_lease(
+352 -45
View File
@@ -23,7 +23,7 @@ use crate::error::{Error, Result, is_err_bucket_not_found};
use crate::runtime::sources as runtime_sources;
use crate::storage_api_contracts::heal::HealOperations as _;
use crate::storage_api_contracts::namespace::NamespaceLocking as _;
use crate::store::ECStore;
use crate::store::{ECStore, await_bucket_namespace_operation};
use futures::future::join_all;
use rustfs_common::heal_channel::HealOpts;
use rustfs_policy::policy::BucketPolicy;
@@ -37,13 +37,19 @@ use std::collections::HashSet;
use std::time::Duration;
use std::{collections::HashMap, sync::Arc};
use time::OffsetDateTime;
use tokio::sync::RwLock;
use tokio::sync::{Mutex, RwLock};
use tokio::time::sleep;
use tokio_util::sync::CancellationToken;
use tracing::{error, warn};
const BUCKET_METADATA_REFRESH_INTERVAL: Duration = Duration::from_secs(15 * 60);
#[derive(Clone, Copy)]
enum MetadataLoadMode {
Initial,
Refresh,
}
pub async fn init_bucket_metadata_sys(api: Arc<ECStore>, buckets: Vec<String>) {
// The metadata system is inherently per-store (it holds the store handle
// and that store's bucket cache), so it lives on the store's own instance
@@ -85,6 +91,20 @@ pub async fn set_bucket_metadata(bucket: String, bm: BucketMetadata) -> Result<(
Ok(())
}
/// Peer LoadBucketMetadata entry point; see
/// [`BucketMetadataSys::reload_from_store`] for the caching contract.
///
/// The outer write guard spans the disk load, mirroring [`update`]: every
/// other cache installer holds this lock (read or write), so the snapshot
/// read here can never land after — and roll back — a newer concurrent
/// install, and the install-plus-registry-sync sequence stays atomic
/// against concurrent removes and reloads.
pub async fn reload_bucket_metadata(bucket: &str) -> Result<()> {
let sys = get_bucket_metadata_sys()?;
let lock = sys.write().await;
lock.reload_from_store(bucket).await
}
/// Drop a bucket's cached metadata from the in-memory map.
///
/// This is the counterpart to [`set_bucket_metadata`] and is invoked when a
@@ -140,7 +160,8 @@ async fn refresh_buckets_metadata_once(sys: Arc<RwLock<BucketMetadataSys>>) {
for chunk in buckets.chunks(count) {
let sys = sys.read().await;
sys.concurrent_load(chunk, &mut failed_buckets).await;
sys.concurrent_load(chunk, &mut failed_buckets, MetadataLoadMode::Refresh)
.await;
}
if !failed_buckets.is_empty() {
@@ -461,6 +482,9 @@ const ABSENT_BUCKET_METADATA_MAX_ENTRIES: u64 = 10_000;
#[derive(Debug)]
pub struct BucketMetadataSys {
metadata_map: RwLock<HashMap<String, Arc<BucketMetadata>>>,
metadata_publish_lock: Mutex<()>,
#[cfg(test)]
lazy_load_lock_probe: std::sync::atomic::AtomicBool,
/// Buckets recently observed to have no persisted metadata. Serving the
/// fabricated default from here (instead of re-reading disk) keeps the
/// per-request cost of repeated lookups for such names bounded — without
@@ -476,6 +500,9 @@ impl BucketMetadataSys {
pub fn new(api: Arc<ECStore>) -> Self {
Self {
metadata_map: RwLock::new(HashMap::new()),
metadata_publish_lock: Mutex::new(()),
#[cfg(test)]
lazy_load_lock_probe: std::sync::atomic::AtomicBool::new(false),
absent_metadata: moka::future::Cache::builder()
.max_capacity(ABSENT_BUCKET_METADATA_MAX_ENTRIES)
.time_to_live(ABSENT_BUCKET_METADATA_TTL)
@@ -502,11 +529,13 @@ impl BucketMetadataSys {
loop {
if buckets.len() < count {
self.concurrent_load(buckets, &mut failed_buckets).await;
self.concurrent_load(buckets, &mut failed_buckets, MetadataLoadMode::Initial)
.await;
break;
}
self.concurrent_load(&buckets[..count], &mut failed_buckets).await;
self.concurrent_load(&buckets[..count], &mut failed_buckets, MetadataLoadMode::Initial)
.await;
buckets = &buckets[count..]
}
@@ -517,7 +546,7 @@ impl BucketMetadataSys {
Ok(())
}
async fn concurrent_load(&self, buckets: &[String], failed_buckets: &mut HashSet<String>) {
async fn concurrent_load(&self, buckets: &[String], failed_buckets: &mut HashSet<String>, mode: MetadataLoadMode) {
let mut futures = Vec::new();
for bucket in buckets.iter() {
@@ -525,16 +554,55 @@ impl BucketMetadataSys {
let bucket = bucket.clone();
futures.push(async move {
sleep(Duration::from_millis(30)).await;
let _ = api
.heal_bucket(
&bucket,
&HealOpts {
recreate: true,
..Default::default()
},
)
.await;
load_bucket_metadata_parse_with_presence(self.api.clone(), bucket.as_str(), true).await
match mode {
MetadataLoadMode::Initial => {
let _ = api
.heal_bucket(
&bucket,
&HealOpts {
recreate: true,
..Default::default()
},
)
.await;
let (bm, persisted) =
load_bucket_metadata_parse_with_presence(self.api.clone(), bucket.as_str(), true).await?;
if persisted {
self.set(bucket, Arc::new(bm)).await;
} else {
let _publish_guard = self.metadata_publish_lock.lock().await;
let mut map = self.metadata_map.write().await;
map.entry(bucket).or_insert_with(|| Arc::new(bm));
}
}
MetadataLoadMode::Refresh => {
let expected = self.metadata_map.read().await.get(&bucket).cloned();
let heal_lock = api.new_ns_lock(&bucket, &bucket).await?;
let heal_guard = heal_lock.get_read_lock(crate::set_disk::get_lock_acquire_timeout()).await?;
await_bucket_namespace_operation(
Some(&heal_guard),
&bucket,
"bucket metadata refresh heal",
api.heal_bucket(&bucket, &HealOpts::default()),
)
.await?;
drop(heal_guard);
let (bm, persisted) =
load_bucket_metadata_parse_with_presence(self.api.clone(), bucket.as_str(), true).await?;
let publish_lock = api.new_ns_lock(&bucket, &bucket).await?;
let guard = publish_lock
.get_read_lock(crate::set_disk::get_lock_acquire_timeout())
.await?;
if guard.is_lock_lost() {
return Err(Error::other(format!(
"bucket namespace lock was lost before bucket metadata refresh publish: {bucket}"
)));
}
self.publish_refresh_if_unchanged(&bucket, expected.as_ref(), bm, persisted)
.await;
}
}
Ok::<(), Error>(())
});
}
@@ -542,26 +610,7 @@ impl BucketMetadataSys {
for (idx, res) in results.into_iter().enumerate() {
match res {
Ok((bm, persisted)) => {
if let Some(bucket) = buckets.get(idx) {
if persisted {
self.set(bucket.clone(), Arc::new(bm)).await;
} else {
// A fabricated default (no persisted metadata
// readable right now) must never REPLACE an
// existing entry: the periodic refresh would
// otherwise downgrade a lock-enabled bucket to an
// authoritative "no lock" default on a transient
// ConfigNotFound, disabling the object-lock
// delete gate and wiping its target/durability
// sync state. Insert-if-vacant keeps the startup
// behavior for legacy buckets without a metadata
// file, atomically under the map write lock.
let mut map = self.metadata_map.write().await;
map.entry(bucket.clone()).or_insert_with(|| Arc::new(bm));
}
}
}
Ok(()) => {}
Err(e) => {
error!("Unable to load bucket metadata, will be retried: {:?}", e);
if let Some(bucket) = buckets.get(idx) {
@@ -572,6 +621,32 @@ impl BucketMetadataSys {
}
}
async fn publish_refresh_if_unchanged(
&self,
bucket: &str,
expected: Option<&Arc<BucketMetadata>>,
metadata: BucketMetadata,
persisted: bool,
) {
if !persisted {
return;
}
let _publish_guard = self.metadata_publish_lock.lock().await;
let metadata = Arc::new(metadata);
let mut map = self.metadata_map.write().await;
let unchanged = expected
.zip(map.get(bucket))
.is_some_and(|(expected, current)| Arc::ptr_eq(expected, current));
if !unchanged {
return;
}
map.insert(bucket.to_string(), Arc::clone(&metadata));
drop(map);
self.absent_metadata.invalidate(bucket).await;
sync_bucket_target_sys(bucket, &metadata).await;
sync_bucket_durability(bucket, &metadata);
}
pub async fn get(&self, bucket: &str) -> Result<Arc<BucketMetadata>> {
if is_meta_bucketname(bucket) {
return Err(Error::ConfigNotFound);
@@ -587,6 +662,7 @@ impl BucketMetadataSys {
pub async fn set(&self, bucket: String, bm: Arc<BucketMetadata>) {
if !is_meta_bucketname(&bucket) {
let _publish_guard = self.metadata_publish_lock.lock().await;
let mut map = self.metadata_map.write().await;
map.insert(bucket.clone(), bm.clone());
drop(map);
@@ -597,6 +673,43 @@ impl BucketMetadataSys {
}
}
/// Reload `bucket`'s metadata from this system's own store and cache it,
/// refusing to treat a load miss as authoritative (the peer
/// LoadBucketMetadata notification path, [`reload_bucket_metadata`]).
///
/// Only metadata actually read from persisted storage reaches the cache.
/// On a miss the fabricated default is discarded and an error is
/// returned: installing it would let a transient ConfigNotFound during
/// the notification overwrite a lock-enabled bucket's cached metadata
/// with an authoritative "no Object Lock" default, disabling the
/// batch-delete retention gate (`object_lock_delete_check_required`) on
/// this node until the next refresh. A miss is also not treated as
/// deletion: bucket deletion propagates through the dedicated
/// DeleteBucketMetadata notification ([`remove_bucket_metadata`]), which
/// is best-effort — a reload racing it can still re-install a just
/// deleted bucket's entry (pre-existing, bounded by the next delete or
/// restart) — but a reload miss removing entries would turn every
/// transient quorum dip into dropped metadata and spurious
/// target/durability teardown.
///
/// The peer-visible error text is deliberately fixed: the notifying peer
/// matches error strings against network-failure needles
/// (`is_network_like_error`), so interpolating a caller-controlled
/// bucket name here could mark a healthy peer offline.
///
/// Lock order: the caller holds the outer metadata-sys guard, and the
/// load acquires the namespace lock on the bucket's metadata config
/// object — the same `outer guard → meta-config namespace lock` order
/// `update`'s load takes; no path acquires these in reverse.
pub(crate) async fn reload_from_store(&self, bucket: &str) -> Result<()> {
let (bm, persisted) = load_bucket_metadata_parse_with_presence(self.api.clone(), bucket, true).await?;
if !persisted {
return Err(Error::other("no persisted bucket metadata readable; peer cache left unchanged"));
}
self.set(bucket.to_string(), Arc::new(bm)).await;
Ok(())
}
/// Remove a bucket's cached metadata from the in-memory map.
///
/// Returns `true` if an entry was present. Reserved meta buckets are ignored.
@@ -604,6 +717,7 @@ impl BucketMetadataSys {
if is_meta_bucketname(bucket) {
return false;
}
let _publish_guard = self.metadata_publish_lock.lock().await;
let mut map = self.metadata_map.write().await;
let removed = map.remove(bucket).is_some();
drop(map);
@@ -735,7 +849,24 @@ impl BucketMetadataSys {
return Ok((Arc::new(bm), true));
}
let (bm, persisted) = match load_bucket_metadata_parse_with_presence(self.api.clone(), bucket, true).await {
let lock = self.api.new_ns_lock(bucket, bucket).await?;
let guard = lock.get_read_lock(crate::set_disk::get_lock_acquire_timeout()).await?;
#[cfg(test)]
if self.lazy_load_lock_probe.load(std::sync::atomic::Ordering::Relaxed) {
let competing = self.api.new_ns_lock(bucket, bucket).await?;
assert!(
competing.get_write_lock(Duration::from_millis(20)).await.is_err(),
"lazy metadata IO must start while the bucket namespace read lock is held"
);
}
let (bm, persisted) = match await_bucket_namespace_operation(
Some(&guard),
bucket,
"lazy bucket metadata load",
Box::pin(load_bucket_metadata_parse_with_presence(self.api.clone(), bucket, true)),
)
.await
{
Ok(res) => res,
Err(err) => {
return if *self.initialized.read().await {
@@ -759,9 +890,33 @@ impl BucketMetadataSys {
// defaults for buckets listed on disk — legacy buckets without a
// metadata file — but never lets one replace an existing entry.)
if persisted {
await_bucket_namespace_operation(
Some(&guard),
bucket,
"lazy bucket metadata existence check",
Box::pin(async {
self.api
.peer_sys
.get_bucket_info(bucket, &crate::storage_api_contracts::bucket::BucketOptions::default())
.await
.map(|_| ())
.map_err(Into::into)
}),
)
.await?;
if guard.is_lock_lost() {
return Err(Error::other(format!(
"bucket namespace lock was lost before lazy bucket metadata publish: {bucket}"
)));
}
let _publish_guard = self.metadata_publish_lock.lock().await;
let mut map = self.metadata_map.write().await;
if let Some(current) = map.get(bucket) {
return Ok((Arc::clone(current), true));
}
map.insert(bucket.to_string(), bm.clone());
drop(map);
self.absent_metadata.invalidate(bucket).await;
sync_bucket_target_sys(bucket, &bm).await;
sync_bucket_durability(bucket, &bm);
} else {
@@ -1047,12 +1202,13 @@ mod tests {
/// Pins the fail-closed caching contract of the lazy `get_config` path
/// and the refresh no-replace rule: fabricated defaults are returned but
/// never served by the map-only `get()`, persisted metadata is cached on
/// lazy load (superseding a recorded absence), and a refresh-load miss
/// never replaces an existing entry.
/// lazy load (superseding a recorded absence), a refresh-load miss never
/// replaces an existing entry or heals a deleted bucket, and initial load
/// still heals buckets discovered from storage.
#[tokio::test]
async fn get_config_never_caches_fabricated_defaults_as_authoritative() {
let (_dirs, ecstore) = isolated_store_over_temp_disks().await;
let sys = BucketMetadataSys::new(ecstore);
let (dirs, ecstore) = isolated_store_over_temp_disks().await;
let sys = Arc::new(BucketMetadataSys::new(ecstore));
// (a) Miss: the fabricated default is returned but not cached.
let (bm, _) = sys
@@ -1078,6 +1234,9 @@ mod tests {
let mut persisted = BucketMetadata::new("absent-bucket");
persisted.policy_config_json = b"persisted-marker".to_vec();
sys.persist_and_set(persisted).await.expect("metadata should persist");
for dir in &dirs {
std::fs::create_dir_all(dir.path().join("absent-bucket")).expect("persisted bucket directory should be created");
}
sys.metadata_map.write().await.clear();
let _ = sys
.get_config("absent-bucket")
@@ -1089,14 +1248,47 @@ mod tests {
.expect("lazily loaded persisted metadata must be cached");
assert_eq!(cached.policy_config_json, b"persisted-marker".to_vec());
// (c) A refresh-load miss (no persisted metadata readable) must not
// replace an existing entry.
// (c) Persisted metadata left behind after physical deletion must not
// be lazily republished as a live bucket generation.
let mut deleted_lazy = BucketMetadata::new("deleted-lazy-bucket");
deleted_lazy.policy_config_json = b"stale-generation".to_vec();
sys.persist_and_set(deleted_lazy)
.await
.expect("stale metadata should persist");
sys.metadata_map.write().await.remove("deleted-lazy-bucket");
assert!(
sys.get_config("deleted-lazy-bucket").await.is_err(),
"lazy load must fail when the physical bucket no longer exists"
);
assert!(sys.get("deleted-lazy-bucket").await.is_err());
// (d) The namespace generation fence must be acquired before lazy
// metadata IO, so a writer can replace the generation atomically.
let fenced_bucket = "fenced-lazy-bucket";
for dir in &dirs {
std::fs::create_dir_all(dir.path().join(fenced_bucket)).unwrap();
}
let mut old_fenced = BucketMetadata::new(fenced_bucket);
old_fenced.policy_config_json = b"old-fenced-generation".to_vec();
sys.persist_and_set(old_fenced).await.unwrap();
sys.metadata_map.write().await.remove(fenced_bucket);
sys.lazy_load_lock_probe.store(true, std::sync::atomic::Ordering::Relaxed);
let (loaded, _) = sys.get_config(fenced_bucket).await.unwrap();
sys.lazy_load_lock_probe.store(false, std::sync::atomic::Ordering::Relaxed);
assert_eq!(loaded.policy_config_json, b"old-fenced-generation".to_vec());
// (e) A refresh-load miss for a bucket that still exists must not
// replace an existing entry with a fabricated default.
let mut kept = BucketMetadata::new("kept-bucket");
kept.policy_config_json = b"kept-marker".to_vec();
sys.set("kept-bucket".to_string(), Arc::new(kept)).await;
for dir in &dirs {
std::fs::create_dir_all(dir.path().join("kept-bucket")).expect("kept bucket directory should be created");
}
let mut failed = HashSet::new();
let refresh_targets = vec!["kept-bucket".to_string()];
sys.concurrent_load(&refresh_targets, &mut failed).await;
sys.concurrent_load(&refresh_targets, &mut failed, MetadataLoadMode::Refresh)
.await;
let kept = sys
.get("kept-bucket")
.await
@@ -1106,6 +1298,50 @@ mod tests {
b"kept-marker".to_vec(),
"a fabricated refresh default must not replace real metadata"
);
// (f) A stale cache entry for a physically deleted bucket must not
// recreate the bucket during periodic refresh.
sys.set("deleted-bucket".to_string(), Arc::new(BucketMetadata::new("deleted-bucket")))
.await;
let deleted_targets = vec!["deleted-bucket".to_string()];
sys.concurrent_load(&deleted_targets, &mut failed, MetadataLoadMode::Refresh)
.await;
assert!(
dirs.iter().all(|dir| !dir.path().join("deleted-bucket").exists()),
"periodic refresh must not recreate a bucket from stale cached metadata"
);
// (g) Metadata loaded for an old bucket generation must not replace
// metadata published by delete plus same-name recreation.
let old = Arc::new(BucketMetadata::new("recreated-bucket"));
sys.set("recreated-bucket".to_string(), Arc::clone(&old)).await;
let mut recreated = BucketMetadata::new("recreated-bucket");
recreated.policy_config_json = b"new-generation".to_vec();
sys.set("recreated-bucket".to_string(), Arc::new(recreated)).await;
let mut stale = BucketMetadata::new("recreated-bucket");
stale.policy_config_json = b"old-generation".to_vec();
sys.publish_refresh_if_unchanged("recreated-bucket", Some(&old), stale, true)
.await;
assert_eq!(sys.get("recreated-bucket").await.unwrap().policy_config_json, b"new-generation".to_vec());
// (f) Refresh retains periodic healing for a partially missing bucket.
sys.set("partial-bucket".to_string(), Arc::new(BucketMetadata::new("partial-bucket")))
.await;
for dir in dirs.iter().take(3) {
std::fs::create_dir_all(dir.path().join("partial-bucket")).unwrap();
}
sys.concurrent_load(&["partial-bucket".to_string()], &mut failed, MetadataLoadMode::Refresh)
.await;
assert!(dirs.iter().all(|dir| dir.path().join("partial-bucket").is_dir()));
// (g) Initial discovery retains the historical unconditional heal.
let initial_targets = vec!["initial-bucket".to_string()];
sys.concurrent_load(&initial_targets, &mut failed, MetadataLoadMode::Initial)
.await;
assert!(
dirs.iter().all(|dir| dir.path().join("initial-bucket").is_dir()),
"initial load must heal buckets discovered from storage"
);
}
#[tokio::test]
@@ -1130,12 +1366,18 @@ mod tests {
#[tokio::test]
async fn update_config_with_persists_tagging_rewrite_across_disk_reload() {
use crate::bucket::metadata::BUCKET_TAGGING_CONFIG;
use crate::storage_api_contracts::bucket::MakeBucketOptions;
use s3s::dto::Tag;
let (_dirs, ecstore) = isolated_store_over_temp_disks().await;
let mut sys = BucketMetadataSys::new(ecstore);
let bucket = "swift-tagging-bucket";
ecstore
.peer_sys
.make_bucket(bucket, &MakeBucketOptions::default())
.await
.expect("bucket volume should be created");
let mut sys = BucketMetadataSys::new(ecstore);
sys.persist_and_set(BucketMetadata::new(bucket))
.await
.expect("initial metadata should persist");
@@ -1240,6 +1482,71 @@ mod tests {
assert_eq!(tags.tag_set.len(), WRITERS, "every concurrent rewrite must survive: {tags:?}");
}
/// Pins the peer reload-notification contract (`reload_from_store`, the
/// LoadBucketMetadata RPC path): only metadata actually read from
/// persisted storage enters the cache. A load miss errors out and leaves
/// the cache untouched — it must neither install a fabricated default
/// for an unknown bucket nor replace an existing entry, since a
/// transient ConfigNotFound during the notification would otherwise
/// downgrade a lock-enabled bucket to an authoritative "no Object Lock"
/// default and disable the batch-delete retention gate on this peer.
#[tokio::test]
async fn peer_reload_never_caches_fabricated_defaults_as_authoritative() {
let (_dirs, ecstore) = isolated_store_over_temp_disks().await;
let sys = BucketMetadataSys::new(ecstore.clone());
// (a) Miss with no cached entry: the reload fails and installs nothing.
let err = sys
.reload_from_store("reload-bucket")
.await
.expect_err("a reload miss must be reported to the notifying peer");
assert!(
err.to_string().contains("no persisted bucket metadata readable"),
"the miss must surface through the dedicated non-persisted branch, got: {err}"
);
assert!(
sys.get("reload-bucket").await.is_err(),
"a reload miss must not install a fabricated default"
);
// (b) Miss with an existing entry: the reload fails and the entry
// (standing in for a lock-enabled bucket's metadata) survives intact.
let mut kept = BucketMetadata::new("reload-bucket");
kept.object_lock_config_xml = b"<ObjectLockConfiguration/>".to_vec();
sys.set("reload-bucket".to_string(), Arc::new(kept)).await;
assert!(sys.reload_from_store("reload-bucket").await.is_err());
let cached = sys
.get("reload-bucket")
.await
.expect("existing entry must survive a reload miss");
assert_eq!(
cached.object_lock_config_xml,
b"<ObjectLockConfiguration/>".to_vec(),
"a reload miss must not replace the cached entry with a fabricated default"
);
// (c) Persisted metadata reloads over a stale cached entry: the
// reload converges the cache to disk truth.
let mut persisted = BucketMetadata::new("reload-bucket");
persisted.policy_config_json = b"persisted-marker".to_vec();
sys.persist_and_set(persisted).await.expect("metadata should persist");
let mut stale = BucketMetadata::new("reload-bucket");
stale.policy_config_json = b"stale-cache-marker".to_vec();
sys.set("reload-bucket".to_string(), Arc::new(stale)).await;
sys.reload_from_store("reload-bucket")
.await
.expect("persisted metadata should reload");
let cached = sys
.get("reload-bucket")
.await
.expect("reloaded persisted metadata must be cached");
assert_eq!(
cached.policy_config_json,
b"persisted-marker".to_vec(),
"a reload must converge the cache to the persisted disk state"
);
}
fn target(bucket: &str, id: &str) -> BucketTarget {
BucketTarget {
source_bucket: bucket.to_string(),
+95 -2
View File
@@ -443,6 +443,16 @@ pub fn set_tonic_canonical_body_digest<T>(request: &mut tonic::Request<T>, canon
Ok(())
}
pub fn set_tonic_mutation_body_digest<T: rustfs_protos::CanonicalMutationBody>(
request: &mut tonic::Request<T>,
) -> std::io::Result<()> {
let canonical_body = request
.get_ref()
.canonical_body()
.map_err(|_| std::io::Error::other("RPC mutation body length cannot be represented"))?;
set_tonic_canonical_body_digest(request, &canonical_body)
}
pub fn verify_tonic_canonical_body_digest<T>(request: &tonic::Request<T>, canonical_body: &[u8]) -> std::io::Result<()> {
let version = request
.metadata()
@@ -466,7 +476,7 @@ pub fn verify_tonic_canonical_body_digest<T>(request: &tonic::Request<T>, canoni
Ok(())
}
/// Verify a mutating disk RPC's canonical body digest with a rolling-upgrade fallback.
/// Verify a mutating RPC's canonical body digest with a rolling-upgrade fallback.
///
/// When the request carries a real (non-`UNSIGNED-PAYLOAD`) content SHA-256 it is verified exactly
/// like [`verify_tonic_canonical_body_digest`]. The digest value is a member of the signed v2
@@ -497,7 +507,7 @@ fn verify_tonic_mutation_body_digest_with_strictness<T>(
Some(digest) if digest != UNSIGNED_PAYLOAD => verify_tonic_canonical_body_digest(request, canonical_body),
_ => {
// RUSTFS_COMPAT_TODO(disk-mutation-body-digest): accept digestless peers during rolling upgrades. Remove after the
// minimum supported RustFS peer version body-binds every mutating disk RPC.
// minimum supported RustFS peer version body-binds every mutating RPC.
if strict {
return Err(std::io::Error::other("RPC mutation requires a body-bound v2 signature"));
}
@@ -677,11 +687,28 @@ mod tests {
use crate::cluster::rpc::context_propagation::REQUEST_ID_HEADER;
use crate::runtime::sources as runtime_sources;
use http::{HeaderMap, Method};
use rustfs_protos::{
CanonicalMutationBody as _, PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS,
proto_gen::node_service::{Mss, SignalServiceRequest},
};
use std::collections::HashMap;
use std::io::{self, Write};
use std::sync::{Arc, Mutex};
use time::OffsetDateTime;
use tracing_subscriber::fmt::MakeWriter;
fn signal_service_request(signal: &str, sub_system: &str, dry_run: &str) -> SignalServiceRequest {
SignalServiceRequest {
vars: Some(Mss {
value: HashMap::from([
(PEER_RESTSIGNAL.to_string(), signal.to_string()),
(PEER_RESTSUB_SYS.to_string(), sub_system.to_string()),
(PEER_RESTDRY_RUN.to_string(), dry_run.to_string()),
]),
}),
}
}
#[derive(Clone, Default)]
struct CapturedLogs {
buffer: Arc<Mutex<Vec<u8>>>,
@@ -1596,6 +1623,72 @@ mod tests {
assert_eq!(stripped.to_string(), "RPC content SHA-256 mismatch");
}
#[test]
fn signal_service_mutation_contract_rejects_tampering_and_replay() {
ensure_test_rpc_secret();
let body = signal_service_request("2", "scanner", "false")
.canonical_body()
.expect("small signal request should encode");
let mut request = tonic::Request::new(());
set_tonic_canonical_body_digest(&mut request, &body).expect("canonical body digest should be attached");
let content_sha256 = request
.metadata()
.get(RPC_CONTENT_SHA256_HEADER)
.and_then(|value| value.to_str().ok());
let headers = gen_tonic_signature_headers("node-a:9000", "node_service.NodeService", "SignalService", content_sha256)
.expect("body-bound auth headers should build");
request.metadata_mut().as_mut().extend(headers.clone());
assert!(
verify_tonic_rpc_signature("node-a:9000", "/node_service.NodeService/SignalService", &headers).is_ok(),
"the first body-bound signal request must authenticate"
);
assert!(verify_tonic_mutation_body_digest(&request, &body).is_ok());
let tampered = signal_service_request("1", "scanner", "false")
.canonical_body()
.expect("small signal request should encode");
let error = verify_tonic_mutation_body_digest(&request, &tampered)
.expect_err("changing the signal must invalidate the signed digest");
assert_eq!(error.to_string(), "RPC content SHA-256 mismatch");
let replay = verify_tonic_rpc_signature("node-a:9000", "/node_service.NodeService/SignalService", &headers)
.expect_err("reusing the signal nonce must fail");
assert_eq!(replay.to_string(), "RPC request replay detected");
}
#[test]
#[serial_test::serial(rpc_body_digest_fallback_counter)]
fn signal_service_mutation_contract_preserves_rollout_fallback_and_strictness() {
let body = signal_service_request("2", "scanner", "false")
.canonical_body()
.expect("small signal request should encode");
let before = global_internode_metrics().snapshot().body_digest_fallback_total;
let digestless = tonic::Request::new(());
assert!(
verify_tonic_mutation_body_digest_with_strictness(&digestless, &body, false).is_ok(),
"old peers must remain compatible while the rollout gate is open"
);
assert_eq!(
global_internode_metrics().snapshot().body_digest_fallback_total,
before + 1,
"accepted digestless signal requests must be visible in the fallback metric"
);
let error = verify_tonic_mutation_body_digest_with_strictness(&digestless, &body, true)
.expect_err("strict mode must reject a digestless signal request");
assert_eq!(error.to_string(), "RPC mutation requires a body-bound v2 signature");
let mut bound = tonic::Request::new(());
set_tonic_canonical_body_digest(&mut bound, &body).expect("canonical body digest should be attached");
bound
.metadata_mut()
.as_mut()
.insert(RPC_AUTH_VERSION_HEADER, HeaderValue::from_static(RPC_AUTH_VERSION_V2));
assert!(verify_tonic_mutation_body_digest_with_strictness(&bound, &body, true).is_ok());
}
#[test]
fn nonce_cache_rejects_replay_after_wall_clock_regression() {
let now = Instant::now();
+3 -3
View File
@@ -30,9 +30,9 @@ pub use client::{
};
pub use http_auth::{
TONIC_RPC_PREFIX, build_auth_headers, gen_signature_headers, gen_tonic_signature_headers, normalize_tonic_rpc_audience,
set_tonic_canonical_body_digest, sign_ns_scanner_capability, sign_tonic_rpc_response_proof, verify_ns_scanner_capability,
verify_rpc_signature, verify_tonic_canonical_body_digest, verify_tonic_mutation_body_digest, verify_tonic_rpc_response_proof,
verify_tonic_rpc_signature,
set_tonic_canonical_body_digest, set_tonic_mutation_body_digest, sign_ns_scanner_capability, sign_tonic_rpc_response_proof,
verify_ns_scanner_capability, verify_rpc_signature, verify_tonic_canonical_body_digest, verify_tonic_mutation_body_digest,
verify_tonic_rpc_response_proof, verify_tonic_rpc_signature,
};
#[cfg(test)]
pub(crate) use internode_data_transport::TcpHttpInternodeDataTransport;
@@ -16,7 +16,7 @@ use crate::cluster::rpc::client::{
TonicInterceptor, embedded_tonic_status, gen_tonic_signature_interceptor, heal_control_time_out_client,
is_network_like_status, message_has_network_needle, node_service_time_out_client, tier_mutation_control_time_out_client,
};
use crate::cluster::rpc::{set_tonic_canonical_body_digest, verify_tonic_rpc_response_proof};
use crate::cluster::rpc::{set_tonic_canonical_body_digest, set_tonic_mutation_body_digest, verify_tonic_rpc_response_proof};
use crate::error::{Error, Result};
use crate::storage_api_contracts::internode::{
SCANNER_ACTIVITY_LEGACY_PROTOCOL_VERSION, SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION, SCANNER_ACTIVITY_PROTOCOL_VERSION,
@@ -50,6 +50,7 @@ use rustfs_protos::proto_gen::node_service::{
TierMutationPeerState, TierMutationPrepareRequest, node_service_client::NodeServiceClient,
tier_mutation_control_service_client::TierMutationControlServiceClient,
};
pub use rustfs_protos::{PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS};
use rustfs_protos::{TierMutationRpcPhase, evict_failed_connection};
use rustfs_utils::XHost;
use serde::{Deserialize, Serialize as _};
@@ -69,9 +70,6 @@ use tonic::transport::Channel;
use tracing::{debug, info, warn};
use uuid::Uuid;
pub const PEER_RESTSIGNAL: &str = "signal";
pub const PEER_RESTSUB_SYS: &str = "sub-sys";
pub const PEER_RESTDRY_RUN: &str = "dry-run";
pub const SERVICE_SIGNAL_REFRESH_CONFIG: u64 = 1;
pub const SERVICE_SIGNAL_RELOAD_DYNAMIC: u64 = 2;
const BACKGROUND_HEAL_STATUS_MAX_MESSAGE_SIZE: usize = 64 * 1024;
@@ -1160,10 +1158,11 @@ impl PeerRestClient {
self.finalize_result(
async {
let mut client = self.get_client().await?;
let request = Request::new(LoadBucketMetadataRequest {
let mut request = Request::new(LoadBucketMetadataRequest {
bucket: bucket.to_string(),
scanner_maintenance_change,
});
set_tonic_mutation_body_digest(&mut request)?;
let response = client.load_bucket_metadata(request).await?.into_inner();
if !response.success {
@@ -1183,9 +1182,10 @@ impl PeerRestClient {
self.finalize_result(
async {
let mut client = self.get_client().await?;
let request = Request::new(DeleteBucketMetadataRequest {
let mut request = Request::new(DeleteBucketMetadataRequest {
bucket: bucket.to_string(),
});
set_tonic_mutation_body_digest(&mut request)?;
let response = client.delete_bucket_metadata(request).await?.into_inner();
if !response.success {
@@ -1205,9 +1205,10 @@ impl PeerRestClient {
self.finalize_result(
async {
let mut client = self.get_client().await?;
let request = Request::new(DeletePolicyRequest {
let mut request = Request::new(DeletePolicyRequest {
policy_name: policy.to_string(),
});
set_tonic_mutation_body_digest(&mut request)?;
let response = client.delete_policy(request).await?.into_inner();
if !response.success {
@@ -1227,9 +1228,10 @@ impl PeerRestClient {
self.finalize_result(
async {
let mut client = self.get_client().await?;
let request = Request::new(LoadPolicyRequest {
let mut request = Request::new(LoadPolicyRequest {
policy_name: policy.to_string(),
});
set_tonic_mutation_body_digest(&mut request)?;
let response = client.load_policy(request).await?.into_inner();
if !response.success {
@@ -1249,11 +1251,12 @@ impl PeerRestClient {
self.finalize_result(
async {
let mut client = self.get_client().await?;
let request = Request::new(LoadPolicyMappingRequest {
let mut request = Request::new(LoadPolicyMappingRequest {
user_or_group: user_or_group.to_string(),
user_type,
is_group,
});
set_tonic_mutation_body_digest(&mut request)?;
let response = client.load_policy_mapping(request).await?.into_inner();
if !response.success {
@@ -1273,9 +1276,10 @@ impl PeerRestClient {
self.finalize_result(
async {
let mut client = self.get_client().await?;
let request = Request::new(DeleteUserRequest {
let mut request = Request::new(DeleteUserRequest {
access_key: access_key.to_string(),
});
set_tonic_mutation_body_digest(&mut request)?;
let response = client.delete_user(request).await?.into_inner();
if !response.success {
@@ -1295,9 +1299,10 @@ impl PeerRestClient {
self.finalize_result(
async {
let mut client = self.get_client().await?;
let request = Request::new(DeleteServiceAccountRequest {
let mut request = Request::new(DeleteServiceAccountRequest {
access_key: access_key.to_string(),
});
set_tonic_mutation_body_digest(&mut request)?;
let response = client.delete_service_account(request).await?.into_inner();
if !response.success {
@@ -1317,10 +1322,11 @@ impl PeerRestClient {
self.finalize_result(
async {
let mut client = self.get_client().await?;
let request = Request::new(LoadUserRequest {
let mut request = Request::new(LoadUserRequest {
access_key: access_key.to_string(),
temp,
});
set_tonic_mutation_body_digest(&mut request)?;
let response = client.load_user(request).await?.into_inner();
if !response.success {
@@ -1340,9 +1346,10 @@ impl PeerRestClient {
self.finalize_result(
async {
let mut client = self.get_client().await?;
let request = Request::new(LoadServiceAccountRequest {
let mut request = Request::new(LoadServiceAccountRequest {
access_key: access_key.to_string(),
});
set_tonic_mutation_body_digest(&mut request)?;
let response = client.load_service_account(request).await?.into_inner();
if !response.success {
@@ -1362,9 +1369,10 @@ impl PeerRestClient {
self.finalize_result(
async {
let mut client = self.get_client().await?;
let request = Request::new(LoadGroupRequest {
let mut request = Request::new(LoadGroupRequest {
group: group.to_string(),
});
set_tonic_mutation_body_digest(&mut request)?;
let response = client.load_group(request).await?.into_inner();
if !response.success {
@@ -1384,7 +1392,8 @@ impl PeerRestClient {
self.finalize_result(
async {
let mut client = self.get_client().await?;
let request = Request::new(ReloadSiteReplicationConfigRequest {});
let mut request = Request::new(ReloadSiteReplicationConfigRequest {});
set_tonic_mutation_body_digest(&mut request)?;
let response = client.reload_site_replication_config(request).await?.into_inner();
if !response.success {
@@ -1408,9 +1417,10 @@ impl PeerRestClient {
vars.insert(PEER_RESTSIGNAL.to_string(), sig.to_string());
vars.insert(PEER_RESTSUB_SYS.to_string(), sub_sys.to_string());
vars.insert(PEER_RESTDRY_RUN.to_string(), dry_run.to_string());
let request = Request::new(SignalServiceRequest {
let mut request = Request::new(SignalServiceRequest {
vars: Some(Mss { value: vars }),
});
set_tonic_mutation_body_digest(&mut request)?;
let response = client.signal_service(request).await?.into_inner();
if !response.success {
@@ -1479,7 +1489,8 @@ impl PeerRestClient {
self.finalize_result(
async {
let mut client = self.get_client().await?;
let request = Request::new(ReloadPoolMetaRequest {});
let mut request = Request::new(ReloadPoolMetaRequest {});
set_tonic_mutation_body_digest(&mut request)?;
let response = client.reload_pool_meta(request).await?.into_inner();
if !response.success {
@@ -1500,9 +1511,10 @@ impl PeerRestClient {
self.finalize_result(
async {
let mut client = self.get_client().await?;
let request = Request::new(StopRebalanceRequest {
let mut request = Request::new(StopRebalanceRequest {
expected_rebalance_id: expected_rebalance_id.unwrap_or_default().to_string(),
});
set_tonic_mutation_body_digest(&mut request)?;
let response = client.stop_rebalance(request).await?.into_inner();
if !response.success {
@@ -1523,7 +1535,8 @@ impl PeerRestClient {
self.finalize_result(
async {
let mut client = self.get_client().await?;
let request = Request::new(LoadRebalanceMetaRequest { start_rebalance });
let mut request = Request::new(LoadRebalanceMetaRequest { start_rebalance });
set_tonic_mutation_body_digest(&mut request)?;
let response = client.load_rebalance_meta(request).await?.into_inner();
@@ -1562,7 +1575,8 @@ impl PeerRestClient {
})
.collect::<Result<Vec<_>>>()?;
let mut client = self.get_client().await?;
let request = Request::new(StartDecommissionRequest { pool_indices });
let mut request = Request::new(StartDecommissionRequest { pool_indices });
set_tonic_mutation_body_digest(&mut request)?;
let response = client.start_decommission(request).await?.into_inner();
if !response.success {
@@ -1585,7 +1599,8 @@ impl PeerRestClient {
let pool_index = u32::try_from(pool_index)
.map_err(|_| Error::other(format!("decommission pool index {pool_index} exceeds RPC range")))?;
let mut client = self.get_client().await?;
let request = Request::new(CancelDecommissionRequest { pool_index });
let mut request = Request::new(CancelDecommissionRequest { pool_index });
set_tonic_mutation_body_digest(&mut request)?;
let response = client.cancel_decommission(request).await?.into_inner();
if !response.success {
@@ -1608,7 +1623,8 @@ impl PeerRestClient {
let pool_index = u32::try_from(pool_index)
.map_err(|_| Error::other(format!("decommission pool index {pool_index} exceeds RPC range")))?;
let mut client = self.get_client().await?;
let request = Request::new(ClearDecommissionRequest { pool_index });
let mut request = Request::new(ClearDecommissionRequest { pool_index });
set_tonic_mutation_body_digest(&mut request)?;
let response = client.clear_decommission(request).await?.into_inner();
if !response.success {
@@ -1628,10 +1644,14 @@ impl PeerRestClient {
pub async fn load_transition_tier_config(&self) -> Result<()> {
match self.load_transition_tier_config_outcome().await {
TierConfigReloadOutcome::Success => Ok(()),
TierConfigReloadOutcome::TransientReconnect(err) | TierConfigReloadOutcome::TransientRetrySameChannel(err) => {
self.finalize_result(Err(err)).await
}
TierConfigReloadOutcome::Terminal(err) => Err(err),
// Only a reconnect-class failure says anything about the channel.
// `finalize_result` marks the peer offline and evicts its connection
// whenever the message looks network-like, and a peer that answered
// and rejected the apply can easily report one ("release RPC failed:
// transport error"). Routing those through here would gate a healthy,
// responding peer out of every unrelated RPC.
TierConfigReloadOutcome::TransientReconnect(err) => self.finalize_result(Err(err)).await,
TierConfigReloadOutcome::TransientRetrySameChannel(err) | TierConfigReloadOutcome::Terminal(err) => Err(err),
}
}
@@ -1657,6 +1677,9 @@ impl PeerRestClient {
Err(err) => return tier_config_reload_connection_outcome(err),
};
let mut request = Request::new(LoadTransitionTierConfigRequest {});
if let Err(err) = set_tonic_mutation_body_digest(&mut request) {
return TierConfigReloadOutcome::Terminal(Error::other(err));
}
request.set_timeout(rustfs_protos::heal_control_execution_timeout());
let response = match client.load_transition_tier_config(request).await {
@@ -1710,13 +1733,24 @@ fn is_tier_config_reload_connection_failure(err: &Error) -> bool {
message_has_network_needle(&message)
}
/// Classifies a reload the peer answered but refused to apply.
///
/// The peer replied, so the channel is healthy and only the remote apply
/// failed. Those failures are transient by nature: the reload reads the tier
/// mutation intents and takes the distributed tier-config lock, both of which
/// fail while any other node is restarting or while the lock quorum is briefly
/// disturbed. Retiring the worker on the first such rejection leaves that peer
/// pinned to the old configuration with nothing left to heal it, so it answers
/// `TierNotFound` for a tier the rest of the cluster already committed until a
/// second admin mutation happens to spawn a fresh worker.
///
/// Convergence is the whole point of this path, so a rejection is retried on
/// the same channel. The worker's exponential backoff caps the cost at one
/// reload every `TIER_CONFIG_RELOAD_RETRY_CAP`, and `Terminal` stays reachable
/// for transport and gRPC status failures, which is where a genuinely
/// unrecoverable peer surfaces.
fn tier_config_reload_remote_failure(error_info: Option<String>) -> TierConfigReloadOutcome {
let error_info = error_info.unwrap_or_default();
if matches!(error_info.as_str(), "errServerNotInitialized" | "ServerNotInitialized") {
TierConfigReloadOutcome::TransientRetrySameChannel(Error::other(error_info))
} else {
TierConfigReloadOutcome::Terminal(Error::other(error_info))
}
TierConfigReloadOutcome::TransientRetrySameChannel(Error::other(error_info.unwrap_or_default()))
}
fn tier_config_reload_status_outcome(status: tonic::Status) -> TierConfigReloadOutcome {
@@ -1726,6 +1760,14 @@ fn tier_config_reload_status_outcome(status: tonic::Status) -> TierConfigReloadO
TierConfigReloadOutcome::TransientReconnect(status.into())
} else if status.code() == Code::Unknown && status.message().starts_with("Service was not ready:") {
TierConfigReloadOutcome::TransientRetrySameChannel(status.into())
} else if status.code() == Code::Unknown
&& is_tier_config_reload_connection_failure(&Error::other(status.message().to_string()))
{
// tonic reports a connection dropped mid-call as `Unknown` carrying the
// transport error text rather than as `Unavailable`, which is what a peer
// restarting under an active mutation produces. Reconnect and retry, so
// the restart does not permanently retire this peer's reload worker.
TierConfigReloadOutcome::TransientReconnect(status.into())
} else {
TierConfigReloadOutcome::Terminal(status.into())
}
@@ -2279,9 +2321,12 @@ mod tests {
tier_config_reload_status_outcome(tonic::Status::cancelled("request cancelled")),
TierConfigReloadOutcome::Terminal(_)
));
// A peer that answered and then refused the apply is retried rather than
// retired: the channel is healthy, so the rejection reflects remote state
// that the next attempt can find healed.
assert!(matches!(
tier_config_reload_remote_failure(Some("backend unavailable".to_string())),
TierConfigReloadOutcome::Terminal(_)
TierConfigReloadOutcome::TransientRetrySameChannel(_)
));
assert!(matches!(
tier_config_reload_remote_failure(Some("errServerNotInitialized".to_string())),
@@ -2305,6 +2350,50 @@ mod tests {
));
}
/// A tier mutation issued while another node restarts must still converge on
/// the nodes that stayed up. Those peers answer the reload RPC and reject the
/// apply, because reloading reads the tier mutation intents and takes the
/// distributed tier-config lock while the lock quorum is still disturbed.
/// Classifying those rejections as terminal retired the reload worker on its
/// first attempt and pinned the peer to the previous configuration, so it
/// served `TierNotFound` for an already-committed tier until an unrelated
/// second admin mutation spawned a new worker.
#[test]
fn tier_config_reload_retries_peers_that_reject_the_apply_mid_restart() {
for error_info in [
"Lock acquisition timeout for resource '.rustfs.sys/config/tier-config.bin.lock' after 5s",
"Resource '.rustfs.sys/config/tier-config.bin.lock' is already locked by node-3",
"Internal error: release RPC failed: transport error",
"save_config_with_opts: err: PreconditionFailed",
"erasure read quorum",
] {
assert!(
matches!(
tier_config_reload_remote_failure(Some(error_info.to_string())),
TierConfigReloadOutcome::TransientRetrySameChannel(_)
),
"a peer that rejected the apply must stay retryable so it converges: {error_info}"
);
}
// An absent error message is still a rejection, not a reason to stop.
assert!(matches!(
tier_config_reload_remote_failure(None),
TierConfigReloadOutcome::TransientRetrySameChannel(_)
));
// tonic surfaces a connection dropped mid-call as `Unknown`, not `Unavailable`.
assert!(matches!(
tier_config_reload_status_outcome(tonic::Status::unknown("transport error")),
TierConfigReloadOutcome::TransientReconnect(_)
));
// An `Unknown` that is not transport-shaped stays terminal.
assert!(matches!(
tier_config_reload_status_outcome(tonic::Status::unknown("peer response unknown")),
TierConfigReloadOutcome::Terminal(_)
));
}
#[tokio::test]
async fn tier_config_reload_single_attempt_clears_offline_gate_without_redial() {
let client = test_peer_client();
@@ -16,6 +16,7 @@ use crate::bucket::metadata_sys;
use crate::cluster::rpc::client::{
TonicInterceptor, gen_tonic_signature_interceptor, is_network_like_disk_error, node_service_time_out_client,
};
use crate::cluster::rpc::set_tonic_mutation_body_digest;
use crate::disk::error::DiskError;
use crate::disk::error::{Error, Result};
use crate::disk::error_reduce::{BUCKET_OP_IGNORED_ERRS, is_all_buckets_not_found, reduce_write_quorum_errs};
@@ -89,6 +90,22 @@ fn reduce_pool_write_quorum_errs(per_pool_errs: &[Option<Error>]) -> Option<Erro
reduce_write_quorum_errs(per_pool_errs, BUCKET_OP_IGNORED_ERRS, pool_write_quorum(per_pool_errs.len()))
}
fn resolve_heal_bucket_mode(opts: &mut HealOpts, pool_errs: &[Option<Error>]) -> Result<()> {
if opts.recreate {
return Ok(());
}
if let Some(err) = pool_errs
.iter()
.flatten()
.find(|err| **err != Error::DiskNotFound && **err != Error::VolumeNotFound)
{
return Err(err.clone());
}
opts.remove = is_all_buckets_not_found(pool_errs);
opts.recreate = !opts.remove;
Ok(())
}
#[async_trait]
pub trait PeerS3Client: Debug + Sync + Send + 'static {
async fn heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result<HealResultItem>;
@@ -159,10 +176,7 @@ impl S3PeerSys {
pool_errs.push(reduce_pool_write_quorum_errs(&per_pool_errs));
}
if !opts.recreate {
opts.remove = is_all_buckets_not_found(&pool_errs);
opts.recreate = !opts.remove;
}
resolve_heal_bucket_mode(&mut opts, &pool_errs)?;
let mut futures = Vec::new();
let heal_bucket_results = Arc::new(RwLock::new(vec![HealResultItem::default(); self.clients.len()]));
@@ -917,10 +931,11 @@ impl PeerS3Client for RemotePeerS3Client {
|| async {
let options: String = serde_json::to_string(opts)?;
let mut client = self.get_client().await?;
let request = Request::new(HealBucketRequest {
let mut request = Request::new(HealBucketRequest {
bucket: bucket.to_string(),
options,
});
set_tonic_mutation_body_digest(&mut request)?;
let response = client.heal_bucket(request).await?.into_inner();
if !response.success {
return if let Some(err) = response.error {
@@ -973,10 +988,11 @@ impl PeerS3Client for RemotePeerS3Client {
|| async {
let options = serde_json::to_string(opts)?;
let mut client = self.get_client().await?;
let request = Request::new(MakeBucketRequest {
let mut request = Request::new(MakeBucketRequest {
name: bucket.to_string(),
options,
});
set_tonic_mutation_body_digest(&mut request)?;
let response = client.make_bucket(request).await?.into_inner();
if !response.success {
@@ -1027,10 +1043,11 @@ impl PeerS3Client for RemotePeerS3Client {
let options = serde_json::to_string(opts)?;
let mut client = self.get_client().await?;
let request = Request::new(DeleteBucketRequest {
let mut request = Request::new(DeleteBucketRequest {
bucket: bucket.to_string(),
options,
});
set_tonic_mutation_body_digest(&mut request)?;
let response = client.delete_bucket(request).await?.into_inner();
if !response.success {
return if let Some(err) = response.error {
@@ -1618,6 +1635,30 @@ mod tests {
assert_eq!(err, Error::VolumeExists);
}
#[test]
fn heal_bucket_mode_fails_closed_on_incomplete_topology() {
let mut opts = HealOpts::default();
assert_eq!(
resolve_heal_bucket_mode(&mut opts, &[Some(Error::ErasureWriteQuorum)]),
Err(Error::ErasureWriteQuorum)
);
assert!(!opts.recreate);
assert!(!opts.remove);
}
#[test]
fn heal_bucket_mode_distinguishes_deleted_and_partial_buckets() {
let mut deleted = HealOpts::default();
resolve_heal_bucket_mode(&mut deleted, &[Some(Error::VolumeNotFound)]).unwrap();
assert!(deleted.remove);
assert!(!deleted.recreate);
let mut partial = HealOpts::default();
resolve_heal_bucket_mode(&mut partial, &[None, Some(Error::VolumeNotFound)]).unwrap();
assert!(!partial.remove);
assert!(partial.recreate);
}
#[tokio::test]
async fn test_make_bucket_reduces_quorum_by_pool_participants() {
let peer_sys = S3PeerSys {
+119 -3
View File
@@ -25,7 +25,7 @@ use crate::disk::error::{Error, Result};
use crate::disk::{
BatchReadVersionReq, BatchReadVersionResp, CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation,
DiskOption, FileInfoVersions, FileReader, FileWriter, PartTransactionAction, ReadMultipleReq, ReadMultipleResp, ReadOptions,
RenameDataResp, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, batch_read_version_one_by_one,
RenameDataResp, SnapshotLeaseToken, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, batch_read_version_one_by_one,
disk_store::{
DEFAULT_RUSTFS_DRIVE_ACTIVE_MONITORING, ENV_RUSTFS_DRIVE_ACTIVE_MONITORING, SKIP_IF_SUCCESS_BEFORE,
get_drive_active_check_interval, get_drive_active_check_timeout, get_drive_disk_info_timeout, get_drive_list_dir_timeout,
@@ -50,8 +50,9 @@ use rustfs_protos::proto_gen::node_service::{
DeleteVersionRequest, DeleteVersionsRequest, DeleteVolumeRequest, DiskInfoRequest, ListDirRequest, ListVolumesRequest,
MakeVolumeRequest, MakeVolumesRequest, PreparePartTransactionRequest, ReadAllRequest, ReadMetadataRequest,
ReadMultipleRequest, ReadMultipleResponse, ReadPartsRequest, ReadVersionRequest, ReadXlRequest, RenameDataRequest,
RenameFileRequest, SettlePartTransactionRequest, StatVolumeRequest, UpdateMetadataRequest, VerifyFileRequest,
WriteAllRequest, WriteMetadataRequest, node_service_client::NodeServiceClient,
RenameFileRequest, SettlePartTransactionRequest, SnapshotLeaseReleaseRequest, SnapshotLeaseRenewRequest,
SnapshotLeaseRequest, SnapshotLeaseResponse, StatVolumeRequest, UpdateMetadataRequest, VerifyFileRequest, WriteAllRequest,
WriteMetadataRequest, node_service_client::NodeServiceClient,
};
use serde::{Serialize, de::DeserializeOwned};
use std::{
@@ -100,6 +101,18 @@ const LOG_COMPONENT_ECSTORE: &str = "ecstore";
const LOG_SUBSYSTEM_REMOTE_DISK: &str = "remote_disk";
const EVENT_REMOTE_DISK_HEALTH: &str = "remote_disk_health";
const EVENT_REMOTE_DISK_RPC: &str = "remote_disk_rpc";
const SNAPSHOT_LEASE_PROTOCOL_VERSION: u32 = 1;
pub const REMOTE_SNAPSHOT_LEASE_TTL: Duration = Duration::from_secs(60);
fn snapshot_lease_token_from_response(response: SnapshotLeaseResponse) -> Result<SnapshotLeaseToken> {
if !response.success {
return Err(response.error.unwrap_or_default().into());
}
if response.protocol_version != SNAPSHOT_LEASE_PROTOCOL_VERSION {
return Err(Error::other("remote snapshot lease protocol is incompatible"));
}
SnapshotLeaseToken::from_slice(&response.token)
}
/// Bind a mutating disk RPC to its canonical body: the digest lands in the request metadata, and
/// the signing interceptor folds it (plus a replay-protected nonce) into the v2 signature scope
@@ -1784,6 +1797,81 @@ impl DiskAPI for RemoteDisk {
.await
}
async fn acquire_snapshot_lease(&self, volume: &str, path: &str) -> Result<SnapshotLeaseToken> {
self.execute_with_timeout(
|| async {
let mut client = self
.get_client()
.await
.map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
let mut request = Request::new(SnapshotLeaseRequest {
disk: self.endpoint.to_string(),
volume: volume.to_string(),
path: path.to_string(),
ttl_ms: u64::try_from(REMOTE_SNAPSHOT_LEASE_TTL.as_millis())
.map_err(|_| Error::other("snapshot lease TTL cannot be represented"))?,
});
let canonical_body = rustfs_protos::canonical_snapshot_lease_request_body(request.get_ref());
attach_mutation_body_digest(&mut request, canonical_body, "acquire_snapshot_lease")?;
let response = client.acquire_snapshot_lease(request).await?.into_inner();
snapshot_lease_token_from_response(response)
},
get_max_timeout_duration(),
)
.await
}
async fn renew_snapshot_lease(&self, volume: &str, path: &str, token: SnapshotLeaseToken) -> Result<SnapshotLeaseToken> {
self.execute_with_timeout(
|| async {
let mut client = self
.get_client()
.await
.map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
let mut request = Request::new(SnapshotLeaseRenewRequest {
disk: self.endpoint.to_string(),
volume: volume.to_string(),
path: path.to_string(),
token: token.as_bytes().to_vec().into(),
ttl_ms: u64::try_from(REMOTE_SNAPSHOT_LEASE_TTL.as_millis())
.map_err(|_| Error::other("snapshot lease TTL cannot be represented"))?,
});
let canonical_body = rustfs_protos::canonical_snapshot_lease_renew_request_body(request.get_ref());
attach_mutation_body_digest(&mut request, canonical_body, "renew_snapshot_lease")?;
let response = client.renew_snapshot_lease(request).await?.into_inner();
snapshot_lease_token_from_response(response)
},
get_max_timeout_duration(),
)
.await
}
async fn release_snapshot_lease(&self, volume: &str, path: &str, token: SnapshotLeaseToken) -> Result<()> {
self.execute_with_timeout(
|| async {
let mut client = self
.get_client()
.await
.map_err(|err| Error::other(format!("can not get client, err: {err}")))?;
let mut request = Request::new(SnapshotLeaseReleaseRequest {
disk: self.endpoint.to_string(),
volume: volume.to_string(),
path: path.to_string(),
token: token.as_bytes().to_vec().into(),
});
let canonical_body = rustfs_protos::canonical_snapshot_lease_release_request_body(request.get_ref());
attach_mutation_body_digest(&mut request, canonical_body, "release_snapshot_lease")?;
let response = client.release_snapshot_lease(request).await?.into_inner();
if !response.success {
return Err(response.error.unwrap_or_default().into());
}
Ok(())
},
get_max_timeout_duration(),
)
.await
}
#[tracing::instrument(level = "trace", skip_all)]
async fn write_metadata(&self, _org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()> {
trace!(
@@ -2930,6 +3018,34 @@ mod tests {
static INIT: Once = Once::new();
#[test]
fn snapshot_lease_response_requires_current_protocol_and_valid_token() {
let token = SnapshotLeaseToken::new();
let response = SnapshotLeaseResponse {
success: true,
token: token.as_bytes().to_vec().into(),
protocol_version: SNAPSHOT_LEASE_PROTOCOL_VERSION,
error: None,
};
assert_eq!(snapshot_lease_token_from_response(response).unwrap(), token);
let incompatible = SnapshotLeaseResponse {
success: true,
token: token.as_bytes().to_vec().into(),
protocol_version: SNAPSHOT_LEASE_PROTOCOL_VERSION + 1,
error: None,
};
assert!(snapshot_lease_token_from_response(incompatible).is_err());
let malformed = SnapshotLeaseResponse {
success: true,
token: Bytes::from_static(b"not-a-uuid"),
protocol_version: SNAPSHOT_LEASE_PROTOCOL_VERSION,
error: None,
};
assert!(snapshot_lease_token_from_response(malformed).is_err());
}
#[test]
fn list_volumes_decode_rejects_a_malformed_entry() {
let valid = serde_json::to_string(&VolumeInfo {
@@ -13,6 +13,7 @@
// limitations under the License.
use crate::cluster::rpc::client::{TonicInterceptor, gen_tonic_signature_interceptor, node_service_time_out_client};
use crate::cluster::rpc::set_tonic_mutation_body_digest;
use async_trait::async_trait;
use bytes::Bytes;
use rustfs_lock::{
@@ -313,10 +314,11 @@ impl LockClient for RemoteClient {
info!("remote acquire_exclusive for {}", request.resource);
let mut client = self.get_client().await?;
let resource_summary = request.resource.to_string();
let req = Request::new(GenerallyLockRequest {
let mut req = Request::new(GenerallyLockRequest {
args: serde_json::to_string(&request)
.map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?,
});
set_tonic_mutation_body_digest(&mut req)?;
let resp = match self.execute_rpc("lock", &resource_summary, client.lock(req)).await {
Ok(resp) => resp.into_inner(),
@@ -347,7 +349,7 @@ impl LockClient for RemoteClient {
let mut client = self.get_client().await?;
let resource_summary = Self::summarize_resources(requests);
let req = Request::new(BatchGenerallyLockRequest {
let mut req = Request::new(BatchGenerallyLockRequest {
args: requests
.iter()
.map(|request| {
@@ -355,6 +357,7 @@ impl LockClient for RemoteClient {
})
.collect::<Result<Vec<_>>>()?,
});
set_tonic_mutation_body_digest(&mut req)?;
let resp = match self
.execute_rpc("lock_batch", &resource_summary, client.lock_batch(req))
@@ -395,7 +398,8 @@ impl LockClient for RemoteClient {
.map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?;
let mut client = self.get_client().await?;
let resource_summary = unlock_request.resource.to_string();
let req = Request::new(GenerallyLockRequest { args: request_string });
let mut req = Request::new(GenerallyLockRequest { args: request_string });
set_tonic_mutation_body_digest(&mut req)?;
let resp = self
.execute_rpc("release", &resource_summary, client.un_lock(req))
.await?
@@ -414,7 +418,7 @@ impl LockClient for RemoteClient {
let unlock_requests = lock_ids.iter().map(Self::create_unlock_request).collect::<Vec<_>>();
let mut client = self.get_client().await?;
let resource_summary = Self::summarize_resources(&unlock_requests);
let req = Request::new(BatchGenerallyLockRequest {
let mut req = Request::new(BatchGenerallyLockRequest {
args: unlock_requests
.iter()
.map(|request| {
@@ -422,6 +426,7 @@ impl LockClient for RemoteClient {
})
.collect::<Result<Vec<_>>>()?,
});
set_tonic_mutation_body_digest(&mut req)?;
let resp = self
.execute_rpc("release_batch", &resource_summary, client.un_lock_batch(req))
@@ -440,10 +445,11 @@ impl LockClient for RemoteClient {
let refresh_request = Self::create_unlock_request(lock_id);
let mut client = self.get_client().await?;
let resource_summary = refresh_request.resource.to_string();
let req = Request::new(GenerallyLockRequest {
let mut req = Request::new(GenerallyLockRequest {
args: serde_json::to_string(&refresh_request)
.map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?,
});
set_tonic_mutation_body_digest(&mut req)?;
let resp = self
.execute_rpc("refresh", &resource_summary, client.refresh(req))
.await?
@@ -459,10 +465,11 @@ impl LockClient for RemoteClient {
let force_request = Self::create_unlock_request(lock_id);
let mut client = self.get_client().await?;
let resource_summary = force_request.resource.to_string();
let req = Request::new(GenerallyLockRequest {
let mut req = Request::new(GenerallyLockRequest {
args: serde_json::to_string(&force_request)
.map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?,
});
set_tonic_mutation_body_digest(&mut req)?;
let resp = self
.execute_rpc("force_release", &resource_summary, client.force_un_lock(req))
.await?
@@ -483,10 +490,11 @@ impl LockClient for RemoteClient {
let mut client = self.get_client().await?;
// Try to acquire a very short-lived lock to test availability
let req = Request::new(GenerallyLockRequest {
let mut req = Request::new(GenerallyLockRequest {
args: serde_json::to_string(&status_request)
.map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?,
});
set_tonic_mutation_body_digest(&mut req)?;
// Try exclusive lock first with very short timeout
let resp = match self.execute_rpc("check_status", &resource_summary, client.lock(req)).await {
@@ -497,10 +505,11 @@ impl LockClient for RemoteClient {
if resp.success {
// If we successfully acquired the lock, the resource was free.
// Immediately release it on a best-effort basis.
let release_req = Request::new(GenerallyLockRequest {
let mut release_req = Request::new(GenerallyLockRequest {
args: serde_json::to_string(&status_request)
.map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?,
});
set_tonic_mutation_body_digest(&mut release_req)?;
let _ = self
.execute_rpc("check_status_release", &resource_summary, client.un_lock(release_req))
.await;
+8
View File
@@ -1365,6 +1365,14 @@ impl DiskAPI for LocalDiskWrapper {
.await
}
async fn renew_snapshot_lease(&self, volume: &str, path: &str, token: SnapshotLeaseToken) -> Result<SnapshotLeaseToken> {
self.track_disk_health(
|| async { self.disk.renew_snapshot_lease(volume, path, token).await },
get_max_timeout_duration(),
)
.await
}
async fn delete_data_dir(&self, volume: &str, path: &str, opts: DeleteOptions) -> Result<DataDirDeleteStatus> {
self.track_disk_health(
|| async { self.disk.delete_data_dir(volume, path, opts).await },
+26 -2
View File
@@ -7908,6 +7908,23 @@ impl DiskAPI for LocalDisk {
}
}
async fn renew_snapshot_lease(&self, volume: &str, path: &str, token: SnapshotLeaseToken) -> Result<SnapshotLeaseToken> {
let key = SnapshotLeaseKey {
volume: volume.to_string(),
path: path.to_string(),
};
let mut registry = self.snapshot_leases.lock().await;
let Some(entry) = registry.entries.get_mut(&key) else {
return Err(DiskError::FileNotFound);
};
if entry.deleting || !entry.tokens.remove(&token) {
return Err(DiskError::FileNotFound);
}
let renewed = SnapshotLeaseToken::new();
entry.tokens.insert(renewed);
Ok(renewed)
}
async fn delete_data_dir(&self, volume: &str, path: &str, opts: DeleteOptions) -> Result<DataDirDeleteStatus> {
let key = SnapshotLeaseKey {
volume: volume.to_string(),
@@ -14957,6 +14974,13 @@ mod test {
.acquire_snapshot_lease(volume, &data_dir)
.await
.expect("second lease should be acquired");
let renewed = disk
.renew_snapshot_lease(volume, &data_dir, first)
.await
.expect("first lease should renew atomically");
disk.release_snapshot_lease(volume, &data_dir, first)
.await
.expect("the superseded token should be idempotent");
let status = disk
.delete_data_dir(
volume,
@@ -14976,9 +15000,9 @@ mod test {
Bytes::from_static(b"later")
);
disk.release_snapshot_lease(volume, &data_dir, first)
disk.release_snapshot_lease(volume, &data_dir, renewed)
.await
.expect("first lease release should succeed");
.expect("renewed lease release should succeed");
assert!(
disk.read_all(volume, &first_part).await.is_ok(),
"one remaining lease must keep the data directory"
+22
View File
@@ -79,6 +79,18 @@ impl SnapshotLeaseToken {
pub fn new() -> Self {
Self(Uuid::new_v4())
}
pub fn from_slice(bytes: &[u8]) -> Result<Self> {
let uuid = Uuid::from_slice(bytes).map_err(|_| Error::other("invalid snapshot lease token"))?;
if uuid.is_nil() {
return Err(Error::other("invalid snapshot lease token"));
}
Ok(Self(uuid))
}
pub fn as_bytes(&self) -> &[u8; 16] {
self.0.as_bytes()
}
}
impl Default for SnapshotLeaseToken {
@@ -284,6 +296,13 @@ impl DiskAPI for Disk {
}
}
async fn renew_snapshot_lease(&self, volume: &str, path: &str, token: SnapshotLeaseToken) -> Result<SnapshotLeaseToken> {
match self {
Disk::Local(local_disk) => local_disk.renew_snapshot_lease(volume, path, token).await,
Disk::Remote(remote_disk) => remote_disk.renew_snapshot_lease(volume, path, token).await,
}
}
async fn delete_data_dir(&self, volume: &str, path: &str, opts: DeleteOptions) -> Result<DataDirDeleteStatus> {
match self {
Disk::Local(local_disk) => local_disk.delete_data_dir(volume, path, opts).await,
@@ -694,6 +713,9 @@ pub trait DiskAPI: Debug + Send + Sync + 'static {
async fn release_snapshot_lease(&self, _volume: &str, _path: &str, _token: SnapshotLeaseToken) -> Result<()> {
Err(Error::other("snapshot leases are not supported by this disk"))
}
async fn renew_snapshot_lease(&self, _volume: &str, _path: &str, _token: SnapshotLeaseToken) -> Result<SnapshotLeaseToken> {
Err(Error::other("snapshot leases are not supported by this disk"))
}
async fn delete_data_dir(&self, volume: &str, path: &str, opts: DeleteOptions) -> Result<DataDirDeleteStatus> {
self.delete(volume, path, opts).await?;
Ok(DataDirDeleteStatus::Deleted)
@@ -1344,8 +1344,11 @@ async fn run_tier_config_reload_worker<F, Fut>(
}
TierConfigReloadFinish::Pending => retry_attempt = 0,
},
TierConfigReloadOutcome::Terminal(_) => match sys.finish_tier_config_reload_worker(&host) {
TierConfigReloadOutcome::Terminal(err) => match sys.finish_tier_config_reload_worker(&host) {
TierConfigReloadFinish::Completed => {
// This peer keeps the previous tier configuration for good, so record
// why. Dropping the error here hides the only evidence of a divergent
// node behind an outcome label that cannot be acted on.
warn!(
event = EVENT_NOTIFICATION_PEER_PROPAGATION,
component = LOG_COMPONENT_ECSTORE,
@@ -1353,6 +1356,7 @@ async fn run_tier_config_reload_worker<F, Fut>(
action = "reload_transition_tier_config",
host,
outcome = "terminal",
error = ?err,
"tier configuration reload stopped after a terminal outcome"
);
return;
+1 -1
View File
@@ -555,7 +555,7 @@ impl SetDisks {
}
}
fn file_info_quorum_hash(meta: &FileInfo) -> [u8; 32] {
pub(super) fn file_info_quorum_hash(meta: &FileInfo) -> [u8; 32] {
let mut hasher = Sha256::new();
Self::update_file_info_quorum_hash(&mut hasher, meta);
let digest = hasher.finalize();
+674 -160
View File
@@ -92,6 +92,69 @@ struct PartFailureSummary {
bitrot_failure: bool,
}
#[derive(Clone)]
struct RecoverableMetaCandidate {
identity: [u8; 32],
file_info: FileInfo,
data_count: usize,
local_payload: bool,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum DanglingDeleteSafety {
UnsafeToDelete,
NoRecoverableCandidate,
}
#[cfg(test)]
struct DanglingCheckPartsFailure {
key: DanglingCheckPartsFailureKey,
}
#[cfg(test)]
type DanglingCheckPartsFailureKey = (String, String, usize);
#[cfg(test)]
type DanglingCheckPartsFailures = HashMap<DanglingCheckPartsFailureKey, DiskError>;
#[cfg(test)]
fn dangling_check_parts_failures() -> &'static std::sync::Mutex<DanglingCheckPartsFailures> {
static FAILURES: std::sync::OnceLock<std::sync::Mutex<DanglingCheckPartsFailures>> = std::sync::OnceLock::new();
FAILURES.get_or_init(|| std::sync::Mutex::new(HashMap::new()))
}
#[cfg(test)]
impl DanglingCheckPartsFailure {
fn install(bucket: &str, object: &str, disk_index: usize, error: DiskError) -> Self {
let key = (bucket.to_string(), object.to_string(), disk_index);
let previous = dangling_check_parts_failures()
.lock()
.expect("dangling check-parts failure registry should not poison")
.insert(key.clone(), error);
assert!(previous.is_none(), "dangling check-parts failure already installed");
Self { key }
}
}
#[cfg(test)]
impl Drop for DanglingCheckPartsFailure {
fn drop(&mut self) {
dangling_check_parts_failures()
.lock()
.expect("dangling check-parts failure registry should not poison")
.remove(&self.key);
}
}
#[cfg(test)]
fn injected_dangling_check_parts_error(bucket: &str, object: &str, disk_index: usize) -> Option<DiskError> {
dangling_check_parts_failures()
.lock()
.expect("dangling check-parts failure registry should not poison")
.get(&(bucket.to_string(), object.to_string(), disk_index))
.cloned()
}
fn first_unhealthy_part_summary(
data_errs_by_part: &HashMap<usize, Vec<usize>>,
parts: &[ObjectPartInfo],
@@ -125,39 +188,17 @@ impl SetDisks {
version_id: &str,
opts: &HealOpts,
) -> disk::error::Result<(HealResultItem, Option<DiskError>)> {
// `allow_meta_regen` is true on the first pass: a version whose data shards
// physically survive (>= data_blocks) but whose xl.meta fell below
// read-quorum is RESCUED (missing xl.meta regenerated) rather than
// dangling-deleted. The re-drive after a rescue sets it false so the
// regeneration can happen at most once (no unbounded recursion).
Box::pin(self.heal_object_with_regen(bucket, object, version_id, opts, true)).await
}
/// Best-effort orphan-data-dir reclaim for an object that is healthy on this
/// set. Wraps [`Self::reclaim_orphan_data_dirs`] with the shared logging so
/// both `heal_object` exits — the already-healthy early return and the
/// post-heal tail — reclaim identically. Never fails the heal: delete errors
/// are logged and swallowed. Callers must gate this on `!opts.dry_run`.
async fn reclaim_orphan_data_dirs_best_effort(&self, bucket: &str, object: &str) {
match self.reclaim_orphan_data_dirs(bucket, object).await {
Ok(removed) if removed > 0 => {
info!(bucket, object, removed, "heal_object: reclaimed orphaned data directories");
}
Ok(_) => {}
Err(e) => {
warn!(bucket, object, error = %e, "heal_object: orphan data-dir reclaim failed");
}
}
Box::pin(self.heal_object_with_explicit_version_regen(bucket, object, version_id, opts, true)).await
}
#[allow(clippy::too_many_lines)]
async fn heal_object_with_regen(
async fn heal_object_with_explicit_version_regen(
&self,
bucket: &str,
object: &str,
version_id: &str,
opts: &HealOpts,
allow_meta_regen: bool,
allow_explicit_version_regen: bool,
) -> disk::error::Result<(HealResultItem, Option<DiskError>)> {
info!(?opts, "Starting heal_object");
@@ -354,22 +395,6 @@ impl SetDisks {
}
}
// DATA-SAFETY GUARD (backlog#920, decision 1): before any
// dangling delete, if the version's DATA shards physically
// survive on >= data_blocks disks it is RECONSTRUCTABLE.
// Regenerate the missing xl.meta from a surviving valid
// FileInfo and re-drive the heal instead of destroying a
// recoverable version. Torn writes (< data_blocks data
// shards) fall through to the existing dangling behavior.
if cannot_heal
&& allow_meta_regen
&& self
.try_regenerate_recoverable_meta(bucket, object, &parts_metadata, &errs, &disks)
.await?
{
return Box::pin(self.heal_object_with_regen(bucket, object, version_id, opts, false)).await;
}
if cannot_heal {
let total_disks = parts_metadata.len();
let healthy_count = total_disks.saturating_sub(disks_to_heal_count);
@@ -422,6 +447,20 @@ impl SetDisks {
);
}
// `disks_with_all_parts` normalizes conflicting entries
// in `parts_metadata` to defaults. Re-read only before
// destructive cleanup so the guard sees every original
// identity.
let (delete_guard_metadata, delete_guard_errs) =
Self::read_all_fileinfo(&disks, "", bucket, object, version_id, true, true, false).await?;
if self
.dangling_delete_safety(bucket, object, &delete_guard_metadata, &delete_guard_errs, &disks)
.await?
== DanglingDeleteSafety::UnsafeToDelete
{
return Ok((result, Some(cannot_heal_err)));
}
// Allow for dangling deletes, on versions that have DataDir missing etc.
// this would end up restoring the correct readable versions.
return match self
@@ -837,17 +876,25 @@ impl SetDisks {
}
}
Err(err) => {
// DATA-SAFETY GUARD (backlog#920, decision 1): meta quorum failed,
// but the version's DATA may still physically survive on enough
// disks (xl.meta lost on > parity disks while part files remain).
// Rescue it by regenerating the missing xl.meta and re-driving heal
// instead of dangling-deleting a reconstructable version.
if allow_meta_regen
if allow_explicit_version_regen
&& !version_id.is_empty()
&& self
.try_regenerate_recoverable_meta(bucket, object, &parts_metadata, &errs, &disks)
.try_regenerate_explicit_version_meta(bucket, object, version_id, &parts_metadata, &errs, &disks)
.await?
{
return Box::pin(self.heal_object_with_regen(bucket, object, version_id, opts, false)).await;
return Box::pin(self.heal_object_with_explicit_version_regen(bucket, object, version_id, opts, false)).await;
}
if self
.dangling_delete_safety(bucket, object, &parts_metadata, &errs, &disks)
.await?
== DanglingDeleteSafety::UnsafeToDelete
{
return Ok((
self.default_heal_result(FileInfo::default(), &errs, bucket, object, version_id)
.await,
Some(err),
));
}
let data_errs_by_part = HashMap::new();
@@ -883,129 +930,226 @@ impl SetDisks {
}
}
/// backlog#920 (decision 1): rescue a version that meta-quorum logic would
/// otherwise dangling-DELETE, when its DATA is still reconstructable.
///
/// Returns `Ok(true)` if the version was rescued (missing xl.meta regenerated
/// on at least one disk, so a re-driven heal can reconstruct it), `Ok(false)`
/// to fall through to the existing dangling-delete behavior.
///
/// Recoverability is computed by physically probing part files across ALL
/// disks in the set with `check_parts` — including disks whose xl.meta is
/// absent (a lost xl.meta does not lose the sibling `part.*` data). If at
/// least `data_blocks` disks hold every part of a surviving valid FileInfo,
/// the object is EC-reconstructable, so we regenerate that FileInfo's xl.meta
/// on every disk whose metadata is absent (via `write_metadata`, which merges
/// into any existing xl.meta). Delete markers, remote/transitioned versions,
/// and genuine torn writes (< `data_blocks` surviving data shards) are NOT
/// rescued — they keep the current dangling-delete-after-grace behavior, so no
/// regression on those paths.
async fn try_regenerate_recoverable_meta(
async fn try_regenerate_explicit_version_meta(
&self,
bucket: &str,
object: &str,
version_id: &str,
parts_metadata: &[FileInfo],
errs: &[Option<DiskError>],
disks: &[Option<DiskStore>],
) -> disk::error::Result<bool> {
let Ok(version_id) = Uuid::parse_str(version_id) else {
return Ok(false);
};
let candidates = parts_metadata
.iter()
.zip(errs.iter())
.filter_map(|(file_info, err)| {
(err.is_none()
&& file_info_is_valid_for_metadata(file_info)
&& file_info.version_id == Some(version_id)
&& file_info.has_valid_erasure_geometry()
&& !file_info.deleted
&& !file_info.is_remote()
&& file_info.data_dir.is_some()
&& !file_info.parts.is_empty()
&& file_info.erasure.data_blocks > 0
&& file_info
.erasure
.data_blocks
.checked_add(file_info.erasure.parity_blocks)
.is_some_and(|shards| shards == disks.len()))
.then_some(file_info)
})
.collect::<Vec<_>>();
let Some(candidate) = candidates.first().copied() else {
return Ok(false);
};
let identity = Self::file_info_quorum_hash(candidate);
if candidates
.iter()
.any(|file_info| Self::file_info_quorum_hash(file_info) != identity)
{
return Ok(false);
}
let mut available = 0usize;
for disk in disks {
let Some(disk) = disk else {
return Ok(false);
};
match disk.check_parts(bucket, object, candidate).await {
Ok(response)
if !response.results.is_empty() && response.results.iter().all(|result| *result == CHECK_PART_SUCCESS) =>
{
available += 1;
}
Ok(_)
| Err(
DiskError::FileNotFound
| DiskError::FileVersionNotFound
| DiskError::PathNotFound
| DiskError::VolumeNotFound,
) => {}
Err(_) => return Ok(false),
}
}
if available < candidate.erasure.data_blocks {
return Ok(false);
}
let mut wrote = 0usize;
for (index, disk) in disks.iter().enumerate() {
let Some(disk) = disk else {
return Ok(false);
};
let metadata_absent = matches!(
errs.get(index).and_then(Option::as_ref),
Some(DiskError::FileNotFound | DiskError::FileVersionNotFound)
);
if !metadata_absent {
continue;
}
let Some(&shard_index) = candidate.erasure.distribution.get(index) else {
return Ok(false);
};
let mut regenerated = candidate.clone();
regenerated.fresh = false;
regenerated.erasure.index = shard_index;
match disk.write_metadata("", bucket, object, regenerated).await {
Ok(()) => wrote += 1,
Err(error) => {
warn!(
bucket,
object,
disk_index = index,
error = %error,
"failed to regenerate recoverable xl.meta"
);
}
}
}
Ok(wrote > 0)
}
/// Best-effort orphan-data-dir reclaim for an object that is healthy on this
/// set. Wraps [`Self::reclaim_orphan_data_dirs`] with the shared logging so
/// both `heal_object` exits — the already-healthy early return and the
/// post-heal tail — reclaim identically. Never fails the heal: delete errors
/// are logged and swallowed. Callers must gate this on `!opts.dry_run`.
async fn reclaim_orphan_data_dirs_best_effort(&self, bucket: &str, object: &str) {
match self.reclaim_orphan_data_dirs(bucket, object).await {
Ok(removed) if removed > 0 => {
info!(bucket, object, removed, "heal_object: reclaimed orphaned data directories");
}
Ok(_) => {}
Err(e) => {
warn!(bucket, object, error = %e, "heal_object: orphan data-dir reclaim failed");
}
}
}
/// Prevent dangling cleanup when surviving state cannot prove that deletion
/// is safe. Part presence proves only recoverability, never commit: the write
/// path can durably rename data before xl.meta is committed.
async fn dangling_delete_safety(
&self,
bucket: &str,
object: &str,
parts_metadata: &[FileInfo],
errs: &[Option<DiskError>],
disks: &[Option<DiskStore>],
) -> disk::error::Result<bool> {
// A surviving valid, non-deleted, non-remote data FileInfo to rebuild from.
let Some(surviving) = parts_metadata
) -> disk::error::Result<DanglingDeleteSafety> {
if disks.iter().any(Option::is_none)
|| errs.iter().flatten().any(|err| {
!matches!(
err,
DiskError::FileNotFound
| DiskError::FileVersionNotFound
| DiskError::PathNotFound
| DiskError::VolumeNotFound
)
})
{
return Ok(DanglingDeleteSafety::UnsafeToDelete);
}
let mut candidates = Vec::<RecoverableMetaCandidate>::with_capacity(parts_metadata.len());
for (fi, err) in parts_metadata.iter().zip(errs.iter()) {
if err.is_some() || !file_info_is_valid_for_metadata(fi) {
continue;
}
let identity = Self::file_info_quorum_hash(fi);
if !candidates.iter().any(|candidate| candidate.identity == identity) {
let local_payload = fi.has_valid_erasure_geometry()
&& !fi.deleted
&& !fi.is_remote()
&& fi.data_dir.is_some()
&& !fi.parts.is_empty()
&& fi.erasure.data_blocks > 0
&& fi
.erasure
.data_blocks
.checked_add(fi.erasure.parity_blocks)
.is_some_and(|shards| shards == disks.len());
candidates.push(RecoverableMetaCandidate {
identity,
file_info: fi.clone(),
data_count: 0,
local_payload,
});
}
}
if candidates
.iter()
.find(|fi| fi.has_valid_erasure_geometry() && !fi.deleted && !fi.is_remote())
.cloned()
else {
return Ok(false);
};
// Without a data_dir + parts there is no data to prove recoverable.
if surviving.data_dir.is_none() || surviving.parts.is_empty() {
return Ok(false);
}
let data_blocks = surviving.erasure.data_blocks;
if data_blocks == 0 {
return Ok(false);
.any(|candidate| candidate.file_info.deleted || candidate.file_info.is_remote())
|| candidates.len() > 1
{
return Ok(DanglingDeleteSafety::UnsafeToDelete);
}
// Physically probe part presence on EVERY online disk using the surviving
// FileInfo's data_dir/parts. `check_parts` stats `object/<data_dir>/part.N`
// directly, so it counts disks that still hold the data even if their
// xl.meta was deleted.
let mut available = 0usize;
for disk in disks.iter().flatten() {
if let Ok(resp) = disk.check_parts(bucket, object, &surviving).await
&& !resp.results.is_empty()
&& resp.results.iter().all(|r| *r == CHECK_PART_SUCCESS)
{
available += 1;
}
}
for candidate in candidates.iter_mut().filter(|candidate| candidate.local_payload) {
for (disk_index, disk) in disks.iter().enumerate() {
let Some(disk) = disk else {
return Ok(DanglingDeleteSafety::UnsafeToDelete);
};
#[cfg(test)]
let check_result = match injected_dangling_check_parts_error(bucket, object, disk_index) {
Some(error) => Err(error),
None => disk.check_parts(bucket, object, &candidate.file_info).await,
};
#[cfg(not(test))]
let check_result = disk.check_parts(bucket, object, &candidate.file_info).await;
// Torn write: fewer than data_blocks surviving data shards is genuinely
// unrecoverable — preserve the current dangling behavior (no resurrection).
if available < data_blocks {
debug!(
bucket,
object,
available,
data_blocks,
"heal_object: version not reconstructable (torn write), keeping dangling behavior"
);
return Ok(false);
}
// Reconstructable: regenerate the surviving xl.meta on every disk whose
// metadata is absent so the version regains read-quorum. Each disk gets its
// OWN shard index: the disk at physical position `index` holds shard
// `distribution[index]` (mirrors `shuffle_disks` + the write path's
// `erasure.index = shuffled_pos + 1`). Copying the surviving disk's index
// verbatim would write an inconsistent xl.meta that the re-heal then treats
// as corrupt.
let distribution = &surviving.erasure.distribution;
let mut wrote = 0usize;
for (index, disk) in disks.iter().enumerate() {
let Some(disk) = disk else { continue };
let meta_absent = matches!(
errs.get(index).and_then(Option::as_ref),
Some(DiskError::FileNotFound | DiskError::FileVersionNotFound)
) || !parts_metadata.get(index).map(FileInfo::is_valid).unwrap_or(false);
if !meta_absent {
continue;
}
// Without a known shard index for this position we cannot write a
// consistent xl.meta; leave it for the normal heal to reconstruct.
let Some(&shard_index) = distribution.get(index) else {
continue;
};
let mut regen = surviving.clone();
regen.fresh = false; // merge into any existing xl.meta on the disk
regen.erasure.index = shard_index;
match disk.write_metadata("", bucket, object, regen).await {
Ok(()) => wrote += 1,
Err(e) => {
warn!(
bucket,
object,
disk_index = index,
error = %e,
"heal_object: failed to regenerate recoverable xl.meta on disk"
);
match check_result {
Ok(resp) if !resp.results.is_empty() && resp.results.iter().all(|result| *result == CHECK_PART_SUCCESS) => {
candidate.data_count += 1;
}
Ok(_) => {}
Err(
DiskError::FileNotFound
| DiskError::FileVersionNotFound
| DiskError::PathNotFound
| DiskError::VolumeNotFound,
) => {}
Err(_) => return Ok(DanglingDeleteSafety::UnsafeToDelete),
}
}
}
if wrote == 0 {
return Ok(false);
}
info!(
bucket,
object,
available,
data_blocks,
regenerated_meta_disks = wrote,
"heal_object: rescued reconstructable sub-quorum version by regenerating xl.meta"
);
Ok(true)
Ok(
if candidates
.iter()
.any(|candidate| candidate.local_payload && candidate.data_count >= candidate.file_info.erasure.data_blocks)
{
DanglingDeleteSafety::UnsafeToDelete
} else {
DanglingDeleteSafety::NoRecoverableCandidate
},
)
}
pub(in crate::set_disk) async fn heal_object_dir_locked(
@@ -1393,7 +1537,7 @@ impl crate::storage_api_contracts::heal::HealOperations for SetDisks {
#[cfg(test)]
mod heal_result_report_tests {
use super::SetDisks;
use super::{DanglingCheckPartsFailure, DanglingDeleteSafety, SetDisks};
use super::{HEAL_RENAME_INCOMPLETE, HealRenameFailureScope};
use crate::disk::endpoint::Endpoint;
use crate::disk::error::DiskError;
@@ -1405,10 +1549,12 @@ mod heal_result_report_tests {
use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _};
use crate::{config::storageclass, store::init_format::save_format_file};
use rustfs_common::heal_channel::{DriveState, HealOpts, HealScanMode};
use rustfs_filemeta::{BLOCK_SIZE_V2, FileInfo};
use rustfs_filemeta::{BLOCK_SIZE_V2, FileInfo, ObjectPartInfo, TRANSITION_COMPLETE};
use std::sync::Arc;
use tempfile::TempDir;
use time::OffsetDateTime;
use tokio::sync::RwLock;
use uuid::Uuid;
async fn real_disk() -> (TempDir, Endpoint, DiskStore) {
let dir = tempfile::tempdir().expect("tempdir should be created");
@@ -1446,6 +1592,68 @@ mod heal_result_report_tests {
.await
}
fn meta_regen_test_fileinfo(object: &str, data_dir: Uuid, mod_time: i64, disk_index: usize) -> FileInfo {
let mut fi = FileInfo::new(object, 2, 2);
fi.data_dir = Some(data_dir);
fi.mod_time = Some(OffsetDateTime::from_unix_timestamp(mod_time).expect("test timestamp should parse"));
fi.size = 1;
fi.parts = vec![ObjectPartInfo {
number: 1,
size: 1,
actual_size: 1,
..Default::default()
}];
fi.erasure.index = fi.erasure.distribution[disk_index];
fi
}
async fn meta_regen_test_set(
bucket: &str,
object: &str,
data_dirs: &[(Uuid, usize)],
) -> (Vec<TempDir>, Arc<SetDisks>, Vec<Option<DiskStore>>) {
let mut temp_dirs = Vec::new();
let mut endpoints = Vec::new();
let mut disks = Vec::new();
for disk_index in 0..4 {
let (temp_dir, endpoint, disk) = real_disk().await;
disk.make_volume(bucket).await.expect("test bucket should be created");
for (data_dir, shard_count) in data_dirs {
if disk_index >= *shard_count {
continue;
}
let part_dir = temp_dir.path().join(bucket).join(object).join(data_dir.to_string());
tokio::fs::create_dir_all(&part_dir)
.await
.expect("test data directory should be created");
tokio::fs::write(part_dir.join("part.1"), [1u8; 2])
.await
.expect("test data shard should be written");
}
temp_dirs.push(temp_dir);
endpoints.push(endpoint);
disks.push(Some(disk));
}
let set = set_disks_with(disks.clone(), endpoints, 2).await;
(temp_dirs, set, disks)
}
async fn seed_meta_regen_test_metadata(
disks: &[Option<DiskStore>],
disk_index: usize,
bucket: &str,
object: &str,
file_info: &FileInfo,
) {
disks[disk_index]
.as_ref()
.expect("metadata test disk should be online")
.write_metadata("", bucket, object, file_info.clone())
.await
.expect("test metadata should be written");
}
async fn formatted_single_disk_no_parity_set() -> (TempDir, Arc<SetDisks>) {
let format = FormatV3::new(1, 1);
let dir = tempfile::tempdir().expect("tempdir should be created");
@@ -1753,6 +1961,312 @@ mod heal_result_report_tests {
assert_eq!(result.before.drives[3].state, DriveState::Ok.to_string());
}
#[tokio::test]
async fn dangling_delete_guard_preserves_conflicting_identities_without_writing_metadata() {
let bucket = "bucket-delete-guard-conflict";
let object = "object.bin";
let old_data_dir = Uuid::parse_str("99999999-9999-9999-9999-999999999999").expect("old data dir should parse");
let new_data_dir = Uuid::parse_str("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa").expect("new data dir should parse");
let (_temp_dirs, set, disks) = meta_regen_test_set(bucket, object, &[(old_data_dir, 4), (new_data_dir, 2)]).await;
let version_id = Uuid::parse_str("bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb").expect("version id should parse");
let mut metadata = vec![
meta_regen_test_fileinfo(object, old_data_dir, 9, 0),
meta_regen_test_fileinfo(object, new_data_dir, 10, 1),
FileInfo::default(),
FileInfo::default(),
];
metadata[0].version_id = Some(version_id);
metadata[1].version_id = Some(version_id);
assert_eq!(
metadata[0].version_id, metadata[1].version_id,
"the conflicting candidates must share one version id"
);
seed_meta_regen_test_metadata(&disks, 0, bucket, object, &metadata[0]).await;
seed_meta_regen_test_metadata(&disks, 1, bucket, object, &metadata[1]).await;
let errs = vec![None, None, Some(DiskError::FileNotFound), Some(DiskError::FileNotFound)];
assert!(
set.dangling_delete_safety(bucket, object, &metadata, &errs, &disks)
.await
.expect("conflicting identities should be classified")
== DanglingDeleteSafety::UnsafeToDelete
);
let reversed = vec![
metadata[1].clone(),
metadata[0].clone(),
FileInfo::default(),
FileInfo::default(),
];
assert!(
set.dangling_delete_safety(bucket, object, &reversed, &errs, &disks)
.await
.expect("reversed identities should be classified")
== DanglingDeleteSafety::UnsafeToDelete
);
let version_id = version_id.to_string();
assert!(
!set.try_regenerate_explicit_version_meta(bucket, object, &version_id, &metadata, &errs, &disks)
.await
.expect("conflicting explicit-version candidates should be rejected"),
"an explicit version must not select between conflicting metadata identities"
);
for disk_index in [2, 3] {
assert!(
matches!(
disks[disk_index]
.as_ref()
.expect("test disk should be online")
.read_version("", bucket, object, "", &ReadOptions::default())
.await,
Err(DiskError::FileNotFound)
),
"the delete guard must not manufacture metadata on missing disks"
);
}
let old = disks[0]
.as_ref()
.expect("first test disk should be online")
.read_version("", bucket, object, "", &ReadOptions::default())
.await
.expect("old metadata should remain readable");
let new = disks[1]
.as_ref()
.expect("second test disk should be online")
.read_version("", bucket, object, "", &ReadOptions::default())
.await
.expect("new metadata should remain readable");
assert_eq!(old.data_dir, Some(old_data_dir));
assert_eq!(new.data_dir, Some(new_data_dir));
}
#[tokio::test]
async fn heal_meta_quorum_failure_preserves_reconstructable_uncommitted_candidate() {
let bucket = "bucket-delete-guard-reconstructable";
let object = "object.bin";
let data_dir = Uuid::parse_str("33333333-3333-3333-3333-333333333333").expect("data dir should parse");
let (_temp_dirs, set, disks) = meta_regen_test_set(bucket, object, &[(data_dir, 2)]).await;
let metadata = [
meta_regen_test_fileinfo(object, data_dir, 3, 0),
FileInfo::default(),
FileInfo::default(),
FileInfo::default(),
];
seed_meta_regen_test_metadata(&disks, 0, bucket, object, &metadata[0]).await;
let (observed_metadata, observed_errs) = SetDisks::read_all_fileinfo(&disks, "", bucket, object, "", true, true, false)
.await
.expect("test metadata should be readable across the set");
assert_eq!(
set.dangling_delete_safety(bucket, object, &observed_metadata, &observed_errs, &disks)
.await
.expect("observed reconstructable candidate should be classified"),
DanglingDeleteSafety::UnsafeToDelete
);
let (_, err) = set
.heal_object(
bucket,
object,
"",
&HealOpts {
no_lock: true,
..Default::default()
},
)
.await
.expect("unsafe dangling state should be reported without deletion");
assert_eq!(err, Some(DiskError::FileNotFound));
let surviving = disks[0]
.as_ref()
.expect("first test disk should be online")
.read_version("", bucket, object, "", &ReadOptions::default())
.await
.expect("the only metadata copy must be preserved");
assert_eq!(surviving.data_dir, Some(data_dir));
assert!(
matches!(
disks[1]
.as_ref()
.expect("second test disk should be online")
.read_version("", bucket, object, "", &ReadOptions::default())
.await,
Err(DiskError::FileNotFound)
),
"the delete guard must not propagate metadata"
);
}
#[tokio::test]
async fn heal_meta_quorum_failure_preserves_candidate_when_required_shard_disk_is_offline() {
let bucket = "bucket-delete-guard-offline";
let object = "object.bin";
let data_dir = Uuid::parse_str("44444444-4444-4444-4444-444444444444").expect("data dir should parse");
let (temp_dirs, set, disks) = meta_regen_test_set(bucket, object, &[(data_dir, 2)]).await;
let metadata = meta_regen_test_fileinfo(object, data_dir, 4, 0);
seed_meta_regen_test_metadata(&disks, 0, bucket, object, &metadata).await;
set.disks.write().await[1] = None;
let (_, err) = set
.heal_object(
bucket,
object,
"",
&HealOpts {
no_lock: true,
..Default::default()
},
)
.await
.expect("offline shard state should be reported without deletion");
assert_eq!(err, Some(DiskError::FileNotFound));
let surviving = disks[0]
.as_ref()
.expect("first test disk should be online")
.read_version("", bucket, object, "", &ReadOptions::default())
.await
.expect("offline uncertainty must preserve the surviving metadata");
assert_eq!(surviving.data_dir, Some(data_dir));
assert!(
temp_dirs[0]
.path()
.join(bucket)
.join(object)
.join(data_dir.to_string())
.join("part.1")
.is_file(),
"offline uncertainty must preserve the last online shard"
);
}
#[tokio::test]
async fn heal_meta_quorum_failure_preserves_candidate_when_part_probe_times_out() {
let bucket = "bucket-delete-guard-timeout";
let object = "object.bin";
let data_dir = Uuid::parse_str("55555555-5555-5555-5555-555555555555").expect("data dir should parse");
let (temp_dirs, set, disks) = meta_regen_test_set(bucket, object, &[(data_dir, 2)]).await;
let metadata = meta_regen_test_fileinfo(object, data_dir, 5, 0);
seed_meta_regen_test_metadata(&disks, 0, bucket, object, &metadata).await;
let _failure = DanglingCheckPartsFailure::install(bucket, object, 1, DiskError::Timeout);
let (_, err) = set
.heal_object(
bucket,
object,
"",
&HealOpts {
no_lock: true,
..Default::default()
},
)
.await
.expect("part probe timeout should be reported without deletion");
assert_eq!(err, Some(DiskError::FileNotFound));
let surviving = disks[0]
.as_ref()
.expect("first test disk should be online")
.read_version("", bucket, object, "", &ReadOptions::default())
.await
.expect("probe uncertainty must preserve the surviving metadata");
assert_eq!(surviving.data_dir, Some(data_dir));
assert!(
temp_dirs[0]
.path()
.join(bucket)
.join(object)
.join(data_dir.to_string())
.join("part.1")
.is_file(),
"probe uncertainty must preserve the last confirmed shard"
);
}
#[tokio::test]
async fn dangling_delete_guard_ignores_set_incompatible_geometry() {
let bucket = "bucket-delete-guard-short-geometry";
let object = "object.bin";
let data_dir = Uuid::parse_str("abababab-abab-abab-abab-abababababab").expect("data dir should parse");
let (_temp_dirs, set, disks) = meta_regen_test_set(bucket, object, &[(data_dir, 1)]).await;
let mut candidate = FileInfo::new(object, 1, 0);
candidate.data_dir = Some(data_dir);
candidate.mod_time = Some(OffsetDateTime::from_unix_timestamp(18).expect("timestamp should parse"));
candidate.size = 1;
candidate.parts = vec![ObjectPartInfo {
number: 1,
size: 1,
actual_size: 1,
..Default::default()
}];
candidate.erasure.index = candidate.erasure.distribution[0];
seed_meta_regen_test_metadata(&disks, 0, bucket, object, &candidate).await;
let metadata = vec![candidate, FileInfo::default(), FileInfo::default(), FileInfo::default()];
let errs = vec![
None,
Some(DiskError::FileNotFound),
Some(DiskError::FileNotFound),
Some(DiskError::FileNotFound),
];
assert!(
set.dangling_delete_safety(bucket, object, &metadata, &errs, &disks)
.await
.expect("set-incompatible geometry should be classified")
== DanglingDeleteSafety::NoRecoverableCandidate
);
}
#[tokio::test]
async fn dangling_delete_guard_preserves_delete_marker_and_remote_metadata() {
let bucket = "bucket-delete-guard-nonlocal";
let object = "object.bin";
let (_temp_dirs, set, disks) = meta_regen_test_set(bucket, object, &[]).await;
let marker = FileInfo {
name: object.to_string(),
version_id: Some(Uuid::parse_str("eeeeeeee-eeee-eeee-eeee-eeeeeeeeeeee").expect("version id should parse")),
deleted: true,
mod_time: Some(OffsetDateTime::from_unix_timestamp(14).expect("marker timestamp should parse")),
..Default::default()
};
let remote_dir = Uuid::parse_str("89898989-8989-8989-8989-898989898989").expect("remote data dir should parse");
let mut remote = meta_regen_test_fileinfo(object, remote_dir, 15, 1);
remote.transition_status = TRANSITION_COMPLETE.to_string();
remote.transition_tier = "WARM".to_string();
remote.transitioned_objname = "remote/object.bin".to_string();
for metadata in [marker, remote] {
let candidates = vec![metadata, FileInfo::default(), FileInfo::default(), FileInfo::default()];
let errs = vec![
None,
Some(DiskError::FileNotFound),
Some(DiskError::FileNotFound),
Some(DiskError::FileNotFound),
];
assert_eq!(
set.dangling_delete_safety(bucket, object, &candidates, &errs, &disks)
.await
.expect("non-local metadata should be classified"),
DanglingDeleteSafety::UnsafeToDelete
);
}
}
#[tokio::test]
async fn dangling_delete_guard_preserves_metadata_read_uncertainty() {
let bucket = "bucket-delete-guard-read-error";
let object = "object.bin";
let (_temp_dirs, set, disks) = meta_regen_test_set(bucket, object, &[]).await;
let metadata = vec![FileInfo::default(); disks.len()];
for read_error in [DiskError::Timeout, DiskError::DiskAccessDenied, DiskError::DiskNotFound] {
let mut errs = vec![Some(DiskError::FileNotFound); disks.len()];
errs[0] = Some(read_error);
assert_eq!(
set.dangling_delete_safety(bucket, object, &metadata, &errs, &disks)
.await
.expect("metadata read uncertainty should be classified"),
DanglingDeleteSafety::UnsafeToDelete
);
}
}
#[tokio::test]
async fn heal_no_parity_bitrot_reports_unrecoverable_integrity_failure() {
let (dir, set) = formatted_single_disk_no_parity_set().await;
+46 -13
View File
@@ -27,7 +27,7 @@ use crate::multipart_listing::paginate_multipart_listing;
use futures::{StreamExt, stream};
use std::future::Future;
#[cfg(test)]
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
use tokio::task::JoinSet;
@@ -36,6 +36,7 @@ const MULTIPART_LIST_IO_CONCURRENCY: usize = 16;
#[cfg(test)]
#[derive(Clone, Copy, PartialEq, Eq)]
pub(crate) enum MultipartCommitPause {
PutPartBeforeLockAcquire,
PutPartBeforeLockLost,
PutPartAfterRename,
BeforeLockLost,
@@ -47,9 +48,10 @@ struct MultipartCommitBarrierState {
bucket: String,
object: String,
pause: MultipartCommitPause,
armed: AtomicBool,
expected_arrivals: usize,
arrivals: AtomicUsize,
arrived: tokio::sync::Notify,
release: tokio::sync::Notify,
release: tokio::sync::Semaphore,
}
#[cfg(test)]
@@ -64,13 +66,24 @@ static MULTIPART_COMMIT_BARRIER: std::sync::OnceLock<std::sync::Mutex<Option<Arc
#[cfg(test)]
impl MultipartCommitBarrier {
pub(crate) fn install(bucket: &str, object: &str, pause: MultipartCommitPause) -> Self {
Self::install_for_arrivals(bucket, object, pause, 1)
}
pub(crate) fn install_for_arrivals(
bucket: &str,
object: &str,
pause: MultipartCommitPause,
expected_arrivals: usize,
) -> Self {
assert!(expected_arrivals > 0, "multipart commit barrier must wait for at least one arrival");
let state = Arc::new(MultipartCommitBarrierState {
bucket: bucket.to_string(),
object: object.to_string(),
pause,
armed: AtomicBool::new(true),
expected_arrivals,
arrivals: AtomicUsize::new(0),
arrived: tokio::sync::Notify::new(),
release: tokio::sync::Notify::new(),
release: tokio::sync::Semaphore::new(0),
});
let mut slot = MULTIPART_COMMIT_BARRIER
.get_or_init(|| std::sync::Mutex::new(None))
@@ -83,20 +96,28 @@ impl MultipartCommitBarrier {
}
pub(crate) async fn wait_until_paused(&self) {
tokio::time::timeout(Duration::from_secs(30), self.state.arrived.notified())
.await
.expect("multipart completion should reach the deterministic commit barrier");
tokio::time::timeout(Duration::from_secs(30), async {
loop {
let arrived = self.state.arrived.notified();
if self.state.arrivals.load(Ordering::Acquire) >= self.state.expected_arrivals {
return;
}
arrived.await;
}
})
.await
.expect("multipart completion should reach the deterministic commit barrier");
}
pub(crate) fn release(&self) {
self.state.release.notify_one();
self.state.release.add_permits(self.state.expected_arrivals);
}
}
#[cfg(test)]
impl Drop for MultipartCommitBarrier {
fn drop(&mut self) {
self.state.release.notify_one();
self.release();
let mut slot = MULTIPART_COMMIT_BARRIER
.get_or_init(|| std::sync::Mutex::new(None))
.lock()
@@ -117,10 +138,20 @@ async fn pause_multipart_commit(bucket: &str, object: &str, pause: MultipartComm
.filter(|barrier| barrier.bucket == bucket && barrier.object == object && barrier.pause == pause)
.cloned();
if let Some(barrier) = barrier
&& barrier.armed.swap(false, Ordering::AcqRel)
&& let Ok(previous) = barrier.arrivals.fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| {
(current < barrier.expected_arrivals).then_some(current + 1)
})
{
barrier.arrived.notify_one();
barrier.release.notified().await;
let arrival = previous + 1;
if arrival == barrier.expected_arrivals {
barrier.arrived.notify_one();
}
barrier
.release
.acquire()
.await
.expect("multipart commit barrier should remain open")
.forget();
}
}
@@ -693,6 +724,8 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
let part_path = format!("{}/{}/{}", upload_id_path, fi.data_dir.unwrap_or_default(), part_suffix);
#[cfg(test)]
pause_multipart_commit(bucket, object, MultipartCommitPause::PutPartBeforeLockAcquire).await;
// Serialize only the commit (rename_part), not the whole upload. Each
// concurrent stream writes to its own unique temp dir (see `tmp_part`
// above), so the encode/stream phase never conflicts and must stay
+79 -11
View File
@@ -2315,23 +2315,50 @@ async fn pause_transition_commit(bucket: &str, object: &str, pause: TransitionCo
fn persisted_transition_version(
remote_version: &str,
) -> std::io::Result<(Option<String>, rustfs_filemeta::TransitionVersionState)> {
persisted_transition_version_with_gate(remote_version, remote_version_state_writer_enabled())
}
fn remote_version_state_writer_enabled() -> bool {
remote_version_state_writer_enabled_for(
rustfs_utils::get_env_bool(
rustfs_config::ENV_TIER_REMOTE_VERSION_STATE_WRITE,
rustfs_config::DEFAULT_TIER_REMOTE_VERSION_STATE_WRITE,
),
rustfs_utils::get_env_bool(
rustfs_config::ENV_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED,
rustfs_config::DEFAULT_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED,
),
)
}
fn remote_version_state_writer_enabled_for(requested: bool, fleet_confirmed: bool) -> bool {
requested && fleet_confirmed
}
fn persisted_transition_version_with_gate(
remote_version: &str,
remote_version_state_writer_enabled: bool,
) -> std::io::Result<(Option<String>, rustfs_filemeta::TransitionVersionState)> {
if remote_version.is_empty() {
return Ok((None, rustfs_filemeta::TransitionVersionState::KnownDisabled));
}
let version_id = Uuid::parse_str(remote_version).map_err(|_| {
std::io::Error::new(
std::io::ErrorKind::Unsupported,
"opaque remote tier versions require the cluster capability gate",
)
})?;
if version_id.is_nil() {
return Err(std::io::Error::new(
match Uuid::parse_str(remote_version) {
Ok(version_id) if version_id.is_nil() => Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
"remote tier returned a nil object version ID",
));
)),
Ok(_) => Ok((Some(remote_version.to_string()), rustfs_filemeta::TransitionVersionState::Exact)),
Err(_) if !remote_version_state_writer_enabled => Err(std::io::Error::new(
std::io::ErrorKind::Unsupported,
"opaque remote tier versions require the operator-attested fleet gate",
)),
Err(_) if remote_version == "null" => {
Ok((Some(remote_version.to_string()), rustfs_filemeta::TransitionVersionState::SuspendedNull))
}
Err(_) => Ok((Some(remote_version.to_string()), rustfs_filemeta::TransitionVersionState::Exact)),
}
Ok((Some(remote_version.to_string()), rustfs_filemeta::TransitionVersionState::Exact))
}
#[cfg(test)]
@@ -2569,7 +2596,10 @@ mod transition_upload_completion_tests {
#[cfg(test)]
mod transition_version_id_tests {
use super::{TransitionUploadCandidate, persisted_transition_version};
use super::{
TransitionUploadCandidate, persisted_transition_version, persisted_transition_version_with_gate,
remote_version_state_writer_enabled_for,
};
use rustfs_filemeta::TransitionVersionState;
use uuid::Uuid;
@@ -2608,6 +2638,44 @@ mod transition_version_id_tests {
"opaque-version-token"
);
}
#[test]
fn remote_version_state_writer_requires_request_and_fleet_confirmation() {
for (case, requested, fleet_confirmed, expected) in [
("old defaults", false, false, false),
("missing fleet confirmation", true, false, false),
("missing local opt-in", false, true, false),
("explicitly unconfirmed fleet", true, false, false),
("rolled-back writer", false, true, false),
("fully upgraded fleet", true, true, true),
] {
assert_eq!(remote_version_state_writer_enabled_for(requested, fleet_confirmed), expected, "{case}");
}
}
#[test]
fn fleet_gate_enables_null_and_opaque_remote_version_states() {
for (remote_version, expected) in [
("null", (Some("null".to_string()), TransitionVersionState::SuspendedNull)),
(
"opaque-version-token",
(Some("opaque-version-token".to_string()), TransitionVersionState::Exact),
),
] {
assert!(
persisted_transition_version_with_gate(remote_version, false).is_err(),
"missing fleet confirmation must reject {remote_version:?}"
);
assert_eq!(
persisted_transition_version_with_gate(remote_version, true).expect("fleet-confirmed state must be persisted"),
expected
);
}
assert_eq!(
persisted_transition_version_with_gate("", true).expect("empty remote version identifies an unversioned tier"),
(None, TransitionVersionState::KnownDisabled)
);
}
}
impl SetDisks {
+1 -1
View File
@@ -87,7 +87,7 @@ fn bucket_deleted_marker_volume(bucket: &str) -> String {
format!("{RUSTFS_META_BUCKET}/{}", bucket_deleted_marker_prefix(bucket))
}
async fn await_bucket_namespace_operation<T, F>(
pub(crate) async fn await_bucket_namespace_operation<T, F>(
guard: Option<&rustfs_lock::NamespaceLockGuard>,
bucket: &str,
operation: &'static str,
+1
View File
@@ -141,6 +141,7 @@ fn should_enqueue_transition_immediately(oi: &ObjectInfo) -> bool {
const MAX_UPLOADS_LIST: usize = 10000;
mod bucket;
pub(crate) use bucket::await_bucket_namespace_operation;
mod heal;
mod heal_walk;
pub use heal_walk::HealWalkVersion;
+3 -1
View File
@@ -390,8 +390,10 @@ mod serial_tests {
// ─── 1️⃣ delete format.json on one disk ──────────────
let format_path = disk_paths[0].join(".rustfs.sys").join("format.json");
std::fs::remove_dir_all(&disk_paths[0]).expect("failed to delete all contents under disk_paths[0]");
let failed_disk_path = disk_paths[0].with_extension("failed");
std::fs::rename(&disk_paths[0], &failed_disk_path).expect("failed to detach disk_paths[0]");
std::fs::create_dir_all(&disk_paths[0]).expect("failed to recreate disk_paths[0] directory");
assert!(!format_path.exists(), "replacement disk already contains format.json before heal");
println!("✅ Deleted format.json on disk: {:?}", disk_paths[0]);
let (_format_result, format_error) = heal_storage.heal_format(false).await.expect("failed to run heal_format");
+24 -34
View File
@@ -14,11 +14,11 @@
//! Swift account operations and validation
use super::metadata_update::{ACCOUNT_META_TAG_PREFIX, MetadataUpdate};
use super::storage_api::account::{BucketOperations, MakeBucketOptions};
use super::{SwiftError, SwiftResult};
use super::{get_swift_bucket_metadata, resolve_swift_object_store_handle, update_swift_bucket_tagging, validate_metadata};
use super::{get_swift_bucket_metadata, resolve_swift_object_store_handle, update_swift_bucket_tagging};
use rustfs_credentials::Credentials;
use s3s::dto::{Tag, Tagging};
use sha2::{Digest, Sha256};
use std::collections::HashMap;
@@ -131,9 +131,8 @@ pub async fn get_account_metadata(account: &str, _credentials: &Option<Credentia
if let Some(tagging) = &bucket_meta.tagging_config {
for tag in &tagging.tag_set {
if let (Some(key), Some(value)) = (&tag.key, &tag.value)
&& let Some(meta_key) = key.strip_prefix("swift-account-meta-")
&& let Some(meta_key) = key.strip_prefix(ACCOUNT_META_TAG_PREFIX)
{
// Strip "swift-account-meta-" prefix
metadata.insert(meta_key.to_string(), value.clone());
}
}
@@ -147,6 +146,12 @@ pub async fn get_account_metadata(account: &str, _credentials: &Option<Credentia
/// Updates account-level metadata such as TempURL keys.
/// Only updates swift-account-meta-* tags, preserving other tags.
///
/// Swift account POST is additive: `update` names the items to write and the
/// items to drop, and everything else keeps its stored value. Replacing the
/// whole set instead would make an unrelated POST — setting a quota, say —
/// delete the account's TempURL signing key, permanently invalidating every
/// outstanding TempURL and FormPost signature for the account.
///
/// The caller must own the account: this metadata holds the account's TempURL
/// signing key, so writing it for someone else's account would let the writer
/// mint valid pre-signed URLs against that account's objects. Reads
@@ -155,11 +160,11 @@ pub async fn get_account_metadata(account: &str, _credentials: &Option<Credentia
///
/// # Arguments
/// * `account` - Account identifier
/// * `metadata` - Metadata key-value pairs to store (keys will be prefixed with `swift-account-meta-`)
/// * `update` - Metadata items to set and to remove (names are prefixed with `swift-account-meta-`)
/// * `credentials` - Keystone credentials of the caller
pub async fn update_account_metadata(
account: &str,
metadata: &HashMap<String, String>,
update: &MetadataUpdate,
credentials: &Option<Credentials>,
) -> SwiftResult<()> {
let Some(credentials) = credentials.as_ref() else {
@@ -171,8 +176,15 @@ pub async fn update_account_metadata(
// These tags are persisted into the bucket metadata file, which every
// later config write rewrites in full — so unbounded metadata inflates
// the cost of unrelated writes for the life of the account.
validate_metadata(metadata)?;
// the cost of unrelated writes for the life of the account. The item
// count is capped against the merged result, inside the rewrite.
update.validate()?;
// An update that names no item changes nothing, so there is no reason to
// bring the account's metadata bucket into existence for it.
if update.is_empty() {
return Ok(());
}
let bucket_name = get_account_metadata_bucket_name(account);
@@ -190,32 +202,10 @@ pub async fn update_account_metadata(
.map_err(|e| SwiftError::InternalServerError(format!("Failed to create account metadata bucket: {}", e)))?;
}
// Rewrite the persisted tags: replace swift-account-meta-* tags with the
// new metadata while preserving other tags. An empty result clears the
// tagging config.
update_swift_bucket_tagging(bucket_name, |current| {
let mut tagging = current.cloned().unwrap_or_else(|| Tagging { tag_set: vec![] });
tagging.tag_set.retain(|tag| {
if let Some(key) = &tag.key {
!key.starts_with("swift-account-meta-")
} else {
true
}
});
for (key, value) in metadata {
tagging.tag_set.push(Tag {
key: Some(format!("swift-account-meta-{}", key)),
value: Some(value.clone()),
});
}
tagging
})
.await?;
Ok(())
// Merge into the persisted tags: only the swift-account-meta-* items this
// update names change, and non-Swift tags are left alone. An empty result
// clears the tagging config.
update_swift_bucket_tagging(bucket_name, |current| update.apply_to_tags(current, ACCOUNT_META_TAG_PREFIX)).await
}
/// Get TempURL key for account
+97 -235
View File
@@ -17,15 +17,13 @@
//! This module implements Swift container CRUD operations and container-bucket translation.
use super::account::validate_account_access;
use super::metadata_update::{CONTAINER_META_TAG_PREFIX, MetadataUpdate};
use super::storage_api::container::{
BucketInfo, BucketOperations, BucketOptions, DeleteBucketOptions, ListOperations as _, MakeBucketOptions,
};
use super::types::Container;
use super::{SwiftError, SwiftResult};
use super::{
get_swift_bucket_metadata, get_swift_bucket_usage, resolve_swift_object_store_handle, update_swift_bucket_tagging,
validate_metadata,
};
use super::{get_swift_bucket_metadata, get_swift_bucket_usage, resolve_swift_object_store_handle, update_swift_bucket_tagging};
use rustfs_credentials::Credentials;
use s3s::dto::{Tag, Tagging};
use sha2::{Digest, Sha256};
@@ -62,30 +60,6 @@ fn sanitize_storage_error<E: std::fmt::Display>(operation: &str, error: E) -> Sw
SwiftError::InternalServerError(format!("{} operation failed", operation))
}
/// Convert Swift container metadata to S3 tags
///
/// Swift container metadata uses X-Container-Meta-* headers.
/// We store these as S3 tags with "swift-meta-" prefix to distinguish from regular bucket tags.
///
/// Example: X-Container-Meta-Color: Blue → S3 Tag: swift-meta-color=Blue
fn swift_metadata_to_s3_tags(metadata: &std::collections::HashMap<String, String>) -> Option<Tagging> {
let mut tags = Vec::new();
for (key, value) in metadata {
// Store with "swift-meta-" prefix to namespace container metadata
tags.push(Tag {
key: Some(format!("swift-meta-{}", key.to_lowercase())),
value: Some(value.clone()),
});
}
if tags.is_empty() {
None
} else {
Some(Tagging { tag_set: tags })
}
}
/// Convert S3 tags back to Swift container metadata
///
/// Extracts only tags with "swift-meta-" prefix, which represent Swift container metadata.
@@ -94,11 +68,9 @@ fn s3_tags_to_swift_metadata(tagging: &Tagging) -> std::collections::HashMap<Str
let mut metadata = std::collections::HashMap::new();
for tag in &tagging.tag_set {
// Only process tags with "swift-meta-" prefix
if let (Some(key), Some(value)) = (&tag.key, &tag.value)
&& let Some(meta_key) = key.strip_prefix("swift-meta-")
&& let Some(meta_key) = key.strip_prefix(CONTAINER_META_TAG_PREFIX)
{
// Skip "swift-meta-"
metadata.insert(meta_key.to_string(), value.clone());
}
}
@@ -473,12 +445,15 @@ pub async fn get_container_metadata(account: &str, container: &str, credentials:
/// - Returns 204 No Content on success
/// - Returns 404 Not Found if container doesn't exist
/// - Metadata is provided via X-Container-Meta-* headers
/// - The update is additive: items the request does not name keep their stored
/// value, and removal is explicit, via `X-Remove-Container-Meta-{name}` or an
/// empty value
#[allow(dead_code)] // Used by handler
pub async fn update_container_metadata(
account: &str,
container: &str,
credentials: &Credentials,
metadata: std::collections::HashMap<String, String>,
update: MetadataUpdate,
) -> SwiftResult<()> {
// Validate account access and extract project_id
let project_id = validate_account_access(account, credentials)?;
@@ -488,8 +463,9 @@ pub async fn update_container_metadata(
// These tags are persisted into the bucket metadata file, which every
// later config write rewrites in full — so unbounded metadata inflates
// the cost of unrelated writes for the life of the container.
validate_metadata(&metadata)?;
// the cost of unrelated writes for the life of the container. The item
// count is capped against the merged result, inside the rewrite.
update.validate()?;
// Create mapper with default config (tenant prefixing enabled)
let mapper = ContainerMapper::default();
@@ -502,7 +478,9 @@ pub async fn update_container_metadata(
return Err(SwiftError::InternalServerError("Storage layer not initialized".to_string()));
};
// Verify container exists
// Verify container exists. Checked before the empty-update shortcut below,
// so a POST to a container that does not exist still answers 404 whether
// or not it carried metadata.
store
.get_bucket_info(&bucket_name, &BucketOptions::default())
.await
@@ -514,32 +492,19 @@ pub async fn update_container_metadata(
}
})?;
// Rewrite the persisted tags: replace swift-meta-* tags with the new
// metadata while preserving non-Swift tags. An empty result clears the
// tagging config.
update_swift_bucket_tagging(bucket_name, |current| {
let mut tagging = current.cloned().unwrap_or_else(|| Tagging { tag_set: vec![] });
// An update that names no item leaves stored metadata alone, so skip the
// persisted write and the peer reload it triggers rather than rewriting
// the config to its current value. The handler reaches here on every
// container POST, including ACL-only and versioning-only ones.
if update.is_empty() {
return Ok(());
}
tagging.tag_set.retain(|tag| {
if let Some(key) = &tag.key {
!key.starts_with("swift-meta-")
} else {
true // Keep tags with no key (shouldn't happen, but be safe)
}
});
if let Some(mut new_tagging) = swift_metadata_to_s3_tags(&metadata) {
tagging.tag_set.append(&mut new_tagging.tag_set);
}
// If metadata.is_empty() and swift_metadata_to_s3_tags returns None,
// we've already removed swift-meta-* tags above, so only non-Swift
// tags remain
tagging
})
.await?;
Ok(())
// Merge into the persisted tags: only the swift-meta-* items this update
// names change, so the container's other metadata — and the ACL and
// versioning tags sharing this tag set — survive. An empty result clears
// the tagging config.
update_swift_bucket_tagging(bucket_name, |current| update.apply_to_tags(current, CONTAINER_META_TAG_PREFIX)).await
}
/// Delete a container
@@ -814,7 +779,7 @@ pub async fn enable_versioning(
value: Some(archive_container.to_string()), // Store Swift container name, not S3 bucket name
});
tagging
Ok(tagging)
})
.await?;
@@ -871,7 +836,7 @@ pub async fn disable_versioning(account: &str, container: &str, credentials: &Cr
.tag_set
.retain(|tag| tag.key.as_deref() != Some("swift-versions-location"));
tagging
Ok(tagging)
})
.await?;
@@ -1026,7 +991,7 @@ pub async fn set_container_acl(
});
}
tagging
Ok(tagging)
})
.await?;
@@ -1374,64 +1339,6 @@ mod tests {
assert!(first_char.is_ascii_lowercase() || first_char.is_ascii_digit());
}
#[test]
fn test_swift_metadata_to_s3_tags() {
let mut metadata = std::collections::HashMap::new();
metadata.insert("color".to_string(), "blue".to_string());
metadata.insert("description".to_string(), "test container".to_string());
let tagging = swift_metadata_to_s3_tags(&metadata).unwrap();
assert_eq!(tagging.tag_set.len(), 2);
// Verify tags have swift-meta- prefix
let color_tag = tagging
.tag_set
.iter()
.find(|t| t.key.as_deref() == Some("swift-meta-color"))
.expect("color tag not found");
assert_eq!(color_tag.value.as_deref(), Some("blue"));
let desc_tag = tagging
.tag_set
.iter()
.find(|t| t.key.as_deref() == Some("swift-meta-description"))
.expect("description tag not found");
assert_eq!(desc_tag.value.as_deref(), Some("test container"));
}
#[test]
fn test_swift_metadata_to_s3_tags_empty() {
let metadata = std::collections::HashMap::new();
let tagging = swift_metadata_to_s3_tags(&metadata);
assert!(tagging.is_none());
}
#[test]
fn test_swift_metadata_to_s3_tags_case_normalization() {
let mut metadata = std::collections::HashMap::new();
metadata.insert("Color".to_string(), "Red".to_string());
metadata.insert("PRIORITY".to_string(), "High".to_string());
let tagging = swift_metadata_to_s3_tags(&metadata).unwrap();
// Keys should be lowercased
assert!(tagging.tag_set.iter().any(|t| t.key.as_deref() == Some("swift-meta-color")));
assert!(
tagging
.tag_set
.iter()
.any(|t| t.key.as_deref() == Some("swift-meta-priority"))
);
// Values should be preserved as-is
let color_tag = tagging
.tag_set
.iter()
.find(|t| t.key.as_deref() == Some("swift-meta-color"))
.unwrap();
assert_eq!(color_tag.value.as_deref(), Some("Red"));
}
#[test]
fn test_s3_tags_to_swift_metadata() {
let tagging = Tagging {
@@ -1487,91 +1394,80 @@ mod tests {
#[test]
fn test_metadata_roundtrip() {
// Test that we can convert metadata -> tags -> metadata without loss
let mut original_metadata = std::collections::HashMap::new();
original_metadata.insert("color".to_string(), "blue".to_string());
original_metadata.insert("owner".to_string(), "alice".to_string());
original_metadata.insert("priority".to_string(), "high".to_string());
// What a POST writes must be what a HEAD reads back: run the items
// through the tag-merge write path and the tag-parse read path.
let update = MetadataUpdate::default()
.set("color", "blue")
.set("owner", "alice")
.set("priority", "high");
let tagging = swift_metadata_to_s3_tags(&original_metadata).unwrap();
let recovered_metadata = s3_tags_to_swift_metadata(&tagging);
let tagging = update
.apply_to_tags(None, CONTAINER_META_TAG_PREFIX)
.expect("merge should be accepted");
let recovered = s3_tags_to_swift_metadata(&tagging);
assert_eq!(recovered_metadata.len(), original_metadata.len());
for (key, value) in &original_metadata {
assert_eq!(
recovered_metadata.get(&key.to_lowercase()),
Some(value),
"Metadata key {} not preserved in roundtrip",
key
);
}
assert_eq!(recovered.len(), 3);
assert_eq!(recovered.get("color").map(String::as_str), Some("blue"));
assert_eq!(recovered.get("owner").map(String::as_str), Some("alice"));
assert_eq!(recovered.get("priority").map(String::as_str), Some("high"));
}
#[test]
fn test_tag_preservation_merge_with_existing() {
// Test merging Swift metadata with existing non-Swift tags
let mut existing_tagging = Tagging {
// A container metadata POST shares its tag set with the container ACL
// and versioning tags, and with whatever S3 tags the bucket carries.
// Only the swift-meta-* items the POST names may change.
let existing = Tagging {
tag_set: vec![
Tag {
key: Some("swift-meta-color".to_string()),
value: Some("blue".to_string()),
},
Tag {
key: Some("swift-acl-read".to_string()),
value: Some(".r:*".to_string()),
},
Tag {
key: Some("swift-versions-location".to_string()),
value: Some("archive".to_string()),
},
Tag {
key: Some("env".to_string()),
value: Some("production".to_string()),
},
Tag {
key: Some("team".to_string()),
value: Some("backend".to_string()),
},
],
};
// Remove old swift-meta-* tags
existing_tagging.tag_set.retain(|tag| {
if let Some(key) = &tag.key {
!key.starts_with("swift-meta-")
} else {
true
}
});
let merged = MetadataUpdate::default()
.set("description", "test")
.apply_to_tags(Some(&existing), CONTAINER_META_TAG_PREFIX)
.expect("merge should be accepted");
let tag = |key: &str| {
merged
.tag_set
.iter()
.find(|t| t.key.as_deref() == Some(key))
.and_then(|t| t.value.as_deref())
};
// Add new Swift metadata
let mut new_metadata = std::collections::HashMap::new();
new_metadata.insert("description".to_string(), "test".to_string());
let mut new_tagging = swift_metadata_to_s3_tags(&new_metadata).unwrap();
// Merge
existing_tagging.tag_set.append(&mut new_tagging.tag_set);
// Verify: should have env, team, and new swift-meta-description
assert_eq!(existing_tagging.tag_set.len(), 3);
let has_env = existing_tagging.tag_set.iter().any(|t| t.key.as_deref() == Some("env"));
let has_team = existing_tagging.tag_set.iter().any(|t| t.key.as_deref() == Some("team"));
let has_description = existing_tagging
.tag_set
.iter()
.any(|t| t.key.as_deref() == Some("swift-meta-description"));
assert!(has_env, "env tag should be preserved");
assert!(has_team, "team tag should be preserved");
assert!(has_description, "swift-meta-description should be added");
assert_eq!(tag("swift-meta-description"), Some("test"), "the new item should be added");
assert_eq!(tag("swift-meta-color"), Some("blue"), "an unnamed item should be preserved");
assert_eq!(tag("swift-acl-read"), Some(".r:*"), "the ACL tag should be preserved");
assert_eq!(tag("swift-versions-location"), Some("archive"), "the versioning tag should be preserved");
assert_eq!(tag("env"), Some("production"), "non-Swift tags should be preserved");
assert_eq!(merged.tag_set.len(), 5);
}
#[test]
fn test_tag_preservation_remove_only_swift() {
// Test that clearing Swift metadata preserves non-Swift tags
let mut existing_tagging = Tagging {
// Removing the last container metadata item leaves the other tags
// and so must not clear the tagging config.
let existing = Tagging {
tag_set: vec![
Tag {
key: Some("swift-meta-color".to_string()),
value: Some("blue".to_string()),
},
Tag {
key: Some("swift-meta-owner".to_string()),
value: Some("alice".to_string()),
},
Tag {
key: Some("env".to_string()),
value: Some("production".to_string()),
@@ -1583,37 +1479,28 @@ mod tests {
],
};
// Remove swift-meta-* tags (simulating empty metadata update)
existing_tagging.tag_set.retain(|tag| {
if let Some(key) = &tag.key {
!key.starts_with("swift-meta-")
} else {
true
}
});
let merged = MetadataUpdate::default()
.remove("color")
.apply_to_tags(Some(&existing), CONTAINER_META_TAG_PREFIX)
.expect("merge should be accepted");
// Verify: should only have env and cost-center
assert_eq!(existing_tagging.tag_set.len(), 2);
let has_env = existing_tagging.tag_set.iter().any(|t| t.key.as_deref() == Some("env"));
let has_cost_center = existing_tagging
.tag_set
.iter()
.any(|t| t.key.as_deref() == Some("cost-center"));
let has_swift_meta = existing_tagging
.tag_set
.iter()
.any(|t| t.key.as_ref().is_some_and(|k| k.starts_with("swift-meta-")));
assert!(has_env, "env tag should be preserved");
assert!(has_cost_center, "cost-center tag should be preserved");
assert!(!has_swift_meta, "all swift-meta-* tags should be removed");
assert_eq!(merged.tag_set.len(), 2);
assert!(merged.tag_set.iter().any(|t| t.key.as_deref() == Some("env")));
assert!(merged.tag_set.iter().any(|t| t.key.as_deref() == Some("cost-center")));
assert!(
!merged
.tag_set
.iter()
.any(|t| t.key.as_ref().is_some_and(|k| k.starts_with(CONTAINER_META_TAG_PREFIX))),
"the removed item should be gone"
);
}
#[test]
fn test_tag_preservation_empty_after_swift_removal() {
// Test that if only Swift tags exist, clearing them results in empty tagging
let mut existing_tagging = Tagging {
// Removing every item when nothing else is tagged empties the tag set,
// which is how the caller knows to clear the tagging config.
let existing = Tagging {
tag_set: vec![
Tag {
key: Some("swift-meta-color".to_string()),
@@ -1626,38 +1513,13 @@ mod tests {
],
};
// Remove swift-meta-* tags
existing_tagging.tag_set.retain(|tag| {
if let Some(key) = &tag.key {
!key.starts_with("swift-meta-")
} else {
true
}
});
let merged = MetadataUpdate::default()
.remove("color")
.remove("owner")
.apply_to_tags(Some(&existing), CONTAINER_META_TAG_PREFIX)
.expect("merge should be accepted");
// Verify: should be empty
assert!(
existing_tagging.tag_set.is_empty(),
"tagging should be empty after removing all swift-meta-* tags"
);
}
#[test]
fn test_tag_preservation_no_existing_tags() {
// Test adding Swift metadata when no tags exist
let existing_tagging = Tagging { tag_set: vec![] };
let mut new_metadata = std::collections::HashMap::new();
new_metadata.insert("color".to_string(), "blue".to_string());
let mut new_tagging = swift_metadata_to_s3_tags(&new_metadata).unwrap();
let mut merged = existing_tagging.clone();
merged.tag_set.append(&mut new_tagging.tag_set);
// Verify: should have only the new Swift tag
assert_eq!(merged.tag_set.len(), 1);
assert_eq!(merged.tag_set[0].key.as_deref(), Some("swift-meta-color"));
assert_eq!(merged.tag_set[0].value.as_deref(), Some("blue"));
assert!(merged.tag_set.is_empty(), "tagging should be empty after removing every swift-meta-* tag");
}
// Object Versioning Tests
+17 -35
View File
@@ -20,6 +20,7 @@
use super::container;
use super::dlo;
use super::metadata_update::MetadataUpdate;
use super::object;
use super::slo;
use super::tempurl;
@@ -321,32 +322,15 @@ async fn handle_authenticated_request(
Err(SwiftError::NotImplemented("Swift Account HEAD operation not yet implemented".to_string()))
}
Method::POST => {
// Account metadata update - extract headers
let mut metadata = std::collections::HashMap::new();
// Account metadata update. Additive, per Swift: the request
// names the items to set (X-Account-Meta-*) and the items to
// drop (X-Remove-Account-Meta-*, or an empty value), and
// everything it does not name keeps its stored value. The
// TempURL signing key lives here, so a replacing POST would
// invalidate every outstanding signature for the account.
let update = MetadataUpdate::from_account_headers(&headers);
// Extract X-Account-Meta-* headers
for (key, value) in &headers {
let key_str = key.as_str();
if let Some(meta_key) = key_str.strip_prefix("x-account-meta-") {
// Strip "x-account-meta-"
if let Ok(value_str) = value.to_str() {
metadata.insert(meta_key.to_string(), value_str.to_string());
}
}
}
// Special handling for TempURL key headers
// X-Account-Meta-Temp-URL-Key or X-Account-Meta-Temp-Url-Key
if let Some(tempurl_key) = headers
.get("x-account-meta-temp-url-key")
.or_else(|| headers.get("x-account-meta-temp-Url-key"))
&& let Ok(key_str) = tempurl_key.to_str()
{
metadata.insert("temp-url-key".to_string(), key_str.to_string());
}
// Update account metadata
super::account::update_account_metadata(&account, &metadata, &credentials_opt).await?;
super::account::update_account_metadata(&account, &update, &credentials_opt).await?;
let trans_id = generate_trans_id();
Response::builder()
@@ -581,17 +565,15 @@ async fn handle_authenticated_request(
container::set_container_acl(&account, &container, new_read, new_write, &credentials).await?;
}
// Update container metadata - now we have access to request headers
let mut metadata = std::collections::HashMap::new();
for (name, value) in headers.iter() {
if let Some(meta_key) = name.as_str().strip_prefix("x-container-meta-")
&& let Ok(value_str) = value.to_str()
{
metadata.insert(meta_key.to_string(), value_str.to_string());
}
}
// Update container metadata. Additive, per Swift: items the
// request does not name keep their stored value, and removal
// is explicit (X-Remove-Container-Meta-*, or an empty value).
// Still called when the request named no item — an ACL-only
// or versioning-only POST — because this is what reports 404
// for a container that does not exist.
let update = MetadataUpdate::from_container_headers(&headers);
container::update_container_metadata(&account, &container, &credentials, metadata).await?;
container::update_container_metadata(&account, &container, &credentials, update).await?;
let trans_id = generate_trans_id();
Response::builder()
@@ -0,0 +1,410 @@
// 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.
//! Account and container metadata updates
//!
//! Swift account and container POSTs are *additive*: an item the request does
//! not mention keeps its stored value, and removal is explicit — either
//! `X-Remove-{Account,Container}-Meta-{name}`, or the item sent with an empty
//! value. Object POST is the one that replaces the whole set; `swift::object`
//! handles that separately and must keep doing so.
//!
//! Getting this wrong is not a cosmetic divergence. Account metadata holds the
//! TempURL signing key, so a POST that set an unrelated item while dropping
//! the rest would invalidate every outstanding TempURL and FormPost signature
//! for the account.
//!
//! This module turns a request's headers into the items to write and the items
//! to drop, and applies that to the bucket tag set the metadata is persisted
//! in.
use super::{MAX_METADATA_COUNT, MAX_METADATA_VALUE_SIZE, SwiftError, SwiftResult};
use axum::http::HeaderMap;
use s3s::dto::{Tag, Tagging};
use std::collections::{BTreeMap, BTreeSet};
/// Request header prefix carrying an account metadata item.
const ACCOUNT_META_HEADER_PREFIX: &str = "x-account-meta-";
/// Request header prefix removing an account metadata item.
const ACCOUNT_META_REMOVE_HEADER_PREFIX: &str = "x-remove-account-meta-";
/// Request header prefix carrying a container metadata item.
const CONTAINER_META_HEADER_PREFIX: &str = "x-container-meta-";
/// Request header prefix removing a container metadata item.
const CONTAINER_META_REMOVE_HEADER_PREFIX: &str = "x-remove-container-meta-";
/// Bucket-tag namespace holding account metadata items.
pub(crate) const ACCOUNT_META_TAG_PREFIX: &str = "swift-account-meta-";
/// Bucket-tag namespace holding container metadata items.
///
/// Deliberately narrower than the `swift-` tags around it: the container ACL
/// (`swift-acl-*`) and versioning (`swift-versions-location`) tags share this
/// tag set and must survive a metadata POST.
pub(crate) const CONTAINER_META_TAG_PREFIX: &str = "swift-meta-";
/// The metadata changes carried by one account or container POST.
///
/// Item names are held lowercased. Swift metadata names are case-insensitive,
/// HTTP header names arrive lowercased anyway, and a removal has to match the
/// name a previous POST stored — so normalizing once here is what makes
/// `X-Remove-Container-Meta-Color` find a stored `color`.
///
/// Ordered rather than hashed so the persisted tag set — and therefore the
/// serialized XML — comes out in a stable order for a given update.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct MetadataUpdate {
/// Items to write, by name.
items: BTreeMap<String, String>,
/// Names to drop.
removals: BTreeSet<String>,
}
impl MetadataUpdate {
/// Write `name` = `value`.
pub fn set(mut self, name: &str, value: &str) -> Self {
let name = name.to_lowercase();
self.removals.remove(&name);
self.items.insert(name, value.to_string());
self
}
/// Drop `name`, if it is stored.
pub fn remove(mut self, name: &str) -> Self {
let name = name.to_lowercase();
self.items.remove(&name);
self.removals.insert(name);
self
}
/// Parse the metadata headers of an account POST.
pub(crate) fn from_account_headers(headers: &HeaderMap) -> Self {
Self::from_headers(headers, ACCOUNT_META_HEADER_PREFIX, ACCOUNT_META_REMOVE_HEADER_PREFIX)
}
/// Parse the metadata headers of a container POST.
pub(crate) fn from_container_headers(headers: &HeaderMap) -> Self {
Self::from_headers(headers, CONTAINER_META_HEADER_PREFIX, CONTAINER_META_REMOVE_HEADER_PREFIX)
}
/// Parse metadata headers under `item_prefix`, and removals under
/// `remove_prefix`. Both prefixes are lowercase, matching how `http`
/// normalizes header names.
///
/// An item sent with an empty value is a removal — the deletion path
/// Swift clients use when they do not send a dedicated remove header.
/// A name carried by both header forms is removed: the explicit removal
/// wins, as it does in Swift, where a remove header is rewritten into an
/// empty-valued item header.
fn from_headers(headers: &HeaderMap, item_prefix: &str, remove_prefix: &str) -> Self {
let mut update = Self::default();
for (name, value) in headers {
let Some(item) = name.as_str().strip_prefix(item_prefix) else {
continue;
};
// A value that is not valid UTF-8 cannot be stored as a tag.
// Skipping it matches how the rest of the Swift handlers treat
// unreadable header values.
let Ok(value) = value.to_str() else {
continue;
};
update = if value.is_empty() {
update.remove(item)
} else {
update.set(item, value)
};
}
for name in headers.keys() {
if let Some(item) = name.as_str().strip_prefix(remove_prefix) {
update = update.remove(item);
}
}
update
}
/// Whether this update changes anything.
///
/// An ACL-only or versioning-only container POST produces an empty update:
/// it names no metadata item, so it must leave stored metadata alone.
pub(crate) fn is_empty(&self) -> bool {
self.items.is_empty() && self.removals.is_empty()
}
/// Reject oversized values before anything is persisted.
///
/// The item *count* is not checked here — an additive POST has to be
/// measured against the merged result, which only [`Self::apply_to_tags`]
/// can see.
pub(crate) fn validate(&self) -> SwiftResult<()> {
for (name, value) in &self.items {
if value.len() > MAX_METADATA_VALUE_SIZE {
return Err(SwiftError::BadRequest(format!(
"Metadata value for '{}' too large: {} bytes (max: {} bytes)",
name,
value.len(),
MAX_METADATA_VALUE_SIZE
)));
}
}
Ok(())
}
/// Merge this update into the persisted tag set.
///
/// `prefix` is the tag namespace holding the items. Tags outside it — the
/// container ACL and versioning tags, plus any S3 tags the bucket carries
/// — are untouched, and so are the namespaced items this update does not
/// name.
///
/// The item-count cap is checked against the merged result rather than the
/// request: because POSTs are additive, a client could otherwise walk past
/// it one header at a time. This runs inside the bucket metadata write
/// guard, so the count it checks is the one about to be persisted.
pub(crate) fn apply_to_tags(&self, current: Option<&Tagging>, prefix: &str) -> SwiftResult<Tagging> {
let mut tagging = current.cloned().unwrap_or_else(|| Tagging { tag_set: vec![] });
tagging.tag_set.retain(|tag| match item_name(tag, prefix) {
Some(name) => !self.items.contains_key(name) && !self.removals.contains(name),
None => true,
});
let merged = tagging.tag_set.iter().filter(|tag| item_name(tag, prefix).is_some()).count() + self.items.len();
if merged > MAX_METADATA_COUNT {
return Err(SwiftError::BadRequest(format!(
"Too many metadata headers: {} (max: {})",
merged, MAX_METADATA_COUNT
)));
}
for (name, value) in &self.items {
tagging.tag_set.push(Tag {
key: Some(format!("{}{}", prefix, name)),
value: Some(value.clone()),
});
}
Ok(tagging)
}
}
/// The metadata item name a tag carries, or `None` if the tag does not belong
/// to this namespace.
fn item_name<'a>(tag: &'a Tag, prefix: &str) -> Option<&'a str> {
tag.key.as_deref()?.strip_prefix(prefix)
}
#[cfg(test)]
mod tests {
use super::*;
use axum::http::{HeaderName, HeaderValue};
fn headers(pairs: &[(&str, &str)]) -> HeaderMap {
let mut map = HeaderMap::new();
for (name, value) in pairs {
map.insert(
HeaderName::from_bytes(name.as_bytes()).expect("test header name should parse"),
HeaderValue::from_str(value).expect("test header value should parse"),
);
}
map
}
fn tags(pairs: &[(&str, &str)]) -> Tagging {
Tagging {
tag_set: pairs
.iter()
.map(|(key, value)| Tag {
key: Some((*key).to_string()),
value: Some((*value).to_string()),
})
.collect(),
}
}
fn tag_value<'a>(tagging: &'a Tagging, key: &str) -> Option<&'a str> {
tagging
.tag_set
.iter()
.find(|tag| tag.key.as_deref() == Some(key))
.and_then(|tag| tag.value.as_deref())
}
#[test]
fn from_headers_collects_items_and_ignores_unrelated_headers() {
let update = MetadataUpdate::from_container_headers(&headers(&[
("x-container-meta-color", "blue"),
("x-container-read", ".r:*"),
("content-type", "text/plain"),
]));
assert_eq!(update, MetadataUpdate::default().set("color", "blue"));
}
#[test]
fn from_headers_lowercases_item_names() {
// `http` normalizes header names on the way in, so the mixed case a
// client sends is already gone by the time a handler sees it; the
// lowercasing here is what makes a directly built update match.
let update = MetadataUpdate::from_container_headers(&headers(&[("X-Container-Meta-Color", "Blue")]));
assert_eq!(update, MetadataUpdate::default().set("COLOR", "Blue"));
}
#[test]
fn empty_value_is_a_removal() {
let update = MetadataUpdate::from_container_headers(&headers(&[("x-container-meta-color", "")]));
assert_eq!(update, MetadataUpdate::default().remove("color"));
}
#[test]
fn remove_header_drops_the_item() {
let update = MetadataUpdate::from_account_headers(&headers(&[("x-remove-account-meta-temp-url-key", "x")]));
assert_eq!(update, MetadataUpdate::default().remove("temp-url-key"));
}
#[test]
fn remove_header_wins_over_a_value_for_the_same_item() {
let update = MetadataUpdate::from_container_headers(&headers(&[
("x-container-meta-color", "blue"),
("x-remove-container-meta-color", "x"),
]));
assert_eq!(update, MetadataUpdate::default().remove("color"));
}
#[test]
fn a_removal_header_is_not_mistaken_for_an_item() {
// "x-remove-container-meta-color" must not also parse as the item
// "remove-container-meta-color" or similar.
let update = MetadataUpdate::from_container_headers(&headers(&[("x-remove-container-meta-color", "x")]));
assert!(update.items.is_empty(), "a removal header must not set an item");
}
#[test]
fn a_request_with_no_metadata_headers_is_empty() {
assert!(MetadataUpdate::from_container_headers(&headers(&[("x-container-read", ".r:*")])).is_empty());
assert!(MetadataUpdate::default().is_empty());
assert!(!MetadataUpdate::default().remove("color").is_empty());
}
#[test]
fn apply_preserves_items_the_update_does_not_name() {
let current = tags(&[
("swift-meta-color", "blue"),
("swift-meta-season", "summer"),
("swift-acl-read", ".r:*"),
("swift-versions-location", "archive"),
("unrelated-s3-tag", "keep"),
]);
let merged = MetadataUpdate::default()
.set("mood", "calm")
.apply_to_tags(Some(&current), CONTAINER_META_TAG_PREFIX)
.expect("merge should be accepted");
assert_eq!(tag_value(&merged, "swift-meta-color"), Some("blue"));
assert_eq!(tag_value(&merged, "swift-meta-season"), Some("summer"));
assert_eq!(tag_value(&merged, "swift-meta-mood"), Some("calm"));
assert_eq!(tag_value(&merged, "swift-acl-read"), Some(".r:*"));
assert_eq!(tag_value(&merged, "swift-versions-location"), Some("archive"));
assert_eq!(tag_value(&merged, "unrelated-s3-tag"), Some("keep"));
}
#[test]
fn apply_overwrites_a_named_item_exactly_once() {
let current = tags(&[("swift-meta-color", "blue")]);
let merged = MetadataUpdate::default()
.set("color", "red")
.apply_to_tags(Some(&current), CONTAINER_META_TAG_PREFIX)
.expect("merge should be accepted");
assert_eq!(merged.tag_set.len(), 1, "overwriting must not duplicate the tag");
assert_eq!(tag_value(&merged, "swift-meta-color"), Some("red"));
}
#[test]
fn apply_drops_only_the_removed_item() {
let current = tags(&[
("swift-meta-color", "blue"),
("swift-meta-season", "summer"),
("swift-acl-read", ".r:*"),
]);
let merged = MetadataUpdate::default()
.remove("color")
.apply_to_tags(Some(&current), CONTAINER_META_TAG_PREFIX)
.expect("merge should be accepted");
assert_eq!(tag_value(&merged, "swift-meta-color"), None);
assert_eq!(tag_value(&merged, "swift-meta-season"), Some("summer"));
assert_eq!(tag_value(&merged, "swift-acl-read"), Some(".r:*"));
}
#[test]
fn apply_to_an_untagged_bucket_starts_from_nothing() {
let merged = MetadataUpdate::default()
.set("color", "blue")
.apply_to_tags(None, ACCOUNT_META_TAG_PREFIX)
.expect("merge should be accepted");
assert_eq!(tag_value(&merged, "swift-account-meta-color"), Some("blue"));
}
#[test]
fn apply_caps_the_merged_item_count_not_the_request() {
let current = Tagging {
tag_set: (0..MAX_METADATA_COUNT)
.map(|i| Tag {
key: Some(format!("{}item{}", CONTAINER_META_TAG_PREFIX, i)),
value: Some("v".to_string()),
})
.collect(),
};
// One more item than the container already stores: rejected, even
// though the request itself carries a single header.
let err = MetadataUpdate::default()
.set("overflow", "v")
.apply_to_tags(Some(&current), CONTAINER_META_TAG_PREFIX)
.expect_err("exceeding the item cap must be rejected");
assert!(matches!(err, SwiftError::BadRequest(_)), "expected BadRequest, got {err:?}");
// Overwriting an item already counted stays at the cap.
MetadataUpdate::default()
.set("item0", "v2")
.apply_to_tags(Some(&current), CONTAINER_META_TAG_PREFIX)
.expect("overwriting an existing item must not trip the cap");
}
#[test]
fn validate_rejects_oversized_values() {
let err = MetadataUpdate::default()
.set("color", &"b".repeat(MAX_METADATA_VALUE_SIZE + 1))
.validate()
.expect_err("an oversized value must be rejected");
assert!(matches!(err, SwiftError::BadRequest(_)), "expected BadRequest, got {err:?}");
MetadataUpdate::default()
.set("color", &"b".repeat(MAX_METADATA_VALUE_SIZE))
.validate()
.expect("a value at the limit must be accepted");
}
}
+6 -3
View File
@@ -44,6 +44,7 @@ pub mod expiration;
pub mod expiration_worker;
pub mod formpost;
pub mod handler;
pub mod metadata_update;
pub mod object;
pub mod quota;
pub mod router;
@@ -57,6 +58,7 @@ pub mod types;
pub mod versioning;
pub use errors::{SwiftError, SwiftResult};
pub use metadata_update::MetadataUpdate;
pub use router::{SwiftRoute, SwiftRouter};
/// Maximum number of metadata headers allowed per resource (Swift standard)
@@ -71,9 +73,10 @@ pub(crate) const MAX_METADATA_VALUE_SIZE: usize = 256;
/// - Total number of metadata entries doesn't exceed MAX_METADATA_COUNT
/// - Individual metadata values don't exceed MAX_METADATA_VALUE_SIZE
///
/// Applies to object, container and account metadata alike: all three are
/// persisted, and container/account metadata additionally lands in the
/// bucket metadata file that every later config write rewrites in full.
/// This is the object-metadata form, where a POST replaces the whole set and
/// the request is therefore the complete result. Account and container POSTs
/// are additive, so their count has to be measured against the merged state
/// instead — see [`MetadataUpdate::apply_to_tags`].
///
/// Returns error if limits are exceeded.
pub(crate) fn validate_metadata(metadata: &std::collections::HashMap<String, String>) -> SwiftResult<()> {
+23 -2
View File
@@ -88,6 +88,10 @@ pub(crate) async fn get_swift_bucket_metadata(bucket: &str) -> SwiftStorageResul
/// disk-truth reloads. Peers are then told to reload, matching what the S3
/// handlers do after a config write.
///
/// A rewrite may also refuse the update outright, for limits that can only be
/// judged against the state being merged into. That verdict is the client's
/// answer, so it is returned as-is rather than folded into a storage error.
///
/// Storage failures are logged in full and reported to the client as a
/// generic error: these now carry real disk and quorum detail, which does not
/// belong in a Swift response body. The one exception is an unreadable
@@ -95,8 +99,12 @@ pub(crate) async fn get_swift_bucket_metadata(bucket: &str) -> SwiftStorageResul
/// to act on it.
pub(crate) async fn update_swift_bucket_tagging<F>(bucket: String, rewrite: F) -> SwiftResult<()>
where
F: FnOnce(Option<&Tagging>) -> Tagging + Send,
F: FnOnce(Option<&Tagging>) -> SwiftResult<Tagging> + Send,
{
// Carries a rewrite's refusal back out past the storage error the config
// write has to fail with to abort the transaction.
let mut rejected = None;
let result = update_config_with(&bucket, BUCKET_TAGGING_CONFIG, |bm| {
// Merging onto an unparseable tag set would silently drop every tag
// the bucket has — including the container ACL and versioning tags —
@@ -106,7 +114,14 @@ where
return Err(SwiftStorageError::other(UNREADABLE_TAGGING_SENTINEL));
}
let tagging = rewrite(bm.tagging_config.as_ref());
let tagging = match rewrite(bm.tagging_config.as_ref()) {
Ok(tagging) => tagging,
Err(err) => {
rejected = Some(err);
return Err(SwiftStorageError::other("swift: tagging rewrite rejected the update"));
}
};
if tagging.tag_set.is_empty() {
Ok(Vec::new())
} else {
@@ -118,6 +133,12 @@ where
})
.await;
// Nothing was written, but that is the rewrite's own decision about the
// request rather than a storage fault, so it is not logged as one.
if let Some(err) = rejected {
return Err(err);
}
if let Err(err) = result {
let unreadable = err.to_string().contains(UNREADABLE_TAGGING_SENTINEL);
tracing::error!(
@@ -16,15 +16,20 @@
//! metadata file, not just the in-memory cache. The metadata has to survive
//! the disk-truth reloads performed by peer LoadBucketMetadata notifications
//! and the periodic refresh loop — and, transitively, a process restart.
//!
//! Durability is what makes the *content* of those writes matter, so this file
//! also covers Swift's additive account/container POST semantics: an item the
//! request does not name keeps its stored value, and removal is explicit. The
//! two are tested together because a reload is the only way to tell a real
//! merge from one that happened to look right in the cache.
#![cfg(feature = "swift")]
use std::collections::HashMap;
use rustfs_credentials::Credentials;
use rustfs_protocols::swift::SwiftError;
use rustfs_protocols::swift::container::{ContainerMapper, update_container_metadata};
use rustfs_protocols::swift::{account, container};
use rustfs_protocols::swift::{MetadataUpdate, SwiftError, account, container};
use rustfs_test_utils::TestECStoreEnv;
use serde_json::json;
use sha2::{Digest, Sha256};
@@ -97,6 +102,8 @@ async fn swift_metadata_writes_are_durable() {
posts_survive_disk_truth_reload(&env).await;
tag_writers_preserve_each_others_state(&env).await;
unrelated_account_posts_preserve_the_tempurl_key().await;
acl_only_posts_leave_container_metadata_alone(&env).await;
versioning_writes_reject_missing_containers().await;
}
@@ -109,11 +116,14 @@ async fn posts_survive_disk_truth_reload(env: &TestECStoreEnv) {
let bucket = ContainerMapper::default().swift_to_s3_bucket(swift_container, project_id);
env.make_bucket(&bucket, false).await;
let mut metadata = HashMap::new();
metadata.insert("color".to_string(), "blue".to_string());
update_container_metadata(&swift_account, swift_container, &credentials, metadata)
.await
.expect("container metadata POST should succeed");
update_container_metadata(
&swift_account,
swift_container,
&credentials,
MetadataUpdate::default().set("color", "blue"),
)
.await
.expect("container metadata POST should succeed");
reload_bucket_metadata_from_disk(&bucket).await;
@@ -124,22 +134,47 @@ async fn posts_survive_disk_truth_reload(env: &TestECStoreEnv) {
"container metadata POST must survive a disk-truth metadata reload"
);
// A follow-up POST replaces the Swift metadata and that replacement must
// survive a reload too (the rewrite merges against disk state, so the
// previous value must actually be gone).
let mut metadata = HashMap::new();
metadata.insert("season".to_string(), "summer".to_string());
update_container_metadata(&swift_account, swift_container, &credentials, metadata)
.await
.expect("second container metadata POST should succeed");
// A follow-up POST names a different item. Swift POSTs are additive, so
// the new item lands and the first one stays — and both are still there
// after a reload, which is what proves the merge ran against disk state
// rather than looking right in the cache.
update_container_metadata(
&swift_account,
swift_container,
&credentials,
MetadataUpdate::default().set("season", "summer"),
)
.await
.expect("second container metadata POST should succeed");
reload_bucket_metadata_from_disk(&bucket).await;
let container_meta = persisted_container_metadata(&bucket).await;
assert_eq!(container_meta.get("season").map(String::as_str), Some("summer"));
assert_eq!(
container_meta.get("color").map(String::as_str),
Some("blue"),
"a POST that does not name an item must not delete it"
);
// X-Remove-Container-Meta-Color is the deletion path, and it has to be
// durable in the other direction: the removed item must not come back
// from the persisted tags on reload.
update_container_metadata(&swift_account, swift_container, &credentials, MetadataUpdate::default().remove("color"))
.await
.expect("container metadata removal should succeed");
reload_bucket_metadata_from_disk(&bucket).await;
let container_meta = persisted_container_metadata(&bucket).await;
assert!(
!container_meta.contains_key("color"),
"replaced container metadata must not resurrect on reload"
"an explicitly removed item must not resurrect on reload"
);
assert_eq!(
container_meta.get("season").map(String::as_str),
Some("summer"),
"removing one item must not disturb the others"
);
// --- Container versioning POST (X-Versions-Location) ---
@@ -163,11 +198,13 @@ async fn posts_survive_disk_truth_reload(env: &TestECStoreEnv) {
);
// --- Account metadata POST (TempURL keys etc.) ---
let mut account_meta = HashMap::new();
account_meta.insert("temp-url-key".to_string(), "s3cr3t".to_string());
account::update_account_metadata(&swift_account, &account_meta, &Some(credentials.clone()))
.await
.expect("account metadata POST should succeed");
account::update_account_metadata(
&swift_account,
&MetadataUpdate::default().set("temp-url-key", "s3cr3t"),
&Some(credentials.clone()),
)
.await
.expect("account metadata POST should succeed");
reload_bucket_metadata_from_disk(&account_metadata_bucket_name(&swift_account)).await;
@@ -196,9 +233,7 @@ async fn tag_writers_preserve_each_others_state(env: &TestECStoreEnv) {
env.make_bucket(&ContainerMapper::default().swift_to_s3_bucket(archive, project_id), false)
.await;
let mut metadata = HashMap::new();
metadata.insert("color".to_string(), "blue".to_string());
update_container_metadata(&swift_account, container, &credentials, metadata)
update_container_metadata(&swift_account, container, &credentials, MetadataUpdate::default().set("color", "blue"))
.await
.expect("container metadata POST should succeed");
container::enable_versioning(&swift_account, container, archive, &credentials)
@@ -252,6 +287,111 @@ async fn tag_writers_preserve_each_others_state(env: &TestECStoreEnv) {
assert!(!acl.read.is_empty(), "disable_versioning must not disturb the ACL");
}
/// The failure additive POSTs exist to prevent. Account metadata holds the
/// TempURL signing key, so a POST that replaced the whole set — setting a
/// quota, say — would delete that key and permanently invalidate every
/// outstanding TempURL and FormPost signature for the account. Durable
/// storage made that loss permanent instead of a cache blip, so the check
/// runs against reloaded disk state.
///
/// Relies on the ambient store the caller's `TestECStoreEnv` published; the
/// account metadata bucket is created by the write path itself.
async fn unrelated_account_posts_preserve_the_tempurl_key() {
let project_id = "swiftmergeproj";
let swift_account = format!("AUTH_{project_id}");
let credentials = Some(keystone_credentials(project_id));
let account_bucket = account_metadata_bucket_name(&swift_account);
account::update_account_metadata(&swift_account, &MetadataUpdate::default().set("temp-url-key", "s3cr3t"), &credentials)
.await
.expect("TempURL key POST should succeed");
// A later POST that has nothing to do with the TempURL key.
account::update_account_metadata(&swift_account, &MetadataUpdate::default().set("quota-bytes", "100"), &credentials)
.await
.expect("quota POST should succeed");
reload_bucket_metadata_from_disk(&account_bucket).await;
let loaded = account::get_account_metadata(&swift_account, &None)
.await
.expect("account metadata should load");
assert_eq!(
loaded.get("temp-url-key").map(String::as_str),
Some("s3cr3t"),
"an unrelated account POST must not delete the TempURL signing key"
);
assert_eq!(loaded.get("quota-bytes").map(String::as_str), Some("100"));
// And the lookup that actually breaks when the key is lost still resolves.
assert_eq!(
account::get_tempurl_key(&swift_account, &None)
.await
.expect("TempURL key should load")
.as_deref(),
Some("s3cr3t"),
"TempURL signature validation must still find the key"
);
// X-Remove-Account-Meta-Quota-Bytes drops that item and nothing else.
account::update_account_metadata(&swift_account, &MetadataUpdate::default().remove("quota-bytes"), &credentials)
.await
.expect("account metadata removal should succeed");
reload_bucket_metadata_from_disk(&account_bucket).await;
let loaded = account::get_account_metadata(&swift_account, &None)
.await
.expect("account metadata should load");
assert!(
!loaded.contains_key("quota-bytes"),
"an explicitly removed item must not resurrect on reload"
);
assert_eq!(
loaded.get("temp-url-key").map(String::as_str),
Some("s3cr3t"),
"removing one item must not disturb the TempURL key"
);
}
/// An ACL-only or versioning-only container POST carries no `X-Container-Meta-*`
/// header, but the handler still runs the metadata step on every POST. That
/// step must leave stored metadata alone rather than treating "named nothing"
/// as "wants everything gone".
async fn acl_only_posts_leave_container_metadata_alone(env: &TestECStoreEnv) {
let project_id = "swiftaclonlyproj";
let swift_account = format!("AUTH_{project_id}");
let credentials = keystone_credentials(project_id);
let container = "gallery";
let bucket = ContainerMapper::default().swift_to_s3_bucket(container, project_id);
env.make_bucket(&bucket, false).await;
update_container_metadata(&swift_account, container, &credentials, MetadataUpdate::default().set("color", "blue"))
.await
.expect("container metadata POST should succeed");
// What the handler does for `POST … -H 'X-Container-Read: .r:*'`: set the
// ACL, then run the metadata step with an update that names no item.
container::set_container_acl(&swift_account, container, Some(".r:*"), None, &credentials)
.await
.expect("ACL-only POST should succeed");
update_container_metadata(&swift_account, container, &credentials, MetadataUpdate::default())
.await
.expect("the metadata step of an ACL-only POST should succeed");
reload_bucket_metadata_from_disk(&bucket).await;
assert_eq!(
persisted_container_metadata(&bucket).await.get("color").map(String::as_str),
Some("blue"),
"an ACL-only POST must not wipe X-Container-Meta-*"
);
let acl = container::get_container_acl(&swift_account, container, &credentials)
.await
.expect("container ACL should load");
assert!(!acl.read.is_empty(), "the ACL that POST set must be stored");
}
/// A container that does not exist must not get metadata persisted for it:
/// the metadata loader turns "nothing on disk" into a fresh default, so an
/// unguarded rewrite would create an orphan metadata file and cache a
@@ -270,6 +410,16 @@ async fn versioning_writes_reject_missing_containers() {
"expected NotFound for a missing container, got {err:?}"
);
// Naming no item skips the persisted write, but must not skip the
// existence check that turns a POST to a missing container into a 404.
let err = update_container_metadata(&swift_account, missing, &credentials, MetadataUpdate::default())
.await
.expect_err("a metadata POST to a missing container must fail");
assert!(
matches!(err, SwiftError::NotFound(_)),
"expected NotFound for a missing container, got {err:?}"
);
let bucket = ContainerMapper::default().swift_to_s3_bucket(missing, project_id);
assert!(
get_bucket_metadata(&bucket).await.is_err(),
@@ -291,8 +441,7 @@ async fn account_metadata_write_rejects_foreign_and_anonymous_callers() {
let victim_account = "AUTH_victimproject";
let attacker_credentials = keystone_credentials("attackerproject");
let mut poisoned = HashMap::new();
poisoned.insert("temp-url-key".to_string(), "attacker-key".to_string());
let poisoned = MetadataUpdate::default().set("temp-url-key", "attacker-key");
let err = account::update_account_metadata(victim_account, &poisoned, &Some(attacker_credentials))
.await
@@ -484,6 +484,59 @@ pub struct DeletePathsResponse {
pub error: ::core::option::Option<Error>,
}
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
pub struct SnapshotLeaseRequest {
#[prost(string, tag = "1")]
pub disk: ::prost::alloc::string::String,
#[prost(string, tag = "2")]
pub volume: ::prost::alloc::string::String,
#[prost(string, tag = "3")]
pub path: ::prost::alloc::string::String,
#[prost(uint64, tag = "4")]
pub ttl_ms: u64,
}
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
pub struct SnapshotLeaseRenewRequest {
#[prost(string, tag = "1")]
pub disk: ::prost::alloc::string::String,
#[prost(string, tag = "2")]
pub volume: ::prost::alloc::string::String,
#[prost(string, tag = "3")]
pub path: ::prost::alloc::string::String,
#[prost(bytes = "bytes", tag = "4")]
pub token: ::prost::bytes::Bytes,
#[prost(uint64, tag = "5")]
pub ttl_ms: u64,
}
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
pub struct SnapshotLeaseReleaseRequest {
#[prost(string, tag = "1")]
pub disk: ::prost::alloc::string::String,
#[prost(string, tag = "2")]
pub volume: ::prost::alloc::string::String,
#[prost(string, tag = "3")]
pub path: ::prost::alloc::string::String,
#[prost(bytes = "bytes", tag = "4")]
pub token: ::prost::bytes::Bytes,
}
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
pub struct SnapshotLeaseResponse {
#[prost(bool, tag = "1")]
pub success: bool,
#[prost(bytes = "bytes", tag = "2")]
pub token: ::prost::bytes::Bytes,
#[prost(uint32, tag = "3")]
pub protocol_version: u32,
#[prost(message, optional, tag = "4")]
pub error: ::core::option::Option<Error>,
}
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
pub struct SnapshotLeaseMutationResponse {
#[prost(bool, tag = "1")]
pub success: bool,
#[prost(message, optional, tag = "2")]
pub error: ::core::option::Option<Error>,
}
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
pub struct ReadMetadataRequest {
#[prost(string, tag = "1")]
pub disk: ::prost::alloc::string::String,
@@ -1595,6 +1648,51 @@ pub mod node_service_client {
.insert(GrpcMethod::new("node_service.NodeService", "Delete"));
self.inner.unary(req, path, codec).await
}
pub async fn acquire_snapshot_lease(
&mut self,
request: impl tonic::IntoRequest<super::SnapshotLeaseRequest>,
) -> std::result::Result<tonic::Response<super::SnapshotLeaseResponse>, tonic::Status> {
self.inner
.ready()
.await
.map_err(|e| tonic::Status::unknown(format!("Service was not ready: {}", e.into())))?;
let codec = tonic_prost::ProstCodec::default();
let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/AcquireSnapshotLease");
let mut req = request.into_request();
req.extensions_mut()
.insert(GrpcMethod::new("node_service.NodeService", "AcquireSnapshotLease"));
self.inner.unary(req, path, codec).await
}
pub async fn renew_snapshot_lease(
&mut self,
request: impl tonic::IntoRequest<super::SnapshotLeaseRenewRequest>,
) -> std::result::Result<tonic::Response<super::SnapshotLeaseResponse>, tonic::Status> {
self.inner
.ready()
.await
.map_err(|e| tonic::Status::unknown(format!("Service was not ready: {}", e.into())))?;
let codec = tonic_prost::ProstCodec::default();
let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/RenewSnapshotLease");
let mut req = request.into_request();
req.extensions_mut()
.insert(GrpcMethod::new("node_service.NodeService", "RenewSnapshotLease"));
self.inner.unary(req, path, codec).await
}
pub async fn release_snapshot_lease(
&mut self,
request: impl tonic::IntoRequest<super::SnapshotLeaseReleaseRequest>,
) -> std::result::Result<tonic::Response<super::SnapshotLeaseMutationResponse>, tonic::Status> {
self.inner
.ready()
.await
.map_err(|e| tonic::Status::unknown(format!("Service was not ready: {}", e.into())))?;
let codec = tonic_prost::ProstCodec::default();
let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/ReleaseSnapshotLease");
let mut req = request.into_request();
req.extensions_mut()
.insert(GrpcMethod::new("node_service.NodeService", "ReleaseSnapshotLease"));
self.inner.unary(req, path, codec).await
}
pub async fn verify_file(
&mut self,
request: impl tonic::IntoRequest<super::VerifyFileRequest>,
@@ -2784,6 +2882,18 @@ pub mod node_service_server {
&self,
request: tonic::Request<super::DeleteRequest>,
) -> std::result::Result<tonic::Response<super::DeleteResponse>, tonic::Status>;
async fn acquire_snapshot_lease(
&self,
request: tonic::Request<super::SnapshotLeaseRequest>,
) -> std::result::Result<tonic::Response<super::SnapshotLeaseResponse>, tonic::Status>;
async fn renew_snapshot_lease(
&self,
request: tonic::Request<super::SnapshotLeaseRenewRequest>,
) -> std::result::Result<tonic::Response<super::SnapshotLeaseResponse>, tonic::Status>;
async fn release_snapshot_lease(
&self,
request: tonic::Request<super::SnapshotLeaseReleaseRequest>,
) -> std::result::Result<tonic::Response<super::SnapshotLeaseMutationResponse>, tonic::Status>;
async fn verify_file(
&self,
request: tonic::Request<super::VerifyFileRequest>,
@@ -3426,6 +3536,90 @@ pub mod node_service_server {
};
Box::pin(fut)
}
"/node_service.NodeService/AcquireSnapshotLease" => {
#[allow(non_camel_case_types)]
struct AcquireSnapshotLeaseSvc<T: NodeService>(pub Arc<T>);
impl<T: NodeService> tonic::server::UnaryService<super::SnapshotLeaseRequest> for AcquireSnapshotLeaseSvc<T> {
type Response = super::SnapshotLeaseResponse;
type Future = BoxFuture<tonic::Response<Self::Response>, tonic::Status>;
fn call(&mut self, request: tonic::Request<super::SnapshotLeaseRequest>) -> Self::Future {
let inner = Arc::clone(&self.0);
let fut = async move { <T as NodeService>::acquire_snapshot_lease(&inner, request).await };
Box::pin(fut)
}
}
let accept_compression_encodings = self.accept_compression_encodings;
let send_compression_encodings = self.send_compression_encodings;
let max_decoding_message_size = self.max_decoding_message_size;
let max_encoding_message_size = self.max_encoding_message_size;
let inner = self.inner.clone();
let fut = async move {
let method = AcquireSnapshotLeaseSvc(inner);
let codec = tonic_prost::ProstCodec::default();
let mut grpc = tonic::server::Grpc::new(codec)
.apply_compression_config(accept_compression_encodings, send_compression_encodings)
.apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size);
let res = grpc.unary(method, req).await;
Ok(res)
};
Box::pin(fut)
}
"/node_service.NodeService/RenewSnapshotLease" => {
#[allow(non_camel_case_types)]
struct RenewSnapshotLeaseSvc<T: NodeService>(pub Arc<T>);
impl<T: NodeService> tonic::server::UnaryService<super::SnapshotLeaseRenewRequest> for RenewSnapshotLeaseSvc<T> {
type Response = super::SnapshotLeaseResponse;
type Future = BoxFuture<tonic::Response<Self::Response>, tonic::Status>;
fn call(&mut self, request: tonic::Request<super::SnapshotLeaseRenewRequest>) -> Self::Future {
let inner = Arc::clone(&self.0);
let fut = async move { <T as NodeService>::renew_snapshot_lease(&inner, request).await };
Box::pin(fut)
}
}
let accept_compression_encodings = self.accept_compression_encodings;
let send_compression_encodings = self.send_compression_encodings;
let max_decoding_message_size = self.max_decoding_message_size;
let max_encoding_message_size = self.max_encoding_message_size;
let inner = self.inner.clone();
let fut = async move {
let method = RenewSnapshotLeaseSvc(inner);
let codec = tonic_prost::ProstCodec::default();
let mut grpc = tonic::server::Grpc::new(codec)
.apply_compression_config(accept_compression_encodings, send_compression_encodings)
.apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size);
let res = grpc.unary(method, req).await;
Ok(res)
};
Box::pin(fut)
}
"/node_service.NodeService/ReleaseSnapshotLease" => {
#[allow(non_camel_case_types)]
struct ReleaseSnapshotLeaseSvc<T: NodeService>(pub Arc<T>);
impl<T: NodeService> tonic::server::UnaryService<super::SnapshotLeaseReleaseRequest> for ReleaseSnapshotLeaseSvc<T> {
type Response = super::SnapshotLeaseMutationResponse;
type Future = BoxFuture<tonic::Response<Self::Response>, tonic::Status>;
fn call(&mut self, request: tonic::Request<super::SnapshotLeaseReleaseRequest>) -> Self::Future {
let inner = Arc::clone(&self.0);
let fut = async move { <T as NodeService>::release_snapshot_lease(&inner, request).await };
Box::pin(fut)
}
}
let accept_compression_encodings = self.accept_compression_encodings;
let send_compression_encodings = self.send_compression_encodings;
let max_decoding_message_size = self.max_decoding_message_size;
let max_encoding_message_size = self.max_encoding_message_size;
let inner = self.inner.clone();
let fut = async move {
let method = ReleaseSnapshotLeaseSvc(inner);
let codec = tonic_prost::ProstCodec::default();
let mut grpc = tonic::server::Grpc::new(codec)
.apply_compression_config(accept_compression_encodings, send_compression_encodings)
.apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size);
let res = grpc.unary(method, req).await;
Ok(res)
};
Box::pin(fut)
}
"/node_service.NodeService/VerifyFile" => {
#[allow(non_camel_case_types)]
struct VerifyFileSvc<T: NodeService>(pub Arc<T>);
+454 -1
View File
@@ -447,10 +447,23 @@ impl CanonicalBodyBuilder {
self.push_bytes(field.as_bytes())
}
fn push_optional_str(&mut self, field: Option<&str>) -> Result<(), std::num::TryFromIntError> {
self.body.push(u8::from(field.is_some()));
self.push_str(field.unwrap_or_default())
}
fn push_bool(&mut self, field: bool) {
self.body.push(u8::from(field));
}
fn push_u32(&mut self, field: u32) {
self.body.extend_from_slice(&field.to_be_bytes());
}
fn push_u64(&mut self, field: u64) {
self.body.extend_from_slice(&field.to_be_bytes());
}
fn push_count(&mut self, count: usize) -> Result<(), std::num::TryFromIntError> {
self.body.extend_from_slice(&u64::try_from(count)?.to_be_bytes());
Ok(())
@@ -461,6 +474,211 @@ impl CanonicalBodyBuilder {
}
}
pub const PEER_RESTSIGNAL: &str = "signal";
pub const PEER_RESTSUB_SYS: &str = "sub-sys";
pub const PEER_RESTDRY_RUN: &str = "dry-run";
/// A stable semantic body for a side-effecting unary RPC.
///
/// Implementations deliberately enumerate handler-consumed fields instead of re-encoding the
/// protobuf message, whose unknown fields and map order are not a mixed-version contract.
pub trait CanonicalMutationBody {
fn canonical_body(&self) -> Result<Vec<u8>, std::num::TryFromIntError>;
}
macro_rules! impl_canonical_mutation_body {
($request:ty, $domain:expr, |$value:ident, $body:ident| $fields:block) => {
impl CanonicalMutationBody for $request {
fn canonical_body(&self) -> Result<Vec<u8>, std::num::TryFromIntError> {
let $value = self;
let mut $body = CanonicalBodyBuilder::new($domain);
$fields
Ok($body.finish())
}
}
};
($request:ty, $domain:expr) => {
impl CanonicalMutationBody for $request {
fn canonical_body(&self) -> Result<Vec<u8>, std::num::TryFromIntError> {
Ok(CanonicalBodyBuilder::new($domain).finish())
}
}
};
}
impl_canonical_mutation_body!(
proto_gen::node_service::SignalServiceRequest,
b"rustfs-signal-service-request-v1\0",
|request, body| {
let vars = request.vars.as_ref().map(|vars| &vars.value);
body.push_optional_str(vars.and_then(|vars| vars.get(PEER_RESTSIGNAL).map(String::as_str)))?;
body.push_optional_str(vars.and_then(|vars| vars.get(PEER_RESTSUB_SYS).map(String::as_str)))?;
body.push_optional_str(vars.and_then(|vars| vars.get(PEER_RESTDRY_RUN).map(String::as_str)))?;
}
);
impl_canonical_mutation_body!(
proto_gen::node_service::HealBucketRequest,
b"rustfs-heal-bucket-request-v1\0",
|request, body| {
body.push_str(&request.bucket)?;
body.push_str(&request.options)?;
}
);
impl_canonical_mutation_body!(
proto_gen::node_service::MakeBucketRequest,
b"rustfs-make-bucket-request-v1\0",
|request, body| {
body.push_str(&request.name)?;
body.push_str(&request.options)?;
}
);
impl_canonical_mutation_body!(
proto_gen::node_service::DeleteBucketRequest,
b"rustfs-delete-bucket-request-v1\0",
|request, body| {
body.push_str(&request.bucket)?;
body.push_str(&request.options)?;
}
);
impl_canonical_mutation_body!(
proto_gen::node_service::GenerallyLockRequest,
b"rustfs-lock-request-v1\0",
|request, body| {
body.push_str(&request.args)?;
}
);
impl_canonical_mutation_body!(
proto_gen::node_service::BatchGenerallyLockRequest,
b"rustfs-lock-batch-request-v1\0",
|request, body| {
body.push_count(request.args.len())?;
for arg in &request.args {
body.push_str(arg)?;
}
}
);
impl_canonical_mutation_body!(
proto_gen::node_service::LoadBucketMetadataRequest,
b"rustfs-load-bucket-metadata-request-v1\0",
|request, body| {
body.push_str(&request.bucket)?;
body.push_bool(request.scanner_maintenance_change);
}
);
impl_canonical_mutation_body!(
proto_gen::node_service::DeleteBucketMetadataRequest,
b"rustfs-delete-bucket-metadata-request-v1\0",
|request, body| {
body.push_str(&request.bucket)?;
}
);
impl_canonical_mutation_body!(
proto_gen::node_service::DeletePolicyRequest,
b"rustfs-delete-policy-request-v1\0",
|request, body| {
body.push_str(&request.policy_name)?;
}
);
impl_canonical_mutation_body!(
proto_gen::node_service::LoadPolicyRequest,
b"rustfs-load-policy-request-v1\0",
|request, body| {
body.push_str(&request.policy_name)?;
}
);
impl_canonical_mutation_body!(
proto_gen::node_service::LoadPolicyMappingRequest,
b"rustfs-load-policy-mapping-request-v1\0",
|request, body| {
body.push_str(&request.user_or_group)?;
body.push_u64(request.user_type);
body.push_bool(request.is_group);
}
);
impl_canonical_mutation_body!(
proto_gen::node_service::DeleteUserRequest,
b"rustfs-delete-user-request-v1\0",
|request, body| {
body.push_str(&request.access_key)?;
}
);
impl_canonical_mutation_body!(
proto_gen::node_service::DeleteServiceAccountRequest,
b"rustfs-delete-service-account-request-v1\0",
|request, body| {
body.push_str(&request.access_key)?;
}
);
impl_canonical_mutation_body!(
proto_gen::node_service::LoadUserRequest,
b"rustfs-load-user-request-v1\0",
|request, body| {
body.push_str(&request.access_key)?;
body.push_bool(request.temp);
}
);
impl_canonical_mutation_body!(
proto_gen::node_service::LoadServiceAccountRequest,
b"rustfs-load-service-account-request-v1\0",
|request, body| {
body.push_str(&request.access_key)?;
}
);
impl_canonical_mutation_body!(
proto_gen::node_service::LoadGroupRequest,
b"rustfs-load-group-request-v1\0",
|request, body| {
body.push_str(&request.group)?;
}
);
impl_canonical_mutation_body!(
proto_gen::node_service::ReloadSiteReplicationConfigRequest,
b"rustfs-reload-site-replication-config-request-v1\0"
);
impl_canonical_mutation_body!(proto_gen::node_service::ReloadPoolMetaRequest, b"rustfs-reload-pool-meta-request-v1\0");
impl_canonical_mutation_body!(
proto_gen::node_service::StopRebalanceRequest,
b"rustfs-stop-rebalance-request-v1\0",
|request, body| {
body.push_str(&request.expected_rebalance_id)?;
}
);
impl_canonical_mutation_body!(
proto_gen::node_service::LoadRebalanceMetaRequest,
b"rustfs-load-rebalance-meta-request-v1\0",
|request, body| {
body.push_bool(request.start_rebalance);
}
);
impl_canonical_mutation_body!(
proto_gen::node_service::StartDecommissionRequest,
b"rustfs-start-decommission-request-v1\0",
|request, body| {
body.push_count(request.pool_indices.len())?;
for pool_index in &request.pool_indices {
body.push_u32(*pool_index);
}
}
);
impl_canonical_mutation_body!(
proto_gen::node_service::CancelDecommissionRequest,
b"rustfs-cancel-decommission-request-v1\0",
|request, body| {
body.push_u32(request.pool_index);
}
);
impl_canonical_mutation_body!(
proto_gen::node_service::ClearDecommissionRequest,
b"rustfs-clear-decommission-request-v1\0",
|request, body| {
body.push_u32(request.pool_index);
}
);
impl_canonical_mutation_body!(
proto_gen::node_service::LoadTransitionTierConfigRequest,
b"rustfs-load-transition-tier-config-request-v1\0"
);
// Canonical request bodies for the mutating NodeService disk RPCs (backlog#1327 body-digest
// binding). Each covers every semantic wire field — including both the msgpack `_bin` payload and
// its JSON compatibility copy — so tampering with either encoding, or stripping `_bin` to force
@@ -575,6 +793,40 @@ pub fn canonical_delete_paths_request_body(
Ok(body.finish())
}
pub fn canonical_snapshot_lease_request_body(
request: &proto_gen::node_service::SnapshotLeaseRequest,
) -> Result<Vec<u8>, std::num::TryFromIntError> {
let mut body = CanonicalBodyBuilder::new(b"rustfs-snapshot-lease-request-v1\0");
body.push_str(&request.disk)?;
body.push_str(&request.volume)?;
body.push_str(&request.path)?;
body.push_u64(request.ttl_ms);
Ok(body.finish())
}
pub fn canonical_snapshot_lease_renew_request_body(
request: &proto_gen::node_service::SnapshotLeaseRenewRequest,
) -> Result<Vec<u8>, std::num::TryFromIntError> {
let mut body = CanonicalBodyBuilder::new(b"rustfs-snapshot-lease-renew-request-v1\0");
body.push_str(&request.disk)?;
body.push_str(&request.volume)?;
body.push_str(&request.path)?;
body.push_bytes(&request.token)?;
body.push_u64(request.ttl_ms);
Ok(body.finish())
}
pub fn canonical_snapshot_lease_release_request_body(
request: &proto_gen::node_service::SnapshotLeaseReleaseRequest,
) -> Result<Vec<u8>, std::num::TryFromIntError> {
let mut body = CanonicalBodyBuilder::new(b"rustfs-snapshot-lease-release-request-v1\0");
body.push_str(&request.disk)?;
body.push_str(&request.volume)?;
body.push_str(&request.path)?;
body.push_bytes(&request.token)?;
Ok(body.finish())
}
pub fn canonical_rename_file_request_body(
request: &proto_gen::node_service::RenameFileRequest,
) -> Result<Vec<u8>, std::num::TryFromIntError> {
@@ -662,7 +914,8 @@ mod disk_mutation_canonical_tests {
use super::proto_gen::node_service::{
DeletePathsRequest, DeleteRequest, DeleteVersionRequest, DeleteVersionsRequest, DeleteVolumeRequest, MakeVolumeRequest,
MakeVolumesRequest, PreparePartTransactionRequest, RenameDataRequest, RenameFileRequest, RenamePartRequest,
SettlePartTransactionRequest, UpdateMetadataRequest, WriteAllRequest, WriteMetadataRequest,
SettlePartTransactionRequest, SnapshotLeaseReleaseRequest, SnapshotLeaseRenewRequest, SnapshotLeaseRequest,
UpdateMetadataRequest, WriteAllRequest, WriteMetadataRequest,
};
use super::*;
@@ -1020,6 +1273,58 @@ mod disk_mutation_canonical_tests {
);
}
#[test]
fn snapshot_lease_canonical_bodies_bind_every_field() {
let acquire = SnapshotLeaseRequest {
disk: "d".into(),
volume: "v".into(),
path: "p".into(),
ttl_ms: 60_000,
};
let mut acquire_bodies = vec![canonical_snapshot_lease_request_body(&acquire).unwrap()];
for mutate in [
|r: &mut SnapshotLeaseRequest| r.disk = "d2".into(),
|r: &mut SnapshotLeaseRequest| r.volume = "v2".into(),
|r: &mut SnapshotLeaseRequest| r.path = "p2".into(),
|r: &mut SnapshotLeaseRequest| r.ttl_ms = 60_001,
] {
let mut request = acquire.clone();
mutate(&mut request);
acquire_bodies.push(canonical_snapshot_lease_request_body(&request).unwrap());
}
assert_all_distinct(&acquire_bodies);
let renew = SnapshotLeaseRenewRequest {
disk: "d".into(),
volume: "v".into(),
path: "p".into(),
token: vec![1; 16].into(),
ttl_ms: 60_000,
};
let mut changed_token = renew.clone();
changed_token.token = vec![2; 16].into();
let mut changed_ttl = renew.clone();
changed_ttl.ttl_ms += 1;
assert_all_distinct(&[
canonical_snapshot_lease_renew_request_body(&renew).unwrap(),
canonical_snapshot_lease_renew_request_body(&changed_token).unwrap(),
canonical_snapshot_lease_renew_request_body(&changed_ttl).unwrap(),
]);
let release = SnapshotLeaseReleaseRequest {
disk: "d".into(),
volume: "v".into(),
path: "p".into(),
token: vec![1; 16].into(),
};
let mut changed_release = release.clone();
changed_release.token = vec![2; 16].into();
assert_ne!(
canonical_snapshot_lease_release_request_body(&release).unwrap(),
canonical_snapshot_lease_release_request_body(&changed_release).unwrap()
);
}
#[test]
fn disk_mutation_canonical_domains_are_distinct_per_message() {
// The same field values must never authenticate one RPC's request as another's.
@@ -1045,6 +1350,154 @@ mod disk_mutation_canonical_tests {
}
}
#[cfg(test)]
mod non_disk_mutation_canonical_tests {
use super::*;
use proto_gen::node_service::*;
use std::collections::HashMap;
macro_rules! assert_fields_bound {
($request:ty, {$($field:ident: $value:expr),+ $(,)?}) => {{
let baseline = <$request>::default();
let expected = baseline.canonical_body().expect("baseline canonical body should encode");
$(
let mut variant = baseline.clone();
variant.$field = $value;
assert_ne!(
expected,
variant.canonical_body().expect("variant canonical body should encode"),
concat!(stringify!($request), " omitted field ", stringify!($field)),
);
)+
}};
}
fn signal_request(signal: Option<&str>, sub_system: Option<&str>, dry_run: Option<&str>) -> SignalServiceRequest {
let mut value = HashMap::new();
if let Some(signal) = signal {
value.insert(PEER_RESTSIGNAL.to_string(), signal.to_string());
}
if let Some(sub_system) = sub_system {
value.insert(PEER_RESTSUB_SYS.to_string(), sub_system.to_string());
}
if let Some(dry_run) = dry_run {
value.insert(PEER_RESTDRY_RUN.to_string(), dry_run.to_string());
}
SignalServiceRequest {
vars: Some(Mss { value }),
}
}
#[test]
fn signal_service_canonical_body_is_versioned_and_binds_every_semantic_field() {
let request = signal_request(Some("2"), Some("scanner"), Some("false"));
let baseline = request.canonical_body().expect("small signal request should encode");
assert!(baseline.starts_with(b"rustfs-signal-service-request-v1\0"));
for variant in [
signal_request(Some("1"), Some("scanner"), Some("false")),
signal_request(Some("2"), Some("heal"), Some("false")),
signal_request(Some("2"), Some("scanner"), Some("true")),
signal_request(None, Some("scanner"), Some("false")),
signal_request(Some("2"), None, Some("false")),
signal_request(Some("2"), Some("scanner"), None),
] {
assert_ne!(baseline, variant.canonical_body().expect("small signal request should encode"));
}
let mut ignored_field = request;
ignored_field
.vars
.as_mut()
.expect("signal vars should exist")
.value
.insert("future-field".to_string(), "ignored".to_string());
assert_eq!(
baseline,
ignored_field
.canonical_body()
.expect("unknown signal field should not affect the semantic body"),
);
}
#[test]
fn signal_service_canonical_body_distinguishes_missing_and_empty_fields() {
assert_ne!(
signal_request(None, Some("scanner"), Some("false"))
.canonical_body()
.expect("missing signal should encode"),
signal_request(Some(""), Some("scanner"), Some("false"))
.canonical_body()
.expect("empty signal should encode"),
);
}
#[test]
fn bucket_and_lock_canonical_bodies_bind_every_semantic_field() {
assert_fields_bound!(HealBucketRequest, { bucket: "bucket".into(), options: "opts".into() });
assert_fields_bound!(MakeBucketRequest, { name: "bucket".into(), options: "opts".into() });
assert_fields_bound!(DeleteBucketRequest, { bucket: "bucket".into(), options: "opts".into() });
assert_fields_bound!(GenerallyLockRequest, { args: "lock".into() });
assert_fields_bound!(BatchGenerallyLockRequest, { args: vec!["first".into(), "second".into()] });
let first = BatchGenerallyLockRequest {
args: vec!["first".into(), "second".into()],
};
let reversed = BatchGenerallyLockRequest {
args: vec!["second".into(), "first".into()],
};
assert_ne!(first.canonical_body().unwrap(), reversed.canonical_body().unwrap());
}
#[test]
fn metadata_and_iam_canonical_bodies_bind_every_semantic_field() {
assert_fields_bound!(LoadBucketMetadataRequest, {
bucket: "bucket".into(),
scanner_maintenance_change: true,
});
assert_fields_bound!(DeleteBucketMetadataRequest, { bucket: "bucket".into() });
assert_fields_bound!(DeletePolicyRequest, { policy_name: "policy".into() });
assert_fields_bound!(LoadPolicyRequest, { policy_name: "policy".into() });
assert_fields_bound!(LoadPolicyMappingRequest, {
user_or_group: "user".into(),
user_type: 1,
is_group: true,
});
assert_fields_bound!(DeleteUserRequest, { access_key: "user".into() });
assert_fields_bound!(DeleteServiceAccountRequest, { access_key: "service".into() });
assert_fields_bound!(LoadUserRequest, { access_key: "user".into(), temp: true });
assert_fields_bound!(LoadServiceAccountRequest, { access_key: "service".into() });
assert_fields_bound!(LoadGroupRequest, { group: "group".into() });
}
#[test]
fn control_plane_canonical_bodies_bind_every_semantic_field() {
assert_fields_bound!(StopRebalanceRequest, { expected_rebalance_id: "rebalance".into() });
assert_fields_bound!(LoadRebalanceMetaRequest, { start_rebalance: true });
assert_fields_bound!(StartDecommissionRequest, { pool_indices: vec![1, 2] });
assert_fields_bound!(CancelDecommissionRequest, { pool_index: 1 });
assert_fields_bound!(ClearDecommissionRequest, { pool_index: 1 });
let first = StartDecommissionRequest {
pool_indices: vec![1, 2],
};
let reversed = StartDecommissionRequest {
pool_indices: vec![2, 1],
};
assert_ne!(first.canonical_body().unwrap(), reversed.canonical_body().unwrap());
let empty_domains = [
ReloadSiteReplicationConfigRequest::default().canonical_body().unwrap(),
ReloadPoolMetaRequest::default().canonical_body().unwrap(),
LoadTransitionTierConfigRequest::default().canonical_body().unwrap(),
];
assert!(empty_domains.iter().all(|body| !body.is_empty()));
assert_ne!(empty_domains[0], empty_domains[1]);
assert_ne!(empty_domains[0], empty_domains[2]);
assert_ne!(empty_domains[1], empty_domains[2]);
}
}
#[cfg(test)]
mod scanner_activity_tests {
use super::{
+122 -85
View File
@@ -342,6 +342,40 @@ message DeletePathsResponse {
optional Error error = 2;
}
message SnapshotLeaseRequest {
string disk = 1;
string volume = 2;
string path = 3;
uint64 ttl_ms = 4;
}
message SnapshotLeaseRenewRequest {
string disk = 1;
string volume = 2;
string path = 3;
bytes token = 4;
uint64 ttl_ms = 5;
}
message SnapshotLeaseReleaseRequest {
string disk = 1;
string volume = 2;
string path = 3;
bytes token = 4;
}
message SnapshotLeaseResponse {
bool success = 1;
bytes token = 2;
uint32 protocol_version = 3;
optional Error error = 4;
}
message SnapshotLeaseMutationResponse {
bool success = 1;
optional Error error = 2;
}
message ReadMetadataRequest {
string disk = 1;
string volume = 2;
@@ -954,104 +988,107 @@ message GetLiveEventsResponse {
service NodeService {
/* -------------------------------meta service-------------------------- */
rpc Ping(PingRequest) returns (PingResponse) {};
rpc HealBucket(HealBucketRequest) returns (HealBucketResponse) {};
rpc ListBucket(ListBucketRequest) returns (ListBucketResponse) {};
rpc MakeBucket(MakeBucketRequest) returns (MakeBucketResponse) {};
rpc GetBucketInfo(GetBucketInfoRequest) returns (GetBucketInfoResponse) {};
rpc DeleteBucket(DeleteBucketRequest) returns (DeleteBucketResponse) {};
rpc Ping(PingRequest) returns (PingResponse) {}; // auth-policy: read-only
rpc HealBucket(HealBucketRequest) returns (HealBucketResponse) {}; // auth-policy: body-bound
rpc ListBucket(ListBucketRequest) returns (ListBucketResponse) {}; // auth-policy: read-only
rpc MakeBucket(MakeBucketRequest) returns (MakeBucketResponse) {}; // auth-policy: body-bound
rpc GetBucketInfo(GetBucketInfoRequest) returns (GetBucketInfoResponse) {}; // auth-policy: read-only
rpc DeleteBucket(DeleteBucketRequest) returns (DeleteBucketResponse) {}; // auth-policy: body-bound
/* -------------------------------disk service-------------------------- */
rpc ReadAll(ReadAllRequest) returns (ReadAllResponse) {};
rpc WriteAll(WriteAllRequest) returns (WriteAllResponse) {};
rpc Delete(DeleteRequest) returns (DeleteResponse) {};
rpc VerifyFile(VerifyFileRequest) returns (VerifyFileResponse) {};
rpc ReadParts(ReadPartsRequest) returns (ReadPartsResponse) {};
rpc CheckParts(CheckPartsRequest) returns (CheckPartsResponse) {};
rpc PreparePartTransaction(PreparePartTransactionRequest) returns (PreparePartTransactionResponse) {};
rpc RenamePart(RenamePartRequest) returns (RenamePartResponse) {};
rpc SettlePartTransaction(SettlePartTransactionRequest) returns (SettlePartTransactionResponse) {};
rpc RenameFile(RenameFileRequest) returns (RenameFileResponse) {};
rpc Write(WriteRequest) returns (WriteResponse) {};
rpc WriteStream(stream WriteRequest) returns (stream WriteResponse) {};
rpc ReadAll(ReadAllRequest) returns (ReadAllResponse) {}; // auth-policy: read-only
rpc WriteAll(WriteAllRequest) returns (WriteAllResponse) {}; // auth-policy: body-bound
rpc Delete(DeleteRequest) returns (DeleteResponse) {}; // auth-policy: body-bound
rpc AcquireSnapshotLease(SnapshotLeaseRequest) returns (SnapshotLeaseResponse) {}; // auth-policy: body-bound
rpc RenewSnapshotLease(SnapshotLeaseRenewRequest) returns (SnapshotLeaseResponse) {}; // auth-policy: body-bound
rpc ReleaseSnapshotLease(SnapshotLeaseReleaseRequest) returns (SnapshotLeaseMutationResponse) {}; // auth-policy: body-bound
rpc VerifyFile(VerifyFileRequest) returns (VerifyFileResponse) {}; // auth-policy: read-only
rpc ReadParts(ReadPartsRequest) returns (ReadPartsResponse) {}; // auth-policy: read-only
rpc CheckParts(CheckPartsRequest) returns (CheckPartsResponse) {}; // auth-policy: read-only
rpc PreparePartTransaction(PreparePartTransactionRequest) returns (PreparePartTransactionResponse) {}; // auth-policy: body-bound
rpc RenamePart(RenamePartRequest) returns (RenamePartResponse) {}; // auth-policy: body-bound
rpc SettlePartTransaction(SettlePartTransactionRequest) returns (SettlePartTransactionResponse) {}; // auth-policy: body-bound
rpc RenameFile(RenameFileRequest) returns (RenameFileResponse) {}; // auth-policy: body-bound
rpc Write(WriteRequest) returns (WriteResponse) {}; // auth-policy: unimplemented
rpc WriteStream(stream WriteRequest) returns (stream WriteResponse) {}; // auth-policy: streaming
// rpc Append(AppendRequest) returns (AppendResponse) {};
rpc ReadAt(stream ReadAtRequest) returns (stream ReadAtResponse) {};
rpc ListDir(ListDirRequest) returns (ListDirResponse) {};
rpc WalkDir(WalkDirRequest) returns (stream WalkDirResponse) {};
rpc RenameData(RenameDataRequest) returns (RenameDataResponse) {};
rpc MakeVolumes(MakeVolumesRequest) returns (MakeVolumesResponse) {};
rpc MakeVolume(MakeVolumeRequest) returns (MakeVolumeResponse) {};
rpc ListVolumes(ListVolumesRequest) returns (ListVolumesResponse) {};
rpc StatVolume(StatVolumeRequest) returns (StatVolumeResponse) {};
rpc DeletePaths(DeletePathsRequest) returns (DeletePathsResponse) {};
rpc UpdateMetadata(UpdateMetadataRequest) returns (UpdateMetadataResponse) {};
rpc ReadMetadata(ReadMetadataRequest) returns (ReadMetadataResponse) {};
rpc WriteMetadata(WriteMetadataRequest) returns (WriteMetadataResponse) {};
rpc ReadVersion(ReadVersionRequest) returns (ReadVersionResponse) {};
rpc BatchReadVersion(BatchReadVersionRequest) returns (BatchReadVersionResponse) {};
rpc ReadXL(ReadXLRequest) returns (ReadXLResponse) {};
rpc DeleteVersion(DeleteVersionRequest) returns (DeleteVersionResponse) {};
rpc DeleteVersions(DeleteVersionsRequest) returns (DeleteVersionsResponse) {};
rpc ReadMultiple(ReadMultipleRequest) returns (ReadMultipleResponse) {};
rpc DeleteVolume(DeleteVolumeRequest) returns (DeleteVolumeResponse) {};
rpc DiskInfo(DiskInfoRequest) returns (DiskInfoResponse) {};
rpc ReadAt(stream ReadAtRequest) returns (stream ReadAtResponse) {}; // auth-policy: streaming
rpc ListDir(ListDirRequest) returns (ListDirResponse) {}; // auth-policy: read-only
rpc WalkDir(WalkDirRequest) returns (stream WalkDirResponse) {}; // auth-policy: streaming
rpc RenameData(RenameDataRequest) returns (RenameDataResponse) {}; // auth-policy: body-bound
rpc MakeVolumes(MakeVolumesRequest) returns (MakeVolumesResponse) {}; // auth-policy: body-bound
rpc MakeVolume(MakeVolumeRequest) returns (MakeVolumeResponse) {}; // auth-policy: body-bound
rpc ListVolumes(ListVolumesRequest) returns (ListVolumesResponse) {}; // auth-policy: read-only
rpc StatVolume(StatVolumeRequest) returns (StatVolumeResponse) {}; // auth-policy: read-only
rpc DeletePaths(DeletePathsRequest) returns (DeletePathsResponse) {}; // auth-policy: body-bound
rpc UpdateMetadata(UpdateMetadataRequest) returns (UpdateMetadataResponse) {}; // auth-policy: body-bound
rpc ReadMetadata(ReadMetadataRequest) returns (ReadMetadataResponse) {}; // auth-policy: read-only
rpc WriteMetadata(WriteMetadataRequest) returns (WriteMetadataResponse) {}; // auth-policy: body-bound
rpc ReadVersion(ReadVersionRequest) returns (ReadVersionResponse) {}; // auth-policy: read-only
rpc BatchReadVersion(BatchReadVersionRequest) returns (BatchReadVersionResponse) {}; // auth-policy: read-only
rpc ReadXL(ReadXLRequest) returns (ReadXLResponse) {}; // auth-policy: read-only
rpc DeleteVersion(DeleteVersionRequest) returns (DeleteVersionResponse) {}; // auth-policy: body-bound
rpc DeleteVersions(DeleteVersionsRequest) returns (DeleteVersionsResponse) {}; // auth-policy: body-bound
rpc ReadMultiple(ReadMultipleRequest) returns (ReadMultipleResponse) {}; // auth-policy: read-only
rpc DeleteVolume(DeleteVolumeRequest) returns (DeleteVolumeResponse) {}; // auth-policy: body-bound
rpc DiskInfo(DiskInfoRequest) returns (DiskInfoResponse) {}; // auth-policy: read-only
/* -------------------------------lock service-------------------------- */
rpc Lock(GenerallyLockRequest) returns (GenerallyLockResponse) {};
rpc UnLock(GenerallyLockRequest) returns (GenerallyLockResponse) {};
rpc ForceUnLock(GenerallyLockRequest) returns (GenerallyLockResponse) {};
rpc Refresh(GenerallyLockRequest) returns (GenerallyLockResponse) {};
rpc LockBatch(BatchGenerallyLockRequest) returns (BatchGenerallyLockResponse) {};
rpc UnLockBatch(BatchGenerallyLockRequest) returns (BatchGenerallyLockResponse) {};
rpc Lock(GenerallyLockRequest) returns (GenerallyLockResponse) {}; // auth-policy: body-bound
rpc UnLock(GenerallyLockRequest) returns (GenerallyLockResponse) {}; // auth-policy: body-bound
rpc ForceUnLock(GenerallyLockRequest) returns (GenerallyLockResponse) {}; // auth-policy: body-bound
rpc Refresh(GenerallyLockRequest) returns (GenerallyLockResponse) {}; // auth-policy: body-bound
rpc LockBatch(BatchGenerallyLockRequest) returns (BatchGenerallyLockResponse) {}; // auth-policy: body-bound
rpc UnLockBatch(BatchGenerallyLockRequest) returns (BatchGenerallyLockResponse) {}; // auth-policy: body-bound
/* -------------------------------peer rest service-------------------------- */
rpc LocalStorageInfo(LocalStorageInfoRequest) returns (LocalStorageInfoResponse) {};
rpc ServerInfo(ServerInfoRequest) returns (ServerInfoResponse) {};
rpc GetCpus(GetCpusRequest) returns (GetCpusResponse) {};
rpc GetNetInfo(GetNetInfoRequest) returns (GetNetInfoResponse) {};
rpc GetPartitions(GetPartitionsRequest) returns (GetPartitionsResponse) {};
rpc GetOsInfo(GetOsInfoRequest) returns (GetOsInfoResponse) {};
rpc GetSELinuxInfo(GetSELinuxInfoRequest) returns (GetSELinuxInfoResponse) {};
rpc GetSysConfig(GetSysConfigRequest) returns (GetSysConfigResponse) {};
rpc GetSysErrors(GetSysErrorsRequest) returns (GetSysErrorsResponse) {};
rpc GetMemInfo(GetMemInfoRequest) returns (GetMemInfoResponse) {};
rpc GetMetrics(GetMetricsRequest) returns (GetMetricsResponse) {};
rpc GetProcInfo(GetProcInfoRequest) returns (GetProcInfoResponse) {};
rpc StartProfiling(StartProfilingRequest) returns (StartProfilingResponse) {};
rpc DownloadProfileData(DownloadProfileDataRequest) returns (DownloadProfileDataResponse) {};
rpc GetBucketStats(GetBucketStatsDataRequest) returns (GetBucketStatsDataResponse) {};
rpc GetSRMetrics(GetSRMetricsDataRequest) returns (GetSRMetricsDataResponse) {};
rpc GetAllBucketStats(GetAllBucketStatsRequest) returns (GetAllBucketStatsResponse) {};
rpc LoadBucketMetadata(LoadBucketMetadataRequest) returns (LoadBucketMetadataResponse) {};
rpc DeleteBucketMetadata(DeleteBucketMetadataRequest) returns (DeleteBucketMetadataResponse) {};
rpc DeletePolicy(DeletePolicyRequest) returns (DeletePolicyResponse) {};
rpc LoadPolicy(LoadPolicyRequest) returns (LoadPolicyResponse) {};
rpc LoadPolicyMapping(LoadPolicyMappingRequest) returns (LoadPolicyMappingResponse) {};
rpc DeleteUser(DeleteUserRequest) returns (DeleteUserResponse) {};
rpc DeleteServiceAccount(DeleteServiceAccountRequest) returns (DeleteServiceAccountResponse) {};
rpc LoadUser(LoadUserRequest) returns (LoadUserResponse) {};
rpc LoadServiceAccount(LoadServiceAccountRequest) returns (LoadServiceAccountResponse) {};
rpc LoadGroup(LoadGroupRequest) returns (LoadGroupResponse) {};
rpc ReloadSiteReplicationConfig(ReloadSiteReplicationConfigRequest) returns (ReloadSiteReplicationConfigResponse) {};
rpc LocalStorageInfo(LocalStorageInfoRequest) returns (LocalStorageInfoResponse) {}; // auth-policy: read-only
rpc ServerInfo(ServerInfoRequest) returns (ServerInfoResponse) {}; // auth-policy: read-only
rpc GetCpus(GetCpusRequest) returns (GetCpusResponse) {}; // auth-policy: read-only
rpc GetNetInfo(GetNetInfoRequest) returns (GetNetInfoResponse) {}; // auth-policy: read-only
rpc GetPartitions(GetPartitionsRequest) returns (GetPartitionsResponse) {}; // auth-policy: read-only
rpc GetOsInfo(GetOsInfoRequest) returns (GetOsInfoResponse) {}; // auth-policy: read-only
rpc GetSELinuxInfo(GetSELinuxInfoRequest) returns (GetSELinuxInfoResponse) {}; // auth-policy: read-only
rpc GetSysConfig(GetSysConfigRequest) returns (GetSysConfigResponse) {}; // auth-policy: read-only
rpc GetSysErrors(GetSysErrorsRequest) returns (GetSysErrorsResponse) {}; // auth-policy: read-only
rpc GetMemInfo(GetMemInfoRequest) returns (GetMemInfoResponse) {}; // auth-policy: read-only
rpc GetMetrics(GetMetricsRequest) returns (GetMetricsResponse) {}; // auth-policy: read-only
rpc GetProcInfo(GetProcInfoRequest) returns (GetProcInfoResponse) {}; // auth-policy: read-only
rpc StartProfiling(StartProfilingRequest) returns (StartProfilingResponse) {}; // auth-policy: unimplemented
rpc DownloadProfileData(DownloadProfileDataRequest) returns (DownloadProfileDataResponse) {}; // auth-policy: unimplemented
rpc GetBucketStats(GetBucketStatsDataRequest) returns (GetBucketStatsDataResponse) {}; // auth-policy: read-only
rpc GetSRMetrics(GetSRMetricsDataRequest) returns (GetSRMetricsDataResponse) {}; // auth-policy: unimplemented
rpc GetAllBucketStats(GetAllBucketStatsRequest) returns (GetAllBucketStatsResponse) {}; // auth-policy: unimplemented
rpc LoadBucketMetadata(LoadBucketMetadataRequest) returns (LoadBucketMetadataResponse) {}; // auth-policy: body-bound
rpc DeleteBucketMetadata(DeleteBucketMetadataRequest) returns (DeleteBucketMetadataResponse) {}; // auth-policy: body-bound
rpc DeletePolicy(DeletePolicyRequest) returns (DeletePolicyResponse) {}; // auth-policy: body-bound
rpc LoadPolicy(LoadPolicyRequest) returns (LoadPolicyResponse) {}; // auth-policy: body-bound
rpc LoadPolicyMapping(LoadPolicyMappingRequest) returns (LoadPolicyMappingResponse) {}; // auth-policy: body-bound
rpc DeleteUser(DeleteUserRequest) returns (DeleteUserResponse) {}; // auth-policy: body-bound
rpc DeleteServiceAccount(DeleteServiceAccountRequest) returns (DeleteServiceAccountResponse) {}; // auth-policy: body-bound
rpc LoadUser(LoadUserRequest) returns (LoadUserResponse) {}; // auth-policy: body-bound
rpc LoadServiceAccount(LoadServiceAccountRequest) returns (LoadServiceAccountResponse) {}; // auth-policy: body-bound
rpc LoadGroup(LoadGroupRequest) returns (LoadGroupResponse) {}; // auth-policy: body-bound
rpc ReloadSiteReplicationConfig(ReloadSiteReplicationConfigRequest) returns (ReloadSiteReplicationConfigResponse) {}; // auth-policy: body-bound
// rpc VerifyBinary() returns () {};
// rpc CommitBinary() returns () {};
rpc SignalService(SignalServiceRequest) returns (SignalServiceResponse) {};
rpc ScannerActivity(ScannerActivityRequest) returns (ScannerActivityResponse) {};
rpc BackgroundHealStatus(BackgroundHealStatusRequest) returns (BackgroundHealStatusResponse) {};
rpc GetMetacacheListing(GetMetacacheListingRequest) returns (GetMetacacheListingResponse) {};
rpc UpdateMetacacheListing(UpdateMetacacheListingRequest) returns (UpdateMetacacheListingResponse) {};
rpc ReloadPoolMeta(ReloadPoolMetaRequest) returns (ReloadPoolMetaResponse) {};
rpc StopRebalance(StopRebalanceRequest) returns (StopRebalanceResponse) {};
rpc LoadRebalanceMeta(LoadRebalanceMetaRequest) returns (LoadRebalanceMetaResponse) {};
rpc StartDecommission(StartDecommissionRequest) returns (StartDecommissionResponse) {};
rpc CancelDecommission(CancelDecommissionRequest) returns (CancelDecommissionResponse) {};
rpc ClearDecommission(ClearDecommissionRequest) returns (ClearDecommissionResponse) {};
rpc LoadTransitionTierConfig(LoadTransitionTierConfigRequest) returns (LoadTransitionTierConfigResponse) {};
rpc GetLiveEvents(GetLiveEventsRequest) returns (GetLiveEventsResponse) {};
rpc SignalService(SignalServiceRequest) returns (SignalServiceResponse) {}; // auth-policy: body-bound
rpc ScannerActivity(ScannerActivityRequest) returns (ScannerActivityResponse) {}; // auth-policy: body-bound
rpc BackgroundHealStatus(BackgroundHealStatusRequest) returns (BackgroundHealStatusResponse) {}; // auth-policy: read-only
rpc GetMetacacheListing(GetMetacacheListingRequest) returns (GetMetacacheListingResponse) {}; // auth-policy: unimplemented
rpc UpdateMetacacheListing(UpdateMetacacheListingRequest) returns (UpdateMetacacheListingResponse) {}; // auth-policy: unimplemented
rpc ReloadPoolMeta(ReloadPoolMetaRequest) returns (ReloadPoolMetaResponse) {}; // auth-policy: body-bound
rpc StopRebalance(StopRebalanceRequest) returns (StopRebalanceResponse) {}; // auth-policy: body-bound
rpc LoadRebalanceMeta(LoadRebalanceMetaRequest) returns (LoadRebalanceMetaResponse) {}; // auth-policy: body-bound
rpc StartDecommission(StartDecommissionRequest) returns (StartDecommissionResponse) {}; // auth-policy: body-bound
rpc CancelDecommission(CancelDecommissionRequest) returns (CancelDecommissionResponse) {}; // auth-policy: body-bound
rpc ClearDecommission(ClearDecommissionRequest) returns (ClearDecommissionResponse) {}; // auth-policy: body-bound
rpc LoadTransitionTierConfig(LoadTransitionTierConfigRequest) returns (LoadTransitionTierConfigResponse) {}; // auth-policy: body-bound
rpc GetLiveEvents(GetLiveEventsRequest) returns (GetLiveEventsResponse) {}; // auth-policy: read-only
}
service HealControlService {
+51 -1
View File
@@ -698,7 +698,7 @@ impl DataUsageCache {
let mut visited = HashSet::new();
visited.insert(hash_path(path).key());
let mut flat = self.flatten_with_guard(root, &mut visited, 0);
if flat.replication_stats.as_ref().is_some_and(|stats| stats.empty()) {
if flat.replication_stats.as_ref().is_some_and(|stats| stats.is_empty()) {
flat.replication_stats = None;
}
Some(flat)
@@ -1574,6 +1574,7 @@ mod tests {
use super::*;
use crate::storage_api::scanner_io::{HTTPRangeSpec, ObjectIO};
use crate::{ScannerGetObjectReader, ScannerPutObjReader};
use rustfs_data_usage::{ReplicationAllStats, ReplicationStats};
use serde_json::Value;
use std::io::Cursor;
use std::pin::Pin;
@@ -2767,6 +2768,55 @@ mod tests {
assert!(flat.children.is_empty());
}
#[test]
fn size_recursive_prunes_empty_and_preserves_threshold_replication_stats() {
let root = hash_path("bucket");
let child = hash_path("bucket/child");
let mut cache = DataUsageCache::default();
cache.replace_hashed(&root, &None, &DataUsageEntry::default());
cache.replace_hashed(
&child,
&Some(root.clone()),
&DataUsageEntry {
replication_stats: Some(ReplicationAllStats::default()),
..Default::default()
},
);
assert!(
cache
.size_recursive("bucket")
.expect("scanner bucket usage should flatten")
.replication_stats
.is_none()
);
cache.replace_hashed(
&child,
&Some(root.clone()),
&DataUsageEntry {
replication_stats: Some(ReplicationAllStats {
targets: HashMap::from([(
"arn:test:threshold".to_string(),
ReplicationStats {
after_threshold_count: 1,
..Default::default()
},
)]),
..Default::default()
}),
..Default::default()
},
);
let flattened = cache.size_recursive("bucket").expect("scanner bucket usage should flatten");
let replication = flattened
.replication_stats
.expect("threshold-only replication stats must survive pruning");
assert_eq!(replication.targets["arn:test:threshold"].after_threshold_count, 1);
}
#[test]
fn checked_flatten_rejects_dangling_child() {
let root_key = hash_path("bucket").key();
+1 -1
View File
@@ -125,7 +125,7 @@ tokio = { workspace = true, features = ["rt-multi-thread", "macros", "net", "sig
tokio-rustls = { workspace = true, default-features = false, features = ["logging", "tls12", "aws-lc-rs"] }
aws-sdk-s3 = { workspace = true, default-features = false, features = ["sigv4a", "default-https-client", "rt-tokio"] }
tokio-stream.workspace = true
tokio-util = { workspace = true, features = ["io", "compat"] }
tokio-util = { workspace = true, features = ["io", "compat", "time"] }
tonic = { workspace = true, features = ["gzip", "deflate"] }
tower = { workspace = true, features = ["timeout"] }
tower-http = { workspace = true, features = ["trace", "compression-full", "cors", "catch-panic", "timeout", "limit", "request-id", "add-extension"] }
+125 -1
View File
@@ -60,6 +60,7 @@ use uuid::Uuid;
static GLOBAL_ENV: OnceLock<(Vec<PathBuf>, Arc<ECStore>)> = OnceLock::new();
static INIT: Once = Once::new();
const TRANSITION_WAIT_TIMEOUT: Duration = Duration::from_secs(15);
const RESTORE_SUSPENDED_WAIT_TIMEOUT: Duration = Duration::from_secs(15);
const ENV_GET_CODEC_STREAMING_ENABLE: &str = "RUSTFS_GET_CODEC_STREAMING_ENABLE";
const ENV_GET_CODEC_STREAMING_ROLLOUT: &str = "RUSTFS_GET_CODEC_STREAMING_ROLLOUT";
const ENV_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED: &str = "RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED";
@@ -161,6 +162,23 @@ async fn create_test_bucket(ecstore: &Arc<ECStore>, bucket_name: &str) {
.expect("Failed to create test bucket");
}
async fn suspend_test_bucket(bucket: &str) {
DefaultBucketUsecase::from_global()
.execute_put_bucket_versioning(build_request(
PutBucketVersioningInput::builder()
.bucket(bucket.to_string())
.versioning_configuration(VersioningConfiguration {
status: Some(BucketVersioningStatus::from_static(BucketVersioningStatus::SUSPENDED)),
..Default::default()
})
.build()
.expect("suspended versioning request should build"),
Method::PUT,
))
.await
.expect("bucket versioning should be suspended");
}
async fn upload_test_object(ecstore: &Arc<ECStore>, bucket: &str, object: &str, data: &[u8]) -> ObjectInfo {
let mut reader = PutObjReader::from_vec(data.to_vec());
(**ecstore)
@@ -338,6 +356,45 @@ async fn wait_for_transition(ecstore: &Arc<ECStore>, bucket: &str, object: &str,
}
}
async fn wait_for_restore_completion(
ecstore: &Arc<ECStore>,
backend: &MockWarmBackend,
bucket: &str,
object: &str,
timeout: Duration,
) -> Result<ObjectInfo, String> {
let deadline = tokio::time::Instant::now() + timeout;
let mut last_state = None;
loop {
if tokio::time::Instant::now() >= deadline {
let tier_gets = backend.get_count().await;
let op_log = backend.op_log().await;
return Err(format!(
"restore copy-back should complete within {timeout:?}; tier_gets={tier_gets}, op_log={op_log:?}; last observed state: {}",
last_state.unwrap_or_else(|| "no object info observed".to_string())
));
}
match (**ecstore).get_object_info(bucket, object, &ObjectOptions::default()).await {
Ok(info) => {
if !info.restore_ongoing && info.restore_expires.is_some() {
return Ok(info);
}
last_state = Some(format!(
"restore_ongoing={}, restore_expires={:?}, transitioned_status={}",
info.restore_ongoing, info.restore_expires, info.transitioned_object.status
));
}
Err(err) => {
last_state = Some(format!("get_object_info failed: {err}"));
}
}
tokio::time::sleep(Duration::from_millis(500)).await;
}
}
// SAFETY: this helper is used only by `#[serial]` tests and runs under the single-threaded Tokio
// runtime (`worker_threads = 1`), so no concurrent test can mutate process environment during the
// `env::set_var` / `env::remove_var` window.
@@ -2084,9 +2141,9 @@ async fn restore_object_usecase_reports_ongoing_conflict() {
create_test_bucket(&ecstore, bucket.as_str()).await;
let uploaded = upload_test_object(&ecstore, bucket.as_str(), object, &payload).await;
assert!(uploaded.version_id.is_none(), "fixture must create the null version");
let _ = transition_uploaded_object_directly(&ecstore, bucket.as_str(), object, &tier_name, &uploaded).await;
backend.clear_op_log().await;
let get_barrier = backend.arm_get_barrier().await;
let restore_request = || RestoreRequest {
@@ -2138,6 +2195,73 @@ async fn restore_object_usecase_reports_ongoing_conflict() {
get_barrier.release();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#4879"]
async fn restore_object_usecase_completes_suspended_null_version_in_place() {
let (_disk_paths, ecstore) = setup_test_env().await;
let usecase = DefaultObjectUsecase::from_global();
let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
let backend = register_mock_tier(&tier_name).await;
let bucket = format!("test-api-restore-suspended-{}", &Uuid::new_v4().simple().to_string()[..8]);
let object = "test/restore/suspended-null.bin";
let payload: Vec<u8> = (0..128 * 1024).map(|i| (i % 251) as u8).collect();
create_test_bucket(&ecstore, bucket.as_str()).await;
let uploaded = upload_test_object(&ecstore, bucket.as_str(), object, &payload).await;
assert!(uploaded.version_id.is_none(), "fixture must create the null version");
let _ = transition_uploaded_object_directly(&ecstore, bucket.as_str(), object, &tier_name, &uploaded).await;
suspend_test_bucket(bucket.as_str()).await;
backend.clear_op_log().await;
let tier_gets_before_restore = backend.get_count().await;
let get_barrier = backend.arm_get_barrier().await;
Box::pin(
usecase.execute_restore_object(build_request(
RestoreObjectInput::builder()
.bucket(bucket.clone())
.key(object.to_string())
.restore_request(Some(RestoreRequest {
days: Some(1),
description: None,
glacier_job_parameters: None,
output_location: None,
select_parameters: None,
tier: None,
type_: None,
}))
.build()
.expect("restore request should build"),
Method::POST,
)),
)
.await
.expect("suspended null-version restore should be accepted");
get_barrier.wait_until_paused().await;
get_barrier.release();
let completed = wait_for_restore_completion(&ecstore, &backend, bucket.as_str(), object, RESTORE_SUSPENDED_WAIT_TIMEOUT)
.await
.unwrap_or_else(|err| panic!("{err}"));
assert!(!completed.restore_ongoing, "the original null version must complete in place");
assert!(completed.restore_expires.is_some(), "completed restore must carry an expiry");
assert!(
completed.version_id.is_none() || completed.version_id.is_some_and(|version_id| version_id.is_nil()),
"suspended restore must remain on the null version"
);
assert_eq!(
live_object_version_count(&ecstore, bucket.as_str(), object).await,
1,
"suspended restore must not create a UUID version"
);
assert_eq!(
backend.get_count().await - tier_gets_before_restore,
1,
"suspended restore copy-back must fetch the tier exactly once"
);
}
/// backlog#1304: the restore-accept compare-and-set itself, under real
/// concurrency. Two POST ?restore for the same transitioned object race each
/// other from separate tasks; the accept guard must let exactly one through —
+8 -1
View File
@@ -7783,7 +7783,12 @@ impl DefaultObjectUsecase {
let bucket_clone = bucket.clone();
let object_clone = object.clone();
let rreq_clone = rreq.clone();
let version_id_clone = obj_info_.version_id.map(|v| v.to_string());
let version_id_clone = obj_info_
.version_id
.map(|v| v.to_string())
.or_else(|| (opts.versioned || opts.version_suspended).then(|| Uuid::nil().to_string()));
let versioned = opts.versioned;
let version_suspended = opts.version_suspended;
let mut restore_operation_metadata = HashMap::new();
if let Some(id) = restore_operation_id {
insert_str(&mut restore_operation_metadata, SUFFIX_RESTORE_OPERATION_ID, id.to_string());
@@ -7797,6 +7802,8 @@ impl DefaultObjectUsecase {
..Default::default()
},
version_id: version_id_clone,
versioned,
version_suspended,
user_defined: restore_operation_metadata,
..Default::default()
};
+3 -3
View File
@@ -69,9 +69,9 @@ pub(crate) use storage_api::{
get_lock_acquire_timeout, get_public_access_block_config, head_prefix_consumer, helper_consumer, init_background_replication,
init_bucket_metadata_sys, init_ecstore_config, init_local_disks_with_instance_ctx, init_lock_clients,
is_all_buckets_not_found, is_err_bucket_not_found, is_err_object_not_found, is_err_version_not_found, is_valid_storage_class,
load_bucket_metadata, options_consumer, prewarm_local_disk_id_map_with_instance_ctx, read_config, record_replication_proxy,
rpc_consumer, runtime_sources_consumer, s3_api_consumer, serialize, set_bucket_metadata, table_catalog_path_hash,
to_s3s_etag, topology_snapshot_from_endpoint_pools_with_capabilities, try_migrate_bucket_metadata, try_migrate_iam_config,
options_consumer, prewarm_local_disk_id_map_with_instance_ctx, read_config, record_replication_proxy, rpc_consumer,
runtime_sources_consumer, s3_api_consumer, serialize, table_catalog_path_hash, to_s3s_etag,
topology_snapshot_from_endpoint_pools_with_capabilities, try_migrate_bucket_metadata, try_migrate_iam_config,
try_migrate_server_config, update_bucket_metadata_config, verify_rpc_signature, wrap_reader,
};
+336 -35
View File
@@ -28,7 +28,9 @@ use crate::storage::storage_api::rpc_consumer::node_service::{
reload_transition_tier_config,
};
use crate::storage::storage_api::runtime_sources_consumer::{EndpointServerPools, runtime_sources};
use crate::storage::storage_api::{sign_tonic_rpc_response_proof, verify_tonic_canonical_body_digest};
use crate::storage::storage_api::{
sign_tonic_rpc_response_proof, verify_tonic_canonical_body_digest, verify_tonic_mutation_body_digest,
};
use bytes::Bytes;
use futures::Stream;
use futures_util::future::join_all;
@@ -40,6 +42,7 @@ use rustfs_filemeta::MetacacheReader;
use rustfs_iam::store::UserType;
use rustfs_lock::LockClient;
use rustfs_protos::{
CanonicalMutationBody,
models::{PingBody, PingBodyBuilder},
proto_gen::node_service::{node_service_server::NodeService as Node, *},
};
@@ -87,6 +90,15 @@ fn signal_service_response(success: bool, error_info: Option<String>) -> Respons
})
}
fn verify_node_mutation_body<T: CanonicalMutationBody>(request: &Request<T>, operation: &'static str) -> Result<(), Status> {
let canonical_body = request
.get_ref()
.canonical_body()
.map_err(|_| Status::invalid_argument(format!("{operation} request length cannot be represented")))?;
verify_tonic_mutation_body_digest(request, &canonical_body)
.map_err(|err| Status::permission_denied(format!("{operation} authentication failed: {err}")))
}
fn supports_dynamic_config_rpc(sub_system: &str) -> bool {
NOTIFY_SUB_SYSTEMS.contains(&sub_system)
|| matches!(
@@ -343,6 +355,7 @@ mod metrics;
pub struct NodeService {
local_peer: LocalPeerS3Client,
context: Option<Arc<runtime_sources::AppContext>>,
snapshot_lease_expiry: disk::SnapshotLeaseExpiryScheduler,
}
impl std::fmt::Debug for NodeService {
@@ -361,7 +374,11 @@ pub fn make_server() -> NodeService {
pub fn make_server_for_context(context: Option<Arc<runtime_sources::AppContext>>) -> NodeService {
let local_peer = LocalPeerS3Client::new(None, None);
NodeService { local_peer, context }
NodeService {
local_peer,
context,
snapshot_lease_expiry: disk::SnapshotLeaseExpiryScheduler::new(),
}
}
#[derive(Clone, Debug, Default)]
@@ -885,6 +902,7 @@ impl Node for NodeService {
}
async fn heal_bucket(&self, request: Request<HealBucketRequest>) -> Result<Response<HealBucketResponse>, Status> {
verify_node_mutation_body(&request, "heal bucket")?;
self.handle_heal_bucket(request).await
}
@@ -893,6 +911,7 @@ impl Node for NodeService {
}
async fn make_bucket(&self, request: Request<MakeBucketRequest>) -> Result<Response<MakeBucketResponse>, Status> {
verify_node_mutation_body(&request, "make bucket")?;
self.handle_make_bucket(request).await
}
@@ -901,6 +920,7 @@ impl Node for NodeService {
}
async fn delete_bucket(&self, request: Request<DeleteBucketRequest>) -> Result<Response<DeleteBucketResponse>, Status> {
verify_node_mutation_body(&request, "delete bucket")?;
self.handle_delete_bucket(request).await
}
@@ -1124,6 +1144,24 @@ impl Node for NodeService {
async fn delete_paths(&self, request: Request<DeletePathsRequest>) -> Result<Response<DeletePathsResponse>, Status> {
self.handle_delete_paths(request).await
}
async fn acquire_snapshot_lease(
&self,
request: Request<SnapshotLeaseRequest>,
) -> Result<Response<SnapshotLeaseResponse>, Status> {
self.handle_acquire_snapshot_lease(request).await
}
async fn renew_snapshot_lease(
&self,
request: Request<SnapshotLeaseRenewRequest>,
) -> Result<Response<SnapshotLeaseResponse>, Status> {
self.handle_renew_snapshot_lease(request).await
}
async fn release_snapshot_lease(
&self,
request: Request<SnapshotLeaseReleaseRequest>,
) -> Result<Response<SnapshotLeaseMutationResponse>, Status> {
self.handle_release_snapshot_lease(request).await
}
async fn read_metadata(&self, request: Request<ReadMetadataRequest>) -> Result<Response<ReadMetadataResponse>, Status> {
self.handle_read_metadata(request).await
}
@@ -1172,18 +1210,22 @@ impl Node for NodeService {
}
async fn lock(&self, request: Request<GenerallyLockRequest>) -> Result<Response<GenerallyLockResponse>, Status> {
verify_node_mutation_body(&request, "lock")?;
self.handle_lock(request).await
}
async fn un_lock(&self, request: Request<GenerallyLockRequest>) -> Result<Response<GenerallyLockResponse>, Status> {
verify_node_mutation_body(&request, "unlock")?;
self.handle_un_lock(request).await
}
async fn force_un_lock(&self, request: Request<GenerallyLockRequest>) -> Result<Response<GenerallyLockResponse>, Status> {
verify_node_mutation_body(&request, "force unlock")?;
self.handle_force_un_lock(request).await
}
async fn refresh(&self, request: Request<GenerallyLockRequest>) -> Result<Response<GenerallyLockResponse>, Status> {
verify_node_mutation_body(&request, "refresh lock")?;
self.handle_refresh(request).await
}
@@ -1191,6 +1233,7 @@ impl Node for NodeService {
&self,
request: Request<BatchGenerallyLockRequest>,
) -> Result<Response<BatchGenerallyLockResponse>, Status> {
verify_node_mutation_body(&request, "lock batch")?;
self.handle_lock_batch(request).await
}
@@ -1198,6 +1241,7 @@ impl Node for NodeService {
&self,
request: Request<BatchGenerallyLockRequest>,
) -> Result<Response<BatchGenerallyLockResponse>, Status> {
verify_node_mutation_body(&request, "unlock batch")?;
self.handle_un_lock_batch(request).await
}
@@ -1317,6 +1361,7 @@ impl Node for NodeService {
&self,
request: Request<LoadBucketMetadataRequest>,
) -> Result<Response<LoadBucketMetadataResponse>, Status> {
verify_node_mutation_body(&request, "load bucket metadata")?;
self.handle_load_bucket_metadata(request).await
}
@@ -1324,10 +1369,12 @@ impl Node for NodeService {
&self,
request: Request<DeleteBucketMetadataRequest>,
) -> Result<Response<DeleteBucketMetadataResponse>, Status> {
verify_node_mutation_body(&request, "delete bucket metadata")?;
self.handle_delete_bucket_metadata(request).await
}
async fn delete_policy(&self, request: Request<DeletePolicyRequest>) -> Result<Response<DeletePolicyResponse>, Status> {
verify_node_mutation_body(&request, "delete policy")?;
let request = request.into_inner();
let policy = request.policy_name;
if policy.is_empty() {
@@ -1358,6 +1405,7 @@ impl Node for NodeService {
}
async fn load_policy(&self, request: Request<LoadPolicyRequest>) -> Result<Response<LoadPolicyResponse>, Status> {
verify_node_mutation_body(&request, "load policy")?;
let request = request.into_inner();
let policy = request.policy_name;
if policy.is_empty() {
@@ -1390,6 +1438,7 @@ impl Node for NodeService {
&self,
request: Request<LoadPolicyMappingRequest>,
) -> Result<Response<LoadPolicyMappingResponse>, Status> {
verify_node_mutation_body(&request, "load policy mapping")?;
let request = request.into_inner();
let user_or_group = request.user_or_group;
if user_or_group.is_empty() {
@@ -1425,6 +1474,7 @@ impl Node for NodeService {
}
async fn delete_user(&self, request: Request<DeleteUserRequest>) -> Result<Response<DeleteUserResponse>, Status> {
verify_node_mutation_body(&request, "delete user")?;
let request = request.into_inner();
let access_key = request.access_key;
if access_key.is_empty() {
@@ -1457,6 +1507,7 @@ impl Node for NodeService {
&self,
request: Request<DeleteServiceAccountRequest>,
) -> Result<Response<DeleteServiceAccountResponse>, Status> {
verify_node_mutation_body(&request, "delete service account")?;
let request = request.into_inner();
let access_key = request.access_key;
if access_key.is_empty() {
@@ -1492,6 +1543,7 @@ impl Node for NodeService {
}
async fn load_user(&self, request: Request<LoadUserRequest>) -> Result<Response<LoadUserResponse>, Status> {
verify_node_mutation_body(&request, "load user")?;
let request = request.into_inner();
let access_key = request.access_key;
let temp = request.temp;
@@ -1529,6 +1581,7 @@ impl Node for NodeService {
&self,
request: Request<LoadServiceAccountRequest>,
) -> Result<Response<LoadServiceAccountResponse>, Status> {
verify_node_mutation_body(&request, "load service account")?;
let request = request.into_inner();
let access_key = request.access_key;
if access_key.is_empty() {
@@ -1560,6 +1613,7 @@ impl Node for NodeService {
}
async fn load_group(&self, request: Request<LoadGroupRequest>) -> Result<Response<LoadGroupResponse>, Status> {
verify_node_mutation_body(&request, "load group")?;
let request = request.into_inner();
let group = request.group;
if group.is_empty() {
@@ -1591,8 +1645,9 @@ impl Node for NodeService {
async fn reload_site_replication_config(
&self,
_request: Request<ReloadSiteReplicationConfigRequest>,
request: Request<ReloadSiteReplicationConfigRequest>,
) -> Result<Response<ReloadSiteReplicationConfigResponse>, Status> {
verify_node_mutation_body(&request, "reload site replication config")?;
let Some(_store) = self.resolve_object_store() else {
return Ok(Response::new(ReloadSiteReplicationConfigResponse {
success: false,
@@ -1612,6 +1667,7 @@ impl Node for NodeService {
}
async fn signal_service(&self, request: Request<SignalServiceRequest>) -> Result<Response<SignalServiceResponse>, Status> {
verify_node_mutation_body(&request, "signal service")?;
let request = request.into_inner();
let vars = match request.vars {
Some(vars) => vars.value,
@@ -1812,8 +1868,9 @@ impl Node for NodeService {
async fn reload_pool_meta(
&self,
_request: Request<ReloadPoolMetaRequest>,
request: Request<ReloadPoolMetaRequest>,
) -> Result<Response<ReloadPoolMetaResponse>, Status> {
verify_node_mutation_body(&request, "reload pool metadata")?;
let Some(store) = self.resolve_object_store() else {
return Ok(Response::new(ReloadPoolMetaResponse {
success: false,
@@ -1839,6 +1896,7 @@ impl Node for NodeService {
}
async fn stop_rebalance(&self, request: Request<StopRebalanceRequest>) -> Result<Response<StopRebalanceResponse>, Status> {
verify_node_mutation_body(&request, "stop rebalance")?;
let Some(store) = self.resolve_object_store() else {
return Ok(Response::new(StopRebalanceResponse {
success: false,
@@ -1859,6 +1917,7 @@ impl Node for NodeService {
&self,
request: Request<LoadRebalanceMetaRequest>,
) -> Result<Response<LoadRebalanceMetaResponse>, Status> {
verify_node_mutation_body(&request, "load rebalance metadata")?;
let LoadRebalanceMetaRequest { start_rebalance } = request.into_inner();
let Some(store) = self.resolve_object_store() else {
log_load_rebalance_meta_rejected!("server_not_initialized", start_rebalance);
@@ -1904,6 +1963,7 @@ impl Node for NodeService {
&self,
request: Request<StartDecommissionRequest>,
) -> Result<Response<StartDecommissionResponse>, Status> {
verify_node_mutation_body(&request, "start decommission")?;
let Some(store) = runtime_sources::current_object_store_handle() else {
return Ok(Response::new(StartDecommissionResponse {
success: false,
@@ -1935,6 +1995,7 @@ impl Node for NodeService {
&self,
request: Request<CancelDecommissionRequest>,
) -> Result<Response<CancelDecommissionResponse>, Status> {
verify_node_mutation_body(&request, "cancel decommission")?;
let Some(store) = runtime_sources::current_object_store_handle() else {
return Ok(Response::new(CancelDecommissionResponse {
success: false,
@@ -1967,6 +2028,7 @@ impl Node for NodeService {
&self,
request: Request<ClearDecommissionRequest>,
) -> Result<Response<ClearDecommissionResponse>, Status> {
verify_node_mutation_body(&request, "clear decommission")?;
let Some(store) = runtime_sources::current_object_store_handle() else {
return Ok(Response::new(ClearDecommissionResponse {
success: false,
@@ -1997,8 +2059,9 @@ impl Node for NodeService {
async fn load_transition_tier_config(
&self,
_request: Request<LoadTransitionTierConfigRequest>,
request: Request<LoadTransitionTierConfigRequest>,
) -> Result<Response<LoadTransitionTierConfigResponse>, Status> {
verify_node_mutation_body(&request, "load transition tier config")?;
let Some(store) = self.resolve_object_store() else {
return Ok(Response::new(LoadTransitionTierConfigResponse {
success: false,
@@ -2049,21 +2112,24 @@ mod tests {
sys::NewServiceAccountOpts,
};
use rustfs_kms::KmsServiceManager;
use rustfs_protos::CanonicalMutationBody as _;
use rustfs_protos::models::PingBodyBuilder;
use rustfs_protos::proto_gen::node_service::{
BackgroundHealStatusRequest, CheckPartsRequest, DeleteBucketMetadataRequest, DeleteBucketRequest, DeletePathsRequest,
DeletePolicyRequest, DeleteRequest, DeleteServiceAccountRequest, DeleteUserRequest, DeleteVersionRequest,
DeleteVersionsRequest, DeleteVolumeRequest, DiskInfoRequest, DownloadProfileDataRequest, GenerallyLockRequest,
GetAllBucketStatsRequest, GetBucketInfoRequest, GetBucketStatsDataRequest, GetCpusRequest, GetMemInfoRequest,
GetMetacacheListingRequest, GetMetricsRequest, GetNetInfoRequest, GetOsInfoRequest, GetPartitionsRequest,
GetProcInfoRequest, GetSeLinuxInfoRequest, GetSrMetricsDataRequest, GetSysConfigRequest, GetSysErrorsRequest,
HealBucketRequest, HealControlRequest, ListBucketRequest, ListDirRequest, ListVolumesRequest, LoadBucketMetadataRequest,
LoadGroupRequest, LoadPolicyMappingRequest, LoadPolicyRequest, LoadRebalanceMetaRequest, LoadServiceAccountRequest,
BackgroundHealStatusRequest, BatchGenerallyLockRequest, CancelDecommissionRequest, CheckPartsRequest,
ClearDecommissionRequest, DeleteBucketMetadataRequest, DeleteBucketRequest, DeletePathsRequest, DeletePolicyRequest,
DeleteRequest, DeleteServiceAccountRequest, DeleteUserRequest, DeleteVersionRequest, DeleteVersionsRequest,
DeleteVolumeRequest, DiskInfoRequest, DownloadProfileDataRequest, GenerallyLockRequest, GetAllBucketStatsRequest,
GetBucketInfoRequest, GetBucketStatsDataRequest, GetCpusRequest, GetMemInfoRequest, GetMetacacheListingRequest,
GetMetricsRequest, GetNetInfoRequest, GetOsInfoRequest, GetPartitionsRequest, GetProcInfoRequest, GetSeLinuxInfoRequest,
GetSrMetricsDataRequest, GetSysConfigRequest, GetSysErrorsRequest, HealBucketRequest, HealControlRequest,
ListBucketRequest, ListDirRequest, ListVolumesRequest, LoadBucketMetadataRequest, LoadGroupRequest,
LoadPolicyMappingRequest, LoadPolicyRequest, LoadRebalanceMetaRequest, LoadServiceAccountRequest,
LoadTransitionTierConfigRequest, LoadUserRequest, LocalStorageInfoRequest, MakeBucketRequest, MakeVolumeRequest,
MakeVolumesRequest, Mss, PingRequest, PreparePartTransactionRequest, ReadAllRequest, ReadAtRequest, ReadMultipleRequest,
ReadVersionRequest, ReadXlRequest, ReloadPoolMetaRequest, ReloadSiteReplicationConfigRequest, RenameDataRequest,
RenameFileRequest, RenamePartRequest, ScannerActivityRequest, ServerInfoRequest, SettlePartTransactionRequest,
SignalServiceRequest, StartProfilingRequest, StatVolumeRequest, StopRebalanceRequest, TierMutationPeerState,
SignalServiceRequest, SnapshotLeaseReleaseRequest, SnapshotLeaseRenewRequest, SnapshotLeaseRequest,
StartDecommissionRequest, StartProfilingRequest, StatVolumeRequest, StopRebalanceRequest, TierMutationPeerState,
TierMutationPrepareRequest, UpdateMetacacheListingRequest, UpdateMetadataRequest, VerifyFileRequest, WriteAllRequest,
WriteMetadataRequest, WriteRequest,
heal_control_service_client::HealControlServiceClient,
@@ -2072,13 +2138,79 @@ mod tests {
node_service_server::NodeServiceServer,
tier_mutation_control_service_server::TierMutationControlService as _,
};
use std::{collections::HashMap, sync::Arc};
use std::{
collections::{HashMap, HashSet},
sync::Arc,
};
use time::OffsetDateTime;
use tokio::net::TcpListener;
use tokio::time::Duration;
use tokio_stream::wrappers::TcpListenerStream;
use tonic::{Request, Response, Status};
const DISK_MUTATION_RPC_METHODS: [&str; 18] = [
"renamedata",
"deleteversion",
"deleteversions",
"writemetadata",
"updatemetadata",
"writeall",
"delete",
"deletepaths",
"renamefile",
"renamepart",
"prepareparttransaction",
"settleparttransaction",
"deletevolume",
"makevolume",
"makevolumes",
"acquiresnapshotlease",
"renewsnapshotlease",
"releasesnapshotlease",
];
fn normalized_rpc_method(method: &str) -> String {
method.replace('_', "").to_ascii_lowercase()
}
fn node_service_auth_policies() -> HashMap<String, &'static str> {
const POLICY_MARKER: &str = "// auth-policy: ";
let schema = include_str!("../../../../crates/protos/src/node.proto");
let service = schema
.split_once("service NodeService {")
.expect("NodeService must exist in node.proto")
.1
.split_once("\n}")
.expect("NodeService must have a closing brace")
.0;
let mut policies = HashMap::new();
for declaration in service.lines().filter_map(|line| line.trim().strip_prefix("rpc ")) {
let (rpc, policy) = declaration
.split_once(POLICY_MARKER)
.expect("every NodeService RPC must declare an auth-policy beside its proto definition");
let method = rpc.split_once('(').expect("RPC declaration must have a request type").0;
assert!(
policies.insert(normalized_rpc_method(method), policy.trim()).is_none(),
"duplicate NodeService RPC {method}",
);
}
assert!(!policies.is_empty(), "NodeService must declare RPC methods");
policies
}
#[test]
fn every_node_service_rpc_declares_an_auth_policy() {
const VALID_POLICIES: [&str; 4] = ["body-bound", "read-only", "streaming", "unimplemented"];
for (method, policy) in node_service_auth_policies() {
assert!(
VALID_POLICIES.contains(&policy),
"NodeService RPC {method} has unsupported auth-policy {policy:?}",
);
}
}
struct HealControlMockStorage;
#[async_trait::async_trait]
@@ -2462,9 +2594,14 @@ mod tests {
async fn every_mutating_handler_enforces_its_body_digest() {
let service = make_server();
let disk = "http://node-a:9000/data/rustfs0".to_string();
let mut covered_methods = HashSet::new();
macro_rules! assert_gated {
($method:ident, $msg:expr, $canonical:path) => {{
assert!(
covered_methods.insert(normalized_rpc_method(stringify!($method))),
concat!("duplicate disk mutation test for ", stringify!($method)),
);
let msg = $msg;
// Correct digest: the gate passes and the handler proceeds to the (unknown) disk
@@ -2580,6 +2717,37 @@ mod tests {
},
rustfs_protos::canonical_delete_request_body
);
assert_gated!(
acquire_snapshot_lease,
SnapshotLeaseRequest {
disk: disk.clone(),
volume: "v".into(),
path: "p".into(),
ttl_ms: 60_000,
},
rustfs_protos::canonical_snapshot_lease_request_body
);
assert_gated!(
renew_snapshot_lease,
SnapshotLeaseRenewRequest {
disk: disk.clone(),
volume: "v".into(),
path: "p".into(),
token: vec![1; 16].into(),
ttl_ms: 60_000,
},
rustfs_protos::canonical_snapshot_lease_renew_request_body
);
assert_gated!(
release_snapshot_lease,
SnapshotLeaseReleaseRequest {
disk: disk.clone(),
volume: "v".into(),
path: "p".into(),
token: vec![1; 16].into(),
},
rustfs_protos::canonical_snapshot_lease_release_request_body
);
assert_gated!(
delete_paths,
DeletePathsRequest {
@@ -2659,6 +2827,12 @@ mod tests {
},
rustfs_protos::canonical_make_volumes_request_body
);
let expected_methods = DISK_MUTATION_RPC_METHODS.into_iter().map(String::from).collect();
assert_eq!(
covered_methods, expected_methods,
"the disk mutation exclusion set must exactly match handlers exercised by the independent digest test",
);
}
#[tokio::test]
@@ -4416,6 +4590,31 @@ mod tests {
);
}
#[tokio::test]
async fn test_load_bucket_metadata_failure_skips_scanner_maintenance() {
let service = create_test_node_service();
let maintenance_generation = rustfs_scanner::scanner_maintenance_generation();
let request = Request::new(LoadBucketMetadataRequest {
bucket: "reload-miss-scanner-guard-bucket".to_string(),
scanner_maintenance_change: true,
});
let response = service.load_bucket_metadata(request).await.expect("rpc should reply");
let load_response = response.into_inner();
// Whether the reload fails on missing server state or on the absent
// persisted metadata, a failed reload must report failure and must
// not tell the scanner a maintenance change landed.
assert!(!load_response.success);
assert!(load_response.error_info.is_some());
assert_eq!(
rustfs_scanner::scanner_maintenance_generation(),
maintenance_generation,
"a failed metadata reload must not advance scanner maintenance activity"
);
}
#[tokio::test]
#[ignore = "requires isolated global object layer state"]
async fn test_load_bucket_metadata_no_object_layer() {
@@ -4726,6 +4925,128 @@ mod tests {
assert_eq!(signal_response.error_info.as_deref(), Some("unsupported service signal: 99"));
}
#[tokio::test]
async fn signal_service_body_digest_gate_runs_before_request_handling() {
let service = create_test_node_service();
let mut vars = HashMap::new();
vars.insert(PEER_RESTSIGNAL.to_string(), "99".to_string());
vars.insert(PEER_RESTSUB_SYS.to_string(), "scanner".to_string());
vars.insert(PEER_RESTDRY_RUN.to_string(), "false".to_string());
let message = SignalServiceRequest {
vars: Some(Mss { value: vars }),
};
let mut other = message.clone();
other
.vars
.as_mut()
.expect("signal vars should exist")
.value
.insert(PEER_RESTSIGNAL.to_string(), "1".to_string());
let mut tampered = Request::new(message.clone());
let other_body = other.canonical_body().expect("small signal request should encode");
set_tonic_canonical_body_digest(&mut tampered, &other_body).expect("digest metadata should encode");
mark_v2_authenticated(&mut tampered);
let error = service
.signal_service(tampered)
.await
.expect_err("a tampered signal request must fail before handler logic");
assert_eq!(error.code(), tonic::Code::PermissionDenied);
let mut signed = Request::new(message);
let body = signed.get_ref().canonical_body().expect("small signal request should encode");
set_tonic_canonical_body_digest(&mut signed, &body).expect("digest metadata should encode");
mark_v2_authenticated(&mut signed);
let response = service
.signal_service(signed)
.await
.expect("a correctly body-bound signal request must reach handler logic")
.into_inner();
assert!(!response.success);
assert_eq!(response.error_info.as_deref(), Some("unsupported service signal: 99"));
}
#[tokio::test]
async fn every_non_disk_mutation_rejects_a_mismatched_body_digest() {
let service = create_test_node_service();
let mut covered_methods = HashSet::new();
macro_rules! assert_tampered {
($method:ident, $message:expr) => {{
assert!(
covered_methods.insert(normalized_rpc_method(stringify!($method))),
concat!("duplicate non-disk mutation test for ", stringify!($method)),
);
let mut request = Request::new($message);
set_tonic_canonical_body_digest(&mut request, b"unrelated-canonical-body")
.expect("digest metadata should encode");
mark_v2_authenticated(&mut request);
let error = service
.$method(request)
.await
.expect_err(concat!(stringify!($method), " must reject a mismatched body digest"));
assert_eq!(
error.code(),
tonic::Code::PermissionDenied,
concat!(stringify!($method), " must authenticate before any mutation"),
);
}};
}
assert_tampered!(heal_bucket, HealBucketRequest::default());
assert_tampered!(make_bucket, MakeBucketRequest::default());
assert_tampered!(delete_bucket, DeleteBucketRequest::default());
assert_tampered!(lock, GenerallyLockRequest::default());
assert_tampered!(un_lock, GenerallyLockRequest::default());
assert_tampered!(force_un_lock, GenerallyLockRequest::default());
assert_tampered!(refresh, GenerallyLockRequest::default());
assert_tampered!(lock_batch, BatchGenerallyLockRequest::default());
assert_tampered!(un_lock_batch, BatchGenerallyLockRequest::default());
assert_tampered!(load_bucket_metadata, LoadBucketMetadataRequest::default());
assert_tampered!(delete_bucket_metadata, DeleteBucketMetadataRequest::default());
assert_tampered!(delete_policy, DeletePolicyRequest::default());
assert_tampered!(load_policy, LoadPolicyRequest::default());
assert_tampered!(load_policy_mapping, LoadPolicyMappingRequest::default());
assert_tampered!(delete_user, DeleteUserRequest::default());
assert_tampered!(delete_service_account, DeleteServiceAccountRequest::default());
assert_tampered!(load_user, LoadUserRequest::default());
assert_tampered!(load_service_account, LoadServiceAccountRequest::default());
assert_tampered!(load_group, LoadGroupRequest::default());
assert_tampered!(reload_site_replication_config, ReloadSiteReplicationConfigRequest::default());
assert_tampered!(signal_service, SignalServiceRequest::default());
assert_tampered!(
scanner_activity,
ScannerActivityRequest {
challenge: vec![7; 16].into(),
protocol_version: rustfs_scanner::SCANNER_ACTIVITY_PROTOCOL_VERSION,
acknowledge_instance_id: String::new(),
acknowledge_dirty_usage_generation: 0,
}
);
assert_tampered!(reload_pool_meta, ReloadPoolMetaRequest::default());
assert_tampered!(stop_rebalance, StopRebalanceRequest::default());
assert_tampered!(load_rebalance_meta, LoadRebalanceMetaRequest::default());
assert_tampered!(start_decommission, StartDecommissionRequest::default());
assert_tampered!(cancel_decommission, CancelDecommissionRequest::default());
assert_tampered!(clear_decommission, ClearDecommissionRequest::default());
assert_tampered!(load_transition_tier_config, LoadTransitionTierConfigRequest::default());
let body_bound_methods: HashSet<_> = node_service_auth_policies()
.into_iter()
.filter_map(|(method, policy)| (policy == "body-bound").then_some(method))
.collect();
let disk_methods: HashSet<_> = DISK_MUTATION_RPC_METHODS.into_iter().map(String::from).collect();
assert!(
disk_methods.is_subset(&body_bound_methods),
"every independently tested disk mutation must remain declared body-bound",
);
let expected_methods: HashSet<_> = body_bound_methods.difference(&disk_methods).cloned().collect();
assert_eq!(
covered_methods, expected_methods,
"proto body-bound non-disk RPCs must exactly match handlers exercised by mismatch tests",
);
}
#[tokio::test]
async fn test_scanner_activity_requires_body_bound_auth_before_storage_lookup() {
let service = create_test_node_service();
@@ -4790,26 +5111,6 @@ mod tests {
.expect_err("a signed malformed challenge must fail before storage lookup");
assert_eq!(malformed_current.code(), tonic::Code::InvalidArgument);
let mut tampered = Request::new(ScannerActivityRequest {
challenge: vec![7; 16].into(),
protocol_version: rustfs_scanner::SCANNER_ACTIVITY_PROTOCOL_VERSION,
acknowledge_instance_id: String::new(),
acknowledge_dirty_usage_generation: 0,
});
let other = ScannerActivityRequest {
challenge: vec![8; 16].into(),
..tampered.get_ref().clone()
};
let other_canonical =
rustfs_protos::canonical_scanner_activity_request_body(&other).expect("scanner activity request should encode");
set_tonic_canonical_body_digest(&mut tampered, &other_canonical).expect("digest metadata should encode");
mark_v2_authenticated(&mut tampered);
let tampered = service
.scanner_activity(tampered)
.await
.expect_err("tampered activity challenge must fail before storage lookup");
assert_eq!(tampered.code(), tonic::Code::PermissionDenied);
let mut downgraded = Request::new(ScannerActivityRequest {
challenge: vec![7; 16].into(),
protocol_version: rustfs_scanner::SCANNER_ACTIVITY_PROTOCOL_VERSION,
+4 -10
View File
@@ -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, remove_bucket_metadata, set_bucket_metadata,
DiskError, StoragePeerS3ClientExt as _, reload_bucket_metadata, remove_bucket_metadata,
};
use rustfs_common::heal_channel::HealOpts;
use rustfs_protos::proto_gen::node_service::*;
@@ -66,21 +66,15 @@ impl NodeService {
}));
}
let Some(store) = self.resolve_object_store() else {
let Some(_store) = self.resolve_object_store() else {
return Ok(Response::new(LoadBucketMetadataResponse {
success: false,
error_info: Some("errServerNotInitialized".to_string()),
}));
};
match load_bucket_metadata(store, &bucket).await {
Ok(meta) => {
if let Err(err) = set_bucket_metadata(bucket.clone(), meta).await {
return Ok(Response::new(LoadBucketMetadataResponse {
success: false,
error_info: Some(err.to_string()),
}));
};
match reload_bucket_metadata(&bucket).await {
Ok(()) => {
if scanner_maintenance_change {
rustfs_scanner::record_scanner_maintenance_change(&bucket);
}
+239 -6
View File
@@ -14,11 +14,12 @@
use super::NodeService;
use crate::storage::storage_api::rpc_consumer::node_service::{
BatchReadVersionReq, BatchReadVersionResp, DeleteOptions, DiskError, DiskInfoOptions, FileInfoVersions, ReadMultipleReq,
ReadMultipleResp, ReadOptions, StorageDiskRpcExt as _, UpdateMetadataOpts, validate_batch_read_version_item_count,
BatchReadVersionReq, BatchReadVersionResp, DeleteOptions, DiskError, DiskInfoOptions, DiskStore, FileInfoVersions,
ReadMultipleReq, ReadMultipleResp, ReadOptions, StorageDiskRpcExt as _, UpdateMetadataOpts,
validate_batch_read_version_item_count,
};
use crate::storage::storage_api::runtime_sources_consumer::runtime_sources;
use crate::storage::storage_api::{PartTransactionAction, verify_tonic_mutation_body_digest};
use crate::storage::storage_api::{PartTransactionAction, SnapshotLeaseToken, verify_tonic_mutation_body_digest};
use bytes::Bytes;
use rustfs_filemeta::FileInfo;
use rustfs_io_metrics::internode_metrics::{
@@ -28,13 +29,96 @@ use rustfs_io_metrics::internode_metrics::{
};
use rustfs_protos::proto_gen::node_service::*;
use serde::de::DeserializeOwned;
use std::io::Cursor;
use std::{collections::HashMap, io::Cursor, time::Duration};
use tokio::sync::mpsc;
use tokio_util::time::DelayQueue;
use tonic::{Request, Response, Status};
use tracing::debug;
/// Initial capacity hint (bytes) for msgpack encode buffers, sized to cover a typical single-
/// version `FileInfo` without repeated growth reallocations. Larger payloads still grow as needed.
const MSGPACK_ENCODE_CAPACITY_HINT: usize = 512;
const SNAPSHOT_LEASE_PROTOCOL_VERSION: u32 = 1;
const SNAPSHOT_LEASE_MIN_TTL: Duration = Duration::from_secs(5);
const SNAPSHOT_LEASE_MAX_TTL: Duration = Duration::from_secs(5 * 60);
struct SnapshotLeaseExpiry {
disk: DiskStore,
volume: String,
path: String,
token: SnapshotLeaseToken,
}
pub(super) struct SnapshotLeaseExpiryScheduler {
tx: mpsc::UnboundedSender<SnapshotLeaseExpiryCommand>,
}
enum SnapshotLeaseExpiryCommand {
Schedule(SnapshotLeaseExpiry, Duration),
Cancel(SnapshotLeaseToken),
}
impl SnapshotLeaseExpiryScheduler {
pub(super) fn new() -> Self {
let (tx, mut rx) = mpsc::unbounded_channel();
tokio::spawn(async move {
let mut expirations: DelayQueue<SnapshotLeaseExpiry> = DelayQueue::new();
let mut keys = HashMap::new();
loop {
tokio::select! {
Some(command) = rx.recv() => {
match command {
SnapshotLeaseExpiryCommand::Schedule(expiry, ttl) => {
if let Some(key) = keys.remove(&expiry.token) {
expirations.remove(&key);
}
let token = expiry.token;
let key = expirations.insert(expiry, ttl);
keys.insert(token, key);
}
SnapshotLeaseExpiryCommand::Cancel(token) => {
if let Some(key) = keys.remove(&token) {
expirations.remove(&key);
}
}
}
}
Some(expired) = futures_util::StreamExt::next(&mut expirations), if !expirations.is_empty() => {
let expiry = expired.into_inner();
keys.remove(&expiry.token);
let _ = expiry
.disk
.release_snapshot_lease(&expiry.volume, &expiry.path, expiry.token)
.await;
}
else => break,
}
}
});
Self { tx }
}
fn schedule(&self, expiry: SnapshotLeaseExpiry, ttl: Duration) -> Result<(), SnapshotLeaseExpiry> {
self.tx
.send(SnapshotLeaseExpiryCommand::Schedule(expiry, ttl))
.map_err(|err| match err.0 {
SnapshotLeaseExpiryCommand::Schedule(expiry, _) => expiry,
SnapshotLeaseExpiryCommand::Cancel(_) => unreachable!(),
})
}
fn cancel(&self, token: SnapshotLeaseToken) {
let _ = self.tx.send(SnapshotLeaseExpiryCommand::Cancel(token));
}
}
fn snapshot_lease_ttl(ttl_ms: u64) -> Result<Duration, Status> {
let ttl = Duration::from_millis(ttl_ms);
if !(SNAPSHOT_LEASE_MIN_TTL..=SNAPSHOT_LEASE_MAX_TTL).contains(&ttl) {
return Err(Status::invalid_argument("snapshot lease TTL is outside the supported range"));
}
Ok(ttl)
}
fn decode_msgpack_or_json<T: DeserializeOwned>(
binary: &[u8],
@@ -165,6 +249,146 @@ fn encode_batch_read_version_response_payloads(
}
impl NodeService {
pub(super) async fn handle_acquire_snapshot_lease(
&self,
request: Request<SnapshotLeaseRequest>,
) -> Result<Response<SnapshotLeaseResponse>, Status> {
verify_disk_mutation_digest(
&request,
rustfs_protos::canonical_snapshot_lease_request_body(request.get_ref()),
"acquire_snapshot_lease",
)?;
let request = request.into_inner();
let ttl = snapshot_lease_ttl(request.ttl_ms)?;
let Some(disk) = self.find_disk(&request.disk).await else {
return Ok(Response::new(SnapshotLeaseResponse {
success: false,
token: Bytes::new(),
protocol_version: SNAPSHOT_LEASE_PROTOCOL_VERSION,
error: Some(DiskError::other("cannot find disk").into()),
}));
};
match disk.acquire_snapshot_lease(&request.volume, &request.path).await {
Ok(token) => {
if let Err(expiry) = self.snapshot_lease_expiry.schedule(
SnapshotLeaseExpiry {
disk,
volume: request.volume,
path: request.path,
token,
},
ttl,
) {
let _ = expiry
.disk
.release_snapshot_lease(&expiry.volume, &expiry.path, expiry.token)
.await;
return Err(Status::internal("snapshot lease expiry scheduler is unavailable"));
}
Ok(Response::new(SnapshotLeaseResponse {
success: true,
token: Bytes::copy_from_slice(token.as_bytes()),
protocol_version: SNAPSHOT_LEASE_PROTOCOL_VERSION,
error: None,
}))
}
Err(err) => Ok(Response::new(SnapshotLeaseResponse {
success: false,
token: Bytes::new(),
protocol_version: SNAPSHOT_LEASE_PROTOCOL_VERSION,
error: Some(err.into()),
})),
}
}
pub(super) async fn handle_renew_snapshot_lease(
&self,
request: Request<SnapshotLeaseRenewRequest>,
) -> Result<Response<SnapshotLeaseResponse>, Status> {
verify_disk_mutation_digest(
&request,
rustfs_protos::canonical_snapshot_lease_renew_request_body(request.get_ref()),
"renew_snapshot_lease",
)?;
let request = request.into_inner();
let ttl = snapshot_lease_ttl(request.ttl_ms)?;
let token =
SnapshotLeaseToken::from_slice(&request.token).map_err(|_| Status::invalid_argument("invalid lease token"))?;
let Some(disk) = self.find_disk(&request.disk).await else {
return Ok(Response::new(SnapshotLeaseResponse {
success: false,
token: Bytes::new(),
protocol_version: SNAPSHOT_LEASE_PROTOCOL_VERSION,
error: Some(DiskError::other("cannot find disk").into()),
}));
};
match disk.renew_snapshot_lease(&request.volume, &request.path, token).await {
Ok(renewed) => {
if let Err(expiry) = self.snapshot_lease_expiry.schedule(
SnapshotLeaseExpiry {
disk,
volume: request.volume,
path: request.path,
token: renewed,
},
ttl,
) {
let _ = expiry
.disk
.release_snapshot_lease(&expiry.volume, &expiry.path, expiry.token)
.await;
return Err(Status::internal("snapshot lease expiry scheduler is unavailable"));
}
self.snapshot_lease_expiry.cancel(token);
Ok(Response::new(SnapshotLeaseResponse {
success: true,
token: Bytes::copy_from_slice(renewed.as_bytes()),
protocol_version: SNAPSHOT_LEASE_PROTOCOL_VERSION,
error: None,
}))
}
Err(err) => Ok(Response::new(SnapshotLeaseResponse {
success: false,
token: Bytes::new(),
protocol_version: SNAPSHOT_LEASE_PROTOCOL_VERSION,
error: Some(err.into()),
})),
}
}
pub(super) async fn handle_release_snapshot_lease(
&self,
request: Request<SnapshotLeaseReleaseRequest>,
) -> Result<Response<SnapshotLeaseMutationResponse>, Status> {
verify_disk_mutation_digest(
&request,
rustfs_protos::canonical_snapshot_lease_release_request_body(request.get_ref()),
"release_snapshot_lease",
)?;
let request = request.into_inner();
let token =
SnapshotLeaseToken::from_slice(&request.token).map_err(|_| Status::invalid_argument("invalid lease token"))?;
let Some(disk) = self.find_disk(&request.disk).await else {
return Ok(Response::new(SnapshotLeaseMutationResponse {
success: false,
error: Some(DiskError::other("cannot find disk").into()),
}));
};
match disk.release_snapshot_lease(&request.volume, &request.path, token).await {
Ok(()) => {
self.snapshot_lease_expiry.cancel(token);
Ok(Response::new(SnapshotLeaseMutationResponse {
success: true,
error: None,
}))
}
Err(err) => Ok(Response::new(SnapshotLeaseMutationResponse {
success: false,
error: Some(err.into()),
})),
}
}
pub(super) async fn handle_disk_info(&self, request: Request<DiskInfoRequest>) -> Result<Response<DiskInfoResponse>, Status> {
let request = request.into_inner();
if let Some(disk) = self.find_disk(&request.disk).await {
@@ -1368,8 +1592,9 @@ impl NodeService {
#[cfg(test)]
mod tests {
use super::{
compat_response_json, decode_msgpack_or_json, encode_batch_read_version_response_payloads, encode_msgpack,
encode_msgpack_named, encode_read_multiple_response_payloads,
SNAPSHOT_LEASE_MAX_TTL, SNAPSHOT_LEASE_MIN_TTL, compat_response_json, decode_msgpack_or_json,
encode_batch_read_version_response_payloads, encode_msgpack, encode_msgpack_named,
encode_read_multiple_response_payloads, snapshot_lease_ttl,
};
use crate::storage::storage_api::ReadMultipleResp;
use crate::storage::storage_api::rpc_consumer::node_service::BatchReadVersionResp;
@@ -1382,6 +1607,14 @@ mod tests {
count: u32,
}
#[test]
fn snapshot_lease_ttl_rejects_values_outside_server_bounds() {
assert!(snapshot_lease_ttl(4_999).is_err());
assert_eq!(snapshot_lease_ttl(5_000).unwrap(), SNAPSHOT_LEASE_MIN_TTL);
assert_eq!(snapshot_lease_ttl(300_000).unwrap(), SNAPSHOT_LEASE_MAX_TTL);
assert!(snapshot_lease_ttl(300_001).is_err());
}
#[test]
fn decode_msgpack_or_json_prefers_binary_payload() {
let payload = SamplePayload {
+24 -7
View File
@@ -236,8 +236,8 @@ pub(crate) mod rpc_consumer {
ECStore, Error, FileInfoVersions, LocalPeerS3Client, MetricType, PEER_RESTDRY_RUN, 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,
remove_bucket_metadata, set_bucket_metadata, validate_batch_read_version_item_count,
find_local_disk_by_ref, get_local_server_property, reload_bucket_metadata, reload_transition_tier_config,
remove_bucket_metadata, validate_batch_read_version_item_count,
};
pub(crate) type StorageResult<T> = super::super::Result<T>;
@@ -431,7 +431,7 @@ pub(crate) mod ecstore_disk {
pub(crate) use rustfs_ecstore::api::disk::{
BatchReadVersionReq, BatchReadVersionResp, CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskStore,
FileInfoVersions, FileReader, FileWriter, OldCurrentSize, PartTransactionAction, RUSTFS_META_BUCKET, ReadMultipleReq,
ReadMultipleResp, ReadOptions, RenameDataResp, UpdateMetadataOpts, VolumeInfo, WalkDirOptions,
ReadMultipleResp, ReadOptions, RenameDataResp, SnapshotLeaseToken, UpdateMetadataOpts, VolumeInfo, WalkDirOptions,
get_object_disk_read_timeout, validate_batch_read_version_item_count,
};
pub(crate) use rustfs_ecstore::api::disk::{endpoint, error, error_reduce};
@@ -606,6 +606,7 @@ pub(crate) type ExpiryState = ecstore_bucket::lifecycle::bucket_lifecycle_ops::E
pub(crate) type FileInfoVersions = ecstore_disk::FileInfoVersions;
pub(crate) type FileReader = ecstore_disk::FileReader;
pub(crate) type FileWriter = ecstore_disk::FileWriter;
pub(crate) type SnapshotLeaseToken = ecstore_disk::SnapshotLeaseToken;
pub(crate) type FS = super::ecfs::FS;
pub(crate) type HashReader = ecstore_rio::HashReader;
pub(crate) type InstanceContext = ecstore_runtime::InstanceContext;
@@ -1075,6 +1076,9 @@ pub(crate) trait StorageDiskRpcExt {
) -> DiskResult<()>;
async fn read_metadata(&self, volume: &str, path: &str) -> DiskResult<bytes::Bytes>;
async fn delete_paths(&self, volume: &str, paths: &[String]) -> DiskResult<()>;
async fn acquire_snapshot_lease(&self, volume: &str, path: &str) -> DiskResult<SnapshotLeaseToken>;
async fn renew_snapshot_lease(&self, volume: &str, path: &str, token: SnapshotLeaseToken) -> DiskResult<SnapshotLeaseToken>;
async fn release_snapshot_lease(&self, volume: &str, path: &str, token: SnapshotLeaseToken) -> DiskResult<()>;
async fn stat_volume(&self, volume: &str) -> DiskResult<VolumeInfo>;
async fn list_volumes(&self) -> DiskResult<Vec<VolumeInfo>>;
async fn make_volume(&self, volume: &str) -> DiskResult<()>;
@@ -1201,6 +1205,18 @@ where
ecstore_disk::DiskAPI::delete_paths(self, volume, paths).await
}
async fn acquire_snapshot_lease(&self, volume: &str, path: &str) -> DiskResult<SnapshotLeaseToken> {
ecstore_disk::DiskAPI::acquire_snapshot_lease(self, volume, path).await
}
async fn renew_snapshot_lease(&self, volume: &str, path: &str, token: SnapshotLeaseToken) -> DiskResult<SnapshotLeaseToken> {
ecstore_disk::DiskAPI::renew_snapshot_lease(self, volume, path, token).await
}
async fn release_snapshot_lease(&self, volume: &str, path: &str, token: SnapshotLeaseToken) -> DiskResult<()> {
ecstore_disk::DiskAPI::release_snapshot_lease(self, volume, path, token).await
}
async fn stat_volume(&self, volume: &str) -> DiskResult<VolumeInfo> {
ecstore_disk::DiskAPI::stat_volume(self, volume).await
}
@@ -1349,10 +1365,6 @@ impl StoragePeerS3ClientExt for LocalPeerS3Client {
}
}
pub(crate) async fn load_bucket_metadata(api: Arc<ECStore>, bucket: &str) -> Result<BucketMetadata> {
ecstore_bucket::metadata::load_bucket_metadata(api, bucket).await
}
#[cfg(test)]
pub(crate) fn bucket_metadata_sys_initialized() -> bool {
ecstore_bucket::metadata_sys::get_global_bucket_metadata_sys().is_some()
@@ -1425,10 +1437,15 @@ pub(crate) async fn get_bucket_website_config(bucket: &str) -> Result<(s3s::dto:
ecstore_bucket::metadata_sys::get_website_config(bucket).await
}
#[cfg(test)]
pub(crate) async fn set_bucket_metadata(bucket: String, bm: BucketMetadata) -> Result<()> {
ecstore_bucket::metadata_sys::set_bucket_metadata(bucket, bm).await
}
pub(crate) async fn reload_bucket_metadata(bucket: &str) -> Result<()> {
ecstore_bucket::metadata_sys::reload_bucket_metadata(bucket).await
}
pub(crate) async fn remove_bucket_metadata(bucket: &str) -> Result<bool> {
ecstore_bucket::metadata_sys::remove_bucket_metadata(bucket).await
}