Compare commits

..

20 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
cxymds cb62079ba6 test(ilm): make active cancellation deterministic (#5403)
* ci: preserve timeout evidence for workspace tests

* test(ilm): make active cancellation deterministic

* ci: allow cold doctest compilation to finish

---------

Co-authored-by: houseme <housemecn@gmail.com>
2026-07-29 13:53:59 +08:00
Zhengchao An d39ffdb1cd fix(filemeta): canonicalize version order after insert (#5413) 2026-07-29 05:49:00 +00:00
Zhengchao An 7d698abc1f ci(checksums): inherit workspace lint policy (#5415) 2026-07-29 12:53:45 +08:00
hector a1a65ad65d fix(action): change quay.io image repository name (#5417) 2026-07-29 12:53:26 +08:00
cxymds 358af6a8de ci: preserve timeout evidence for workspace tests (#5402)
* ci: preserve timeout evidence for workspace tests

* ci: allow cold doctest compilation to finish

* ci: allow cold nextest compilation to finish

* ci: allow cold nextest compilation to finish

---------

Co-authored-by: houseme <housemecn@gmail.com>
2026-07-29 04:49:03 +00:00
Zhengchao An 90d1a15d13 fix(ecstore): classify peer RPC failures by gRPC status code (#5400) 2026-07-29 11:35:09 +08:00
48 changed files with 3917 additions and 1248 deletions
+58 -14
View File
@@ -156,16 +156,69 @@ jobs:
github-token: ${{ secrets.GITHUB_TOKEN }}
cache-save-if: ${{ github.ref == 'refs/heads/main' }}
- name: Prepare test evidence
run: |
mkdir -p artifacts/test-and-lint
{
echo "run_id=${GITHUB_RUN_ID}"
echo "job=${GITHUB_JOB}"
echo "runner=${RUNNER_NAME}"
echo "started_at=$(date --utc --iso-8601=seconds)"
} > artifacts/test-and-lint/run-metadata.txt
# Clippy runs before the test pass: lint failures are the most common
# CI-only breakage and should surface in minutes, not after 20+ minutes
# of tests.
- name: Run clippy lints
run: cargo clippy --all-targets -- -D warnings
- name: Run tests
- name: Run nextest tests
run: |
cargo nextest run --profile ci --all --exclude e2e_test
cargo test --all --doc
mkdir -p artifacts/test-and-lint
set +e
NEXTEST_HIDE_PROGRESS_BAR=1 timeout --verbose --signal=TERM --kill-after=30s 75m \
cargo nextest run --profile ci --all --exclude e2e_test \
--status-level all --final-status-level all \
2>&1 | tee artifacts/test-and-lint/nextest.log
status=${PIPESTATUS[0]}
{
echo "command=cargo nextest run --profile ci --all --exclude e2e_test"
echo "exit_status=${status}"
echo "finished_at=$(date --utc --iso-8601=seconds)"
echo
echo "Remaining test-related processes:"
pgrep -af 'cargo|nextest|target/.*/deps/' || true
} > artifacts/test-and-lint/nextest-diagnostics.txt
exit "${status}"
- name: Run documentation tests
run: |
mkdir -p artifacts/test-and-lint
set +e
timeout --verbose --signal=TERM --kill-after=30s 15m \
cargo test --all --doc \
2>&1 | tee artifacts/test-and-lint/doctest.log
status=${PIPESTATUS[0]}
{
echo "command=cargo test --all --doc"
echo "exit_status=${status}"
echo "finished_at=$(date --utc --iso-8601=seconds)"
echo
echo "Remaining test-related processes:"
pgrep -af 'cargo|rustdoc|target/.*/deps/' || true
} > artifacts/test-and-lint/doctest-diagnostics.txt
exit "${status}"
- name: Upload test reports and diagnostics
if: always()
uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
with:
name: junit-test-and-lint-${{ github.run_number }}
path: |
target/nextest/ci/junit.xml
artifacts/test-and-lint
retention-days: 3
if-no-files-found: error
# rustfs/backlog#1289: fail if a seed rule's log anchor no longer exists
# verbatim in the source tree (log message drifted without updating the
@@ -174,15 +227,6 @@ jobs:
- name: Check log-analyzer rule anchors
run: ./scripts/check_log_analyzer_rules.sh
- name: Upload test junit report
if: always()
uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
with:
name: junit-test-and-lint-${{ github.run_number }}
path: target/nextest/ci/junit.xml
retention-days: 3
if-no-files-found: ignore
# Explicit gate for migration-critical suites. These tests already ran in
# the full nextest pass above; a single filtered nextest invocation keeps
# the named gate without rebuilding or re-running them one package at a time.
@@ -328,7 +372,7 @@ jobs:
github-token: ${{ secrets.GITHUB_TOKEN }}
- name: Build debug binary
run: cargo build -p rustfs --bins
run: cargo build -p rustfs --bins --features e2e-test-hooks
- name: Upload debug binary
uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
@@ -358,7 +402,7 @@ jobs:
github-token: ${{ secrets.GITHUB_TOKEN }}
- name: Build debug binary with rio-v2
run: cargo build -p rustfs --bins --features rio-v2
run: cargo build -p rustfs --bins --features rio-v2,e2e-test-hooks
- name: Upload debug binary
uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
+1 -1
View File
@@ -66,7 +66,7 @@ env:
CARGO_TERM_COLOR: always
REGISTRY_DOCKERHUB: rustfs/rustfs
REGISTRY_GHCR: ghcr.io/${{ github.repository }}
REGISTRY_QUAY: quay.io/${{ secrets.QUAY_USERNAME }}/rustfs
REGISTRY_QUAY: quay.io/rustfs/rustfs
DOCKER_PLATFORMS: linux/amd64,linux/arm64
jobs:
+3
View File
@@ -25,6 +25,9 @@ keywords = ["checksum-calculation", "verification", "integrity", "authenticity",
categories = ["web-programming", "development-tools", "network-programming"]
documentation = "https://docs.rs/rustfs-checksums/latest/rustfs_checksum/"
[lints]
workspace = true
[dependencies]
bytes = { workspace = true, features = ["serde"] }
crc-fast = { workspace = true }
+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();
+10 -2
View File
@@ -95,6 +95,7 @@ const MANUAL_ASYNC_PARALLEL_OBJECTS: usize = 64;
const MANUAL_ACTIVE_CANCEL_OBJECTS: usize = 512;
const MANUAL_RESTART_CANCEL_OBJECTS: usize = 512;
const MANUAL_ACTIVE_CANCEL_RUNNING_TIMEOUT: StdDuration = StdDuration::from_secs(15);
const MANUAL_TRANSITION_CANCEL_BARRIER_ENV: &str = "RUSTFS_E2E_MANUAL_TRANSITION_CANCEL_BARRIER";
const MANUAL_ASYNC_CONFLICT_TERMINAL_TIMEOUT: StdDuration = StdDuration::from_secs(90);
const MANUAL_RESTART_RECOVERY_TIMEOUT: StdDuration = StdDuration::from_secs(80);
const OBJECT_KEY: &str = "tier/鲁A12345/report.bin";
@@ -1684,8 +1685,15 @@ async fn test_manual_transition_async_active_cancel_reports_terminal_cancelled()
cold_client.create_bucket().bucket(TIER_BUCKET).send().await?;
let mut hot = RustFSTestEnvironment::new().await?;
hot.start_rustfs_server_with_env(vec![], &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")])
.await?;
hot.start_rustfs_server_with_env(
vec![],
&[
("RUSTFS_SCANNER_ENABLED", "false"),
("RUSTFS_SCANNER_CYCLE", "3600"),
(MANUAL_TRANSITION_CANCEL_BARRIER_ENV, "1"),
],
)
.await?;
let hot_client = hot.create_s3_client();
add_rustfs_tier(&hot, &cold).await?;
+2 -2
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,
};
}
@@ -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(),
+135 -19
View File
@@ -14,7 +14,7 @@
use crate::cluster::rpc::http_auth::RPC_CONTENT_SHA256_HEADER;
use crate::cluster::rpc::{gen_tonic_signature_headers, normalize_tonic_rpc_audience};
use crate::disk::error::{DiskError, Error as DiskErrorType};
use crate::disk::error::{DiskError, Error as DiskErrorType, RpcStatusError};
use crate::runtime::sources as runtime_sources;
use http::Uri;
use rustfs_protos::{
@@ -107,10 +107,79 @@ pub async fn node_service_time_out_client_no_auth(
node_service_time_out_client(addr, TonicInterceptor::NoOp(NoOpInterceptor)).await
}
/// The typed `tonic::Status` an internode RPC failure was converted from, if
/// this error carries one.
pub(crate) fn embedded_tonic_status(io_err: &std::io::Error) -> Option<&tonic::Status> {
io_err.get_ref()?.downcast_ref::<RpcStatusError>().map(RpcStatusError::status)
}
/// Decide whether a gRPC status reports a peer we cannot currently reach,
/// rather than an application outcome from a live peer.
///
/// `Unavailable` is the one code that means "no service behind this channel":
/// the client transport raises it when the connection is broken, and the
/// server's own not-ready gates use it deliberately.
///
/// `Unknown` is the client transport's escape hatch for a cause it could not
/// map to a code — tower's "Service was not ready: <cause>", an h2 error with
/// no gRPC mapping. Our handlers never return it, so there its message is the
/// only evidence available and the anchored needles decide.
///
/// Every other code is an answer from a live peer and is never a transport
/// failure, whatever its message says. That distinction is the point of
/// classifying by code: a peer relaying its own downstream trouble as
/// `Internal("connection refused ...")`, or a handler interpolating a local
/// `io::Error` into `Status::internal`, answered us perfectly well. Marking it
/// offline over that text is the bug this classification replaces. Likewise a
/// `Cancelled` "Timeout expired" from the per-RPC channel deadline means the
/// peer is slow, not gone; gating it would turn load into a partition.
pub(crate) fn is_network_like_status(status: &tonic::Status) -> bool {
match status.code() {
tonic::Code::Unavailable => true,
tonic::Code::Unknown => message_has_network_needle(&status.to_string()),
_ => false,
}
}
/// Substring fallback for failures that only exist as text: dial errors
/// wrapped by `get_client`, remote `error_info` payloads, and statuses
/// flattened through `format!`. Needles must stay anchored to transport
/// context — a bare word like "unavailable" also matches application text
/// (e.g. a bucket named "unavailable-logs") and would take a healthy peer
/// offline.
pub(crate) fn message_has_network_needle(message: &str) -> bool {
let message = message.to_ascii_lowercase();
[
"temporarily offline",
"transport error",
// tonic >= 0.14 renders Code::Unavailable as
// `code: 'The service is currently unavailable'`.
"code: 'the service is currently unavailable'",
// RUSTFS_COMPAT_TODO(tonic-013-status-render): releases up to 1.0.0-alpha.38 shipped tonic 0.13, which rendered the same status as `status: Unavailable`, and peers relay that text in error_info. Remove after the minimum supported RustFS peer version ships tonic >= 0.14.
"status: unavailable",
"error trying to connect",
"connection refused",
"connection reset",
"broken pipe",
"not connected",
"unexpected eof",
"timed out",
"deadline has elapsed",
"connection closed",
"connection aborted",
"tcp connect error",
]
.iter()
.any(|needle| message.contains(needle))
}
pub(crate) fn is_network_like_disk_error(err: &DiskErrorType) -> bool {
match err {
DiskError::Timeout => true,
DiskError::Io(io_err) => {
if let Some(status) = embedded_tonic_status(io_err) {
return is_network_like_status(status);
}
if matches!(
io_err.kind(),
ErrorKind::TimedOut
@@ -124,24 +193,7 @@ pub(crate) fn is_network_like_disk_error(err: &DiskErrorType) -> bool {
return true;
}
let message = io_err.to_string().to_ascii_lowercase();
[
"transport error",
"unavailable",
"error trying to connect",
"connection refused",
"connection reset",
"broken pipe",
"not connected",
"unexpected eof",
"timed out",
"deadline has elapsed",
"connection closed",
"connection aborted",
"tcp connect error",
]
.iter()
.any(|needle| message.contains(needle))
message_has_network_needle(&io_err.to_string())
}
_ => false,
}
@@ -269,6 +321,70 @@ mod tests {
let _ = provider.shutdown();
}
#[test]
fn network_like_disk_error_uses_typed_status_code() {
// Transport-level Unavailable statuses justify retry/eviction.
assert!(is_network_like_disk_error(&DiskError::from(tonic::Status::unavailable(
"storage layer is not initialized"
))));
// Application statuses from a live peer must not look network-like,
// even when their message contains transport-sounding words.
assert!(!is_network_like_disk_error(&DiskError::from(tonic::Status::internal(
"failed to heal bucket \"unavailable-logs\""
))));
assert!(!is_network_like_disk_error(&DiskError::from(tonic::Status::unauthenticated(
"No valid auth token"
))));
// A slow peer that blew the per-RPC deadline is still answering.
assert!(!is_network_like_disk_error(&DiskError::from(tonic::Status::cancelled("Timeout expired"))));
}
#[test]
fn embedded_tonic_status_is_recovered_across_error_conversions() {
// DiskError and StorageError share one wrapper, so a status keeps its
// typed classification whichever error it was converted into first.
let from_storage: DiskErrorType = crate::error::Error::from(tonic::Status::unavailable("peer gone")).into();
let DiskError::Io(io_err) = &from_storage else {
panic!("status-derived disk error should stay an Io error");
};
assert_eq!(embedded_tonic_status(io_err).map(|status| status.code()), Some(tonic::Code::Unavailable));
let from_disk = crate::error::Error::from(DiskError::from(tonic::Status::unavailable("peer gone")));
let crate::error::Error::Io(io_err) = &from_disk else {
panic!("status-derived storage error should stay an Io error");
};
assert_eq!(embedded_tonic_status(io_err).map(|status| status.code()), Some(tonic::Code::Unavailable));
}
#[test]
fn network_like_disk_error_ignores_transport_words_in_application_statuses() {
// Same contract as the peer client: a status the peer answered with
// is not a transport failure, so it must not drive a reconnect even
// when its message describes one.
assert!(!is_network_like_disk_error(&DiskError::from(tonic::Status::internal(
"connection refused while dialing downstream backend"
))));
assert!(!is_network_like_disk_error(&DiskError::from(tonic::Status::unauthenticated(
"connection reset while validating token"
))));
}
#[test]
fn network_like_disk_error_requires_anchored_unavailable_needle() {
// Regression: a bare "unavailable" needle used to match application
// text such as a bucket name.
assert!(!is_network_like_disk_error(&DiskError::other("bucket \"unavailable-logs\" not found")));
// Anchored renderings of a flattened Unavailable status still match.
assert!(is_network_like_disk_error(&DiskError::other(
"code: 'The service is currently unavailable', message: \"peer gone\""
)));
assert!(is_network_like_disk_error(&DiskError::other(
"status: Unavailable, message: \"peer gone\""
)));
assert!(is_network_like_disk_error(&DiskError::other("connection refused")));
assert!(!is_network_like_disk_error(&DiskError::FileNotFound));
}
#[test]
fn test_signature_interceptor_keeps_auth_headers() {
ensure_test_rpc_secret();
+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;
@@ -13,10 +13,10 @@
// limitations under the License.
use crate::cluster::rpc::client::{
TonicInterceptor, gen_tonic_signature_interceptor, heal_control_time_out_client, node_service_time_out_client,
tier_mutation_control_time_out_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;
@@ -471,26 +469,21 @@ impl PeerRestClient {
self.offline.store(false, Ordering::Release);
}
/// Whether this failure means the peer is unreachable, so it should be
/// gated offline and its connection evicted.
///
/// RPC failures are classified by their typed gRPC code first
/// (`is_network_like_status`); an application error from a live peer must
/// never take it offline no matter what its message says. The substring
/// fallback only covers failures that exist purely as text, such as the
/// dial errors `get_client` wraps.
fn is_network_like_error(err: &Error) -> bool {
let message = err.to_string().to_ascii_lowercase();
[
"temporarily offline",
"transport error",
"unavailable",
"error trying to connect",
"connection refused",
"connection reset",
"broken pipe",
"not connected",
"unexpected eof",
"timed out",
"deadline has elapsed",
"connection closed",
"connection aborted",
"tcp connect error",
]
.iter()
.any(|needle| message.contains(needle))
if let Error::Io(io_err) = err
&& let Some(status) = embedded_tonic_status(io_err)
{
return is_network_like_status(status);
}
message_has_network_needle(&err.to_string())
}
fn mark_offline_and_spawn_recovery(&self) {
@@ -1165,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 {
@@ -1188,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 {
@@ -1210,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 {
@@ -1232,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 {
@@ -1254,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 {
@@ -1278,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 {
@@ -1300,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 {
@@ -1322,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 {
@@ -1345,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 {
@@ -1367,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 {
@@ -1389,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 {
@@ -1413,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 {
@@ -1484,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 {
@@ -1505,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 {
@@ -1528,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();
@@ -1567,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 {
@@ -1590,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 {
@@ -1613,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 {
@@ -1633,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),
}
}
@@ -1662,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 {
@@ -1702,39 +1720,37 @@ fn tier_config_reload_connection_outcome(err: Error) -> TierConfigReloadOutcome
}
fn is_tier_config_reload_connection_failure(err: &Error) -> bool {
let message = err.to_string().to_ascii_lowercase();
let message = err.to_string();
// A bare "unavailable" is only trusted inside the local dial-failure
// wrapper from `get_client`, never in application text.
if message
.to_ascii_lowercase()
.split_once("can not get client, err:")
.is_some_and(|(_, local_error)| local_error.contains("unavailable"))
{
return true;
}
[
"temporarily offline",
"transport error",
"error trying to connect",
"connection refused",
"connection reset",
"connection closed",
"connection aborted",
"broken pipe",
"not connected",
"unexpected eof",
"timed out",
"deadline has elapsed",
"tcp connect error",
]
.iter()
.any(|needle| message.contains(needle))
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 {
@@ -1744,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())
}
@@ -2151,6 +2175,126 @@ mod tests {
assert!(!PeerRestClient::is_network_like_error(&Error::NotImplemented));
}
#[test]
fn peer_rest_client_network_classifier_uses_typed_status_code() {
// The one code that means "nothing is answering on this channel".
assert!(PeerRestClient::is_network_like_error(&Error::from(tonic::Status::unavailable(
"storage layer is not initialized"
))));
// Application statuses from a live peer must not mark it offline,
// even when their message contains transport-sounding words.
assert!(!PeerRestClient::is_network_like_error(&Error::from(tonic::Status::internal(
"failed to reload metadata for bucket \"unavailable-logs\""
))));
assert!(!PeerRestClient::is_network_like_error(&Error::from(tonic::Status::unauthenticated(
"No valid auth token"
))));
// A request-budget expiry answered by a live peer is an application
// outcome, not a transport failure.
assert!(!PeerRestClient::is_network_like_error(&Error::from(tonic::Status::deadline_exceeded(
"heal control request expired"
))));
// Unknown is the transport's escape hatch for a cause it could not
// map, and our handlers never return it, so there the text decides.
assert!(PeerRestClient::is_network_like_error(&Error::from(tonic::Status::unknown(
"Service was not ready: transport error"
))));
assert!(!PeerRestClient::is_network_like_error(&Error::from(tonic::Status::unknown(
"peer response unknown"
))));
}
#[test]
fn peer_rest_client_network_classifier_ignores_transport_words_in_application_statuses() {
// The reason classification reads the code rather than the text: a
// peer that answers is reachable, even when what it says describes a
// connection failure of its own. A handler interpolating a local
// io::Error into Status::internal, or relaying trouble with its own
// downstream, must not cost us the channel to a healthy peer.
for status in [
tonic::Status::internal("connection refused while dialing downstream backend"),
tonic::Status::internal("write failed: broken pipe"),
tonic::Status::unauthenticated("connection reset while validating token"),
tonic::Status::failed_precondition("scanner lease timed out"),
tonic::Status::deadline_exceeded("heal control request timed out"),
] {
let rendered = status.to_string();
assert!(
!PeerRestClient::is_network_like_error(&Error::from(status)),
"an answered application status must not mark the peer offline: {rendered}"
);
}
}
#[test]
fn peer_rest_client_network_classifier_keeps_slow_peers_online() {
// The per-RPC channel deadline (RUSTFS_INTERNODE_RPC_TIMEOUT, 30s)
// surfaces as Cancelled "Timeout expired" carrying the transport
// cause as its source. A peer that is merely slow must stay online:
// gating it would spend a full recovery cycle fast-failing every RPC
// to a host that is still answering, turning load into a partition.
let timeout_status = tonic::Status::cancelled("Timeout expired");
assert!(!PeerRestClient::is_network_like_error(&Error::from(timeout_status)));
let sourced = tonic::Status::from_error(Box::new(std::io::Error::other("Timeout expired")));
assert!(
std::error::Error::source(&sourced).is_some(),
"the transport builds this status through Status::from_error, which attaches the cause"
);
assert!(!PeerRestClient::is_network_like_error(&Error::from(sourced)));
}
#[test]
fn rpc_status_errors_keep_their_rendering_and_hide_peer_metadata() {
let err = Error::from(tonic::Status::unavailable("peer gone"));
assert_eq!(
err.to_string(),
"Io error: code: 'The service is currently unavailable', message: \"peer gone\""
);
// tonic's own Debug prints the MetadataMap, i.e. every response header
// the peer sent; those must not reach a log through this error.
let mut status = tonic::Status::unavailable("peer gone");
status
.metadata_mut()
.insert("authorization", "Bearer secret".parse().expect("valid header value"));
let rendered = format!("{:?}", Error::from(status));
assert!(!rendered.contains("Bearer secret"), "{rendered}");
assert!(!rendered.contains("MetadataMap"), "{rendered}");
}
#[test]
fn peer_rest_client_network_classifier_ignores_application_text_containing_unavailable() {
// Regression: a bare "unavailable" needle used to match application
// strings like these and take a healthy peer offline.
assert!(!PeerRestClient::is_network_like_error(&Error::other(
"peer replication statistics provider is unavailable"
)));
assert!(!PeerRestClient::is_network_like_error(&Error::other(
"bucket \"unavailable-logs\" not found"
)));
// Anchored renderings of a flattened Unavailable status still match:
// tonic >= 0.14 form ...
assert!(PeerRestClient::is_network_like_error(&Error::other(
"peer tier mutation commit RPC failed: code: 'The service is currently unavailable', message: \"peer gone\""
)));
// ... which is only anchored as long as tonic renders Unavailable this
// way. A tonic bump that reworded it leaves the typed path correct but
// this needle stale, so pin the coupling rather than discover it in a
// partition.
assert!(
tonic::Status::unavailable("peer gone")
.to_string()
.to_ascii_lowercase()
.contains("code: 'the service is currently unavailable'"),
"tonic reworded Code::Unavailable; update the anchored needle"
);
// ... and the tonic <= 0.13 form peers may relay in error_info.
assert!(PeerRestClient::is_network_like_error(&Error::other(
"peer tier mutation commit RPC failed: status: Unavailable, message: \"peer gone\""
)));
}
#[test]
fn tier_config_reload_outcome_keeps_tonic_and_remote_errors_typed() {
assert!(matches!(
@@ -2177,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())),
@@ -2193,6 +2340,58 @@ mod tests {
tier_config_reload_connection_outcome(Error::other("can not get client, err: connection unavailable")),
TierConfigReloadOutcome::TransientReconnect(_)
));
// The bare word is trusted only to the right of the dial-failure
// prefix, not anywhere in the message.
assert!(matches!(
tier_config_reload_connection_outcome(Error::other(
"bucket unavailable-logs rejected it, then: can not get client, err: some other reason"
)),
TierConfigReloadOutcome::Terminal(_)
));
}
/// 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]
@@ -2495,6 +2694,39 @@ mod tests {
assert!(!client.offline.load(Ordering::Acquire));
}
#[tokio::test]
async fn peer_rest_client_finalize_result_keeps_online_for_app_errors_mentioning_unavailable() {
// Regression: application error text containing "unavailable" (a
// remote error_info payload, or a bucket named "unavailable-logs" in
// a typed application status) must not take a healthy peer offline.
let client = test_peer_client();
let err = client
.finalize_result::<()>(Err(Error::other("peer replication statistics provider is unavailable")))
.await
.expect_err("application error should still be returned");
assert!(err.to_string().contains("provider is unavailable"));
assert!(!client.offline.load(Ordering::Acquire));
let err = client
.finalize_result::<()>(Err(Error::from(tonic::Status::internal(
"failed to reload metadata for bucket \"unavailable-logs\"",
))))
.await
.expect_err("application status should still be returned");
assert!(err.to_string().contains("unavailable-logs"));
assert!(!client.offline.load(Ordering::Acquire));
}
#[tokio::test]
async fn peer_rest_client_finalize_result_marks_offline_for_typed_unavailable_status() {
let client = test_peer_client();
client
.finalize_result::<()>(Err(Error::from(tonic::Status::unavailable("storage layer is not initialized"))))
.await
.expect_err("network error should still be returned");
assert!(client.offline.load(Ordering::Acquire));
}
#[tokio::test(flavor = "current_thread")]
async fn peer_rest_recovery_probe_logs_keep_request_id_span_context() {
let logs = CapturedLogs::default();
@@ -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 {
@@ -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;
+46 -1
View File
@@ -337,9 +337,54 @@ impl From<DiskError> for std::io::Error {
}
}
/// The single in-band representation of a failed internode RPC: it carries the
/// typed `tonic::Status` so failure classifiers can read the gRPC code instead
/// of substring-matching the rendered message (see `is_network_like_status`).
/// Both `DiskError` and `StorageError` wrap statuses in this type, so one
/// downcast recovers the status regardless of which error the status was
/// converted into first.
pub(crate) struct RpcStatusError(tonic::Status);
impl RpcStatusError {
pub(crate) fn status(&self) -> &tonic::Status {
&self.0
}
}
impl From<tonic::Status> for RpcStatusError {
fn from(status: tonic::Status) -> Self {
Self(status)
}
}
impl std::fmt::Display for RpcStatusError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
std::fmt::Display::fmt(&self.0, f)
}
}
/// `tonic::Status`'s own `Debug` prints its `MetadataMap`, i.e. every response
/// header and trailer the peer sent. Those are remote-controlled and can carry
/// credentials injected by a proxy in front of the peer, so keep them out of
/// anything that reaches a log.
impl std::fmt::Debug for RpcStatusError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("RpcStatusError")
.field("code", &self.0.code())
.field("message", &self.0.message())
.finish()
}
}
impl StdError for RpcStatusError {
fn source(&self) -> Option<&(dyn StdError + 'static)> {
Some(&self.0)
}
}
impl From<tonic::Status> for DiskError {
fn from(e: tonic::Status) -> Self {
DiskError::other(e.message().to_string())
DiskError::Io(io::Error::other(RpcStatusError(e)))
}
}
+5 -1
View File
@@ -818,7 +818,11 @@ impl From<s3s::xml::DeError> for Error {
impl From<tonic::Status> for Error {
fn from(e: tonic::Status) -> Self {
Error::other(e.to_string())
// Keep the typed status as the io::Error payload instead of a
// flattened string so RPC failure classifiers can read the gRPC code
// via downcast (see `PeerRestClient::is_network_like_error`).
// `RpcStatusError` renders exactly as `e.to_string()` did.
Error::other(crate::disk::error::RpcStatusError::from(e))
}
}
@@ -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();
+6 -328
View File
@@ -104,7 +104,7 @@ use crate::{
disk::{
CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskOption, DiskStore, FileInfoVersions,
RUSTFS_META_BUCKET, RUSTFS_META_MULTIPART_BUCKET, RUSTFS_META_TMP_BUCKET, ReadMultipleReq, ReadMultipleResp, ReadOptions,
SnapshotLeaseToken, UpdateMetadataOpts, endpoint::Endpoint, error::DiskError, format::FormatV3, new_disk,
UpdateMetadataOpts, endpoint::Endpoint, error::DiskError, format::FormatV3, new_disk,
},
error::{StorageError, to_object_err},
object_api::{GetObjectReader, ObjectInfo, PutObjReader},
@@ -116,7 +116,6 @@ use bytes::Bytes;
use bytesize::ByteSize;
use chrono::Utc;
use futures::future::join_all;
use futures::task::AtomicWaker;
use glob::Pattern;
use http::HeaderMap;
use md5::{Digest as Md5Digest, Md5};
@@ -165,7 +164,7 @@ use std::hash::{BuildHasher, Hash, Hasher};
use std::mem::{self};
use std::pin::Pin;
use std::sync::OnceLock;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::atomic::{AtomicU64, Ordering};
use std::task::{Context, Poll};
use std::time::{Instant, SystemTime, UNIX_EPOCH};
use std::{
@@ -360,163 +359,6 @@ struct SetDiskLockGuardedReader {
guard: Option<ObjectLockDiagGuard>,
}
#[derive(Clone)]
struct SnapshotLease {
disk: DiskStore,
volume: String,
path: String,
token: SnapshotLeaseToken,
}
struct SnapshotLeaseState {
leases: parking_lot::Mutex<Vec<SnapshotLease>>,
renewal_failed: AtomicBool,
renewal_waker: AtomicWaker,
cancel: CancellationToken,
runtime: tokio::runtime::Handle,
}
impl Drop for SnapshotLeaseState {
fn drop(&mut self) {
self.cancel.cancel();
let leases = mem::take(&mut *self.leases.lock());
if leases.is_empty() {
return;
}
self.runtime.spawn(async move {
join_all(leases.into_iter().map(|lease| async move {
lease
.disk
.release_snapshot_lease(&lease.volume, &lease.path, lease.token)
.await
}))
.await;
});
}
}
#[derive(Clone)]
struct SnapshotLeaseHandle(Arc<SnapshotLeaseState>);
impl SnapshotLeaseHandle {
fn new(leases: Vec<SnapshotLease>) -> Self {
let state = Arc::new(SnapshotLeaseState {
leases: parking_lot::Mutex::new(leases),
renewal_failed: AtomicBool::new(false),
renewal_waker: AtomicWaker::new(),
cancel: CancellationToken::new(),
runtime: tokio::runtime::Handle::current(),
});
let weak = Arc::downgrade(&state);
let cancel = state.cancel.clone();
tokio::spawn(async move {
let renew_interval = crate::cluster::rpc::remote_disk::REMOTE_SNAPSHOT_LEASE_TTL / 3;
loop {
tokio::select! {
_ = cancel.cancelled() => return,
_ = tokio::time::sleep(renew_interval) => {}
}
let Some(state) = weak.upgrade() else {
return;
};
let leases = state.leases.lock().clone();
let results = join_all(
leases
.iter()
.map(|lease| lease.disk.renew_snapshot_lease(&lease.volume, &lease.path, lease.token)),
)
.await;
let mut current = state.leases.lock();
let mut failed = false;
for (lease, result) in current.iter_mut().zip(results) {
match result {
Ok(token) => lease.token = token,
Err(_) => failed = true,
}
}
drop(current);
if failed {
state.renewal_failed.store(true, Ordering::Release);
state.renewal_waker.wake();
return;
}
}
});
Self(state)
}
fn renewal_failed(&self) -> bool {
self.0.renewal_failed.load(Ordering::Acquire)
}
}
struct SnapshotLeaseReader {
inner: Box<dyn AsyncRead + Unpin + Send + Sync>,
lease: Option<SnapshotLeaseHandle>,
}
impl AsyncRead for SnapshotLeaseReader {
fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
let Some(lease) = self.lease.as_ref() else {
return Pin::new(&mut self.inner).poll_read(cx, buf);
};
if lease.renewal_failed() {
self.lease.take();
return Poll::Ready(Err(std::io::Error::other("snapshot lease renewal failed")));
}
lease.0.renewal_waker.register(cx.waker());
if lease.renewal_failed() {
self.lease.take();
return Poll::Ready(Err(std::io::Error::other("snapshot lease renewal failed")));
}
let filled_before = buf.filled().len();
let poll = Pin::new(&mut self.inner).poll_read(cx, buf);
if matches!(poll, Poll::Ready(Err(_)))
|| matches!(poll, Poll::Ready(Ok(())) if buf.filled().len() == filled_before && buf.remaining() > 0)
{
self.lease.take();
}
poll
}
}
async fn acquire_snapshot_leases(
disks: &[Option<DiskStore>],
volume: &str,
path: &str,
read_quorum: usize,
) -> Option<SnapshotLeaseHandle> {
let candidates = disks.iter().cloned().collect::<Option<Vec<_>>>()?;
if candidates.len() < read_quorum {
return None;
}
let results = join_all(candidates.iter().map(|disk| disk.acquire_snapshot_lease(volume, path))).await;
let mut leases = Vec::with_capacity(candidates.len());
let mut failed = false;
for (disk, result) in candidates.into_iter().zip(results) {
match result {
Ok(token) => leases.push(SnapshotLease {
disk,
volume: volume.to_string(),
path: path.to_string(),
token,
}),
Err(_) => failed = true,
}
}
if failed || leases.len() < read_quorum {
join_all(leases.into_iter().map(|lease| async move {
lease
.disk
.release_snapshot_lease(&lease.volume, &lease.path, lease.token)
.await
}))
.await;
return None;
}
Some(SnapshotLeaseHandle::new(leases))
}
impl AsyncRead for SetDiskLockGuardedReader {
fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
let had_capacity = buf.remaining() > 0;
@@ -532,24 +374,15 @@ impl AsyncRead for SetDiskLockGuardedReader {
fn finish_set_disk_read_lock(
mut reader: GetObjectReader,
read_lock_guard: Option<ObjectLockDiagGuard>,
snapshot_lease: Option<SnapshotLeaseHandle>,
lock_optimization_enabled: bool,
bucket: &str,
object: &str,
) -> GetObjectReader {
if reader.buffered_body.is_some() {
if lock_optimization_enabled || reader.buffered_body.is_some() {
release_materialized_read_lock(bucket, object, read_lock_guard);
return reader;
}
if let Some(lease) = snapshot_lease {
release_materialized_read_lock(bucket, object, read_lock_guard);
reader.stream = Box::new(SnapshotLeaseReader {
inner: reader.stream,
lease: Some(lease),
});
return reader;
}
if let Some(guard) = read_lock_guard {
reader.stream = Box::new(SetDiskLockGuardedReader {
inner: reader.stream,
@@ -1191,9 +1024,8 @@ pub fn get_object_lock_diag_slow_hold_threshold() -> Duration {
}
/// Check if lock optimization is enabled.
/// Fully materialized reads release the read lock before returning. Streaming
/// reads may replace it with data-directory snapshot leases when every
/// candidate disk supports the lease protocol.
/// When enabled, fully materialized reads may release the read lock before
/// returning to the caller. Streaming reads keep the lock until EOF or drop.
///
/// **Note**: Cached via `OnceLock` in production — env var changes require
/// process restart. In test builds the env var is read directly so that
@@ -5643,88 +5475,6 @@ mod tests {
(dir, disk)
}
#[tokio::test]
async fn snapshot_lease_acquisition_is_all_or_nothing() {
let (_dir1, disk1) = make_single_local_disk().await;
let (_dir2, disk2) = make_single_local_disk().await;
let bucket = "snapshot-lease-acquire";
let data_dir = "object/11111111-1111-1111-1111-111111111111";
let part = format!("{data_dir}/part.1");
disk1.make_volume(bucket).await.expect("first volume should be created");
disk2.make_volume(bucket).await.expect("second volume should be created");
disk1
.write_all(bucket, &part, Bytes::from_static(b"shard"))
.await
.expect("first shard should be written");
let lease = acquire_snapshot_leases(&[Some(disk1.clone()), Some(disk2)], bucket, data_dir, 1).await;
assert!(lease.is_none(), "one candidate missing snapshot data must retain the namespace lock");
assert_eq!(
disk1
.delete_data_dir(
bucket,
data_dir,
DeleteOptions {
recursive: true,
..Default::default()
},
)
.await
.expect("released partial lease should not defer cleanup"),
crate::disk::DataDirDeleteStatus::Deleted
);
}
#[tokio::test]
async fn snapshot_lease_acquisition_rejects_unavailable_candidate() {
let (_dir, disk) = make_single_local_disk().await;
let bucket = "snapshot-lease-unavailable";
let data_dir = "object/11111111-1111-1111-1111-111111111111";
let part = format!("{data_dir}/part.1");
disk.make_volume(bucket).await.expect("volume should be created");
disk.write_all(bucket, &part, Bytes::from_static(b"shard"))
.await
.expect("shard should be written");
let lease = acquire_snapshot_leases(&[Some(disk.clone()), None], bucket, data_dir, 1).await;
assert!(lease.is_none(), "one unavailable candidate must retain the namespace lock");
assert_eq!(
disk.delete_data_dir(
bucket,
data_dir,
DeleteOptions {
recursive: true,
..Default::default()
},
)
.await
.expect("unavailable candidate fallback must not leave a lease"),
crate::disk::DataDirDeleteStatus::Deleted
);
}
#[tokio::test]
async fn snapshot_lease_reader_fails_when_renewal_fails() {
let state = Arc::new(SnapshotLeaseState {
leases: parking_lot::Mutex::new(Vec::new()),
renewal_failed: AtomicBool::new(true),
renewal_waker: AtomicWaker::new(),
cancel: CancellationToken::new(),
runtime: tokio::runtime::Handle::current(),
});
let mut reader = SnapshotLeaseReader {
inner: Box::new(Cursor::new(Bytes::from_static(b"payload"))),
lease: Some(SnapshotLeaseHandle(state)),
};
let mut body = Vec::new();
let err = reader
.read_to_end(&mut body)
.await
.expect_err("renewal failure must terminate the body");
assert_eq!(err.kind(), std::io::ErrorKind::Other);
assert!(body.is_empty());
}
async fn make_set_disks_with(disks: Vec<Option<DiskStore>>) -> Arc<SetDisks> {
let drive_count = disks.len();
let endpoints = (0..drive_count)
@@ -9301,78 +9051,6 @@ mod tests {
);
}
#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn streaming_get_snapshot_survives_concurrent_overwrite() {
temp_env::async_with_vars([(rustfs_config::ENV_OBJECT_LOCK_OPTIMIZATION_ENABLE, Some("true"))], async {
let set_disks = make_local_bucket_test_set_disks().await;
let bucket = "snapshot-streaming-overwrite";
let object = "object";
let old_body = vec![0x41; 2 * 1024 * 1024];
let new_body = vec![0x42; old_body.len()];
let opts = ObjectOptions::default();
set_disks
.make_bucket(bucket, &MakeBucketOptions::default())
.await
.expect("bucket should be created");
let mut old_reader = PutObjReader::from_vec(old_body.clone());
set_disks
.put_object(bucket, object, &mut old_reader, &opts)
.await
.expect("old object should be written");
let mut snapshot = set_disks
.get_object_reader(bucket, object, None, HeaderMap::new(), &opts)
.await
.expect("snapshot reader should open");
let overwrite_set = Arc::clone(&set_disks);
let overwrite_body = new_body.clone();
let overwrite_opts = opts.clone();
let overwrite = tokio::spawn(async move {
let mut reader = PutObjReader::from_vec(overwrite_body);
overwrite_set.put_object(bucket, object, &mut reader, &overwrite_opts).await
});
tokio::time::timeout(Duration::from_secs(5), overwrite)
.await
.expect("overwrite should not wait for the response body")
.expect("overwrite task should join")
.expect("overwrite should succeed");
let mut restored = Vec::new();
snapshot
.stream
.read_to_end(&mut restored)
.await
.expect("leased snapshot should remain readable");
assert_eq!(restored, old_body);
let mut latest = set_disks
.get_object_reader(bucket, object, None, HeaderMap::new(), &opts)
.await
.expect("latest reader should open");
let mut latest_body = Vec::new();
latest
.stream
.read_to_end(&mut latest_body)
.await
.expect("latest object should remain readable");
assert_eq!(latest_body, new_body);
let cancelled = set_disks
.get_object_reader(bucket, object, None, HeaderMap::new(), &opts)
.await
.expect("cancelled reader should open");
drop(cancelled);
let mut replacement = PutObjReader::from_vec(vec![0x43; 2 * 1024 * 1024]);
tokio::time::timeout(Duration::from_secs(5), set_disks.put_object(bucket, object, &mut replacement, &opts))
.await
.expect("reader drop must release its snapshot")
.expect("replacement after cancellation should succeed");
})
.await;
}
#[tokio::test]
async fn set_level_batched_large_put_get_restores_body() {
const BATCHED_LARGE_SIZE: usize = 64 * 1024 * 1024;
+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
+97 -35
View File
@@ -550,7 +550,13 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
&self.ctx.tier_config_mgr(),
)
.await?;
return Ok(finish_set_disk_read_lock(gr, read_lock_guard.take(), None, bucket, object));
return Ok(finish_set_disk_read_lock(
gr,
read_lock_guard.take(),
lock_optimization_enabled,
bucket,
object,
));
}
// App-layer object data cache probe: metadata (etag/size) is resolved
@@ -677,18 +683,6 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
return Ok(reader);
}
let snapshot_lease = if lock_optimization_enabled {
match fi.data_dir.filter(|data_dir| !data_dir.is_nil()) {
Some(data_dir) => {
let data_dir_path = format!("{object}/{data_dir}");
acquire_snapshot_leases(&disks, bucket, &data_dir_path, fi.erasure.data_blocks).await
}
None => None,
}
} else {
None
};
match codec_streaming_gate.decision {
GetCodecStreamingDecision::Use => {
match Self::get_object_decode_reader_with_fileinfo(
@@ -717,7 +711,13 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
// Carry the hook probe result so the app layer skips its
// now-redundant lookup on the streaming miss path (ODC-16).
reader.body_source = body_source;
return Ok(finish_set_disk_read_lock(reader, read_lock_guard.take(), snapshot_lease, bucket, object));
return Ok(finish_set_disk_read_lock(
reader,
read_lock_guard.take(),
lock_optimization_enabled,
bucket,
object,
));
}
core::io_primitives::GetCodecStreamingReaderBuildOutcome::Fallback(reason) => {
record_get_codec_streaming_gate_decision(
@@ -756,21 +756,15 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
let set_index = self.set_index;
let pool_index = self.pool_index;
let skip_verify = opts.skip_verify_bitrot;
let producer_snapshot_lease = snapshot_lease.clone();
if let Some(lease) = snapshot_lease {
if lock_optimization_enabled {
release_materialized_read_lock(&bucket, &object, read_lock_guard.take());
reader.stream = Box::new(SnapshotLeaseReader {
inner: reader.stream,
lease: Some(lease),
});
debug!(bucket, object, "Lock optimization: replaced read lock with snapshot leases");
debug!(bucket, object, "Lock optimization: released read lock before streaming read");
}
// The producer shares the lease lifetime with the body so cancellation
// cannot release the snapshot while the duplex task is still unwinding.
// When lock optimization is disabled, keep the read-lock guard in the
// task so it lives for the duration of the streaming read.
tokio::spawn(async move {
let _guard = read_lock_guard;
let _snapshot_lease = producer_snapshot_lease;
let mut writer = GetObjectDownstreamWriter::new(wd);
// Do not wrap the entire read+write pipeline in `disk_read_timeout`.
// `get_object_with_fileinfo` also waits on `writer`, so an outer timeout
@@ -2321,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)]
@@ -2575,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;
@@ -2614,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;
+59
View File
@@ -356,6 +356,14 @@ impl FileMeta {
.versions
.partition_point(|existing| existing.header.sorts_before(&new_shallow.header));
self.versions.insert(insert_pos, new_shallow);
// `partition_point` only returns the canonical slot when `versions` is
// already canonically ordered, and nothing establishes that: `FileMeta::load`
// replays the on-disk order verbatim, and metadata written before the
// canonical-order fix ordered equal-`mod_time` ties by insertion rather than
// by `sorts_before`. Re-sort so an insert leaves the list canonical either
// way, matching what `set_idx` already does on the replace path. Cheap: after
// a correctly placed insert the slice is a single sorted run.
self.sort_by_mod_time();
Ok(())
// if !ver.valid() {
@@ -1215,6 +1223,57 @@ mod test {
assert!(fm.versions[0].header.sorts_before(&fm.versions[1].header));
}
/// `add_version_filemata` positions an inserted version with
/// `partition_point(sorts_before)`, which only yields the canonical slot when
/// `versions` is already canonically ordered. Nothing establishes that
/// precondition on load: `FileMeta::unmarshal_msg` pushes versions in file
/// order, and metadata written before the canonical-order fix ordered
/// equal-`mod_time` ties by insertion instead of by `sorts_before`. An insert
/// into such a list must still leave it canonical, the same way `set_idx`
/// already re-sorts on the replace path.
#[test]
fn add_version_filemata_canonicalizes_versions_loaded_out_of_order() {
let mod_time = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("valid test timestamp");
let first = version_for_ordering(VersionType::Object, Uuid::from_u128(70), mod_time, 70);
let second = version_for_ordering(VersionType::Object, Uuid::from_u128(80), mod_time, 80);
let (early, late) = if first.header().sorts_before(&second.header()) {
(first, second)
} else {
(second, first)
};
// Persist the pair the wrong way round to emulate a pre-canonical-order
// xl.meta, bypassing add_version_filemata so the bad order really lands on
// the wire.
let mut legacy = FileMeta::new();
legacy
.versions
.push(FileMetaShallowVersion::try_from(late).expect("shallow later version"));
legacy
.versions
.push(FileMetaShallowVersion::try_from(early).expect("shallow earlier version"));
let mut loaded =
FileMeta::load(&legacy.marshal_msg().expect("serialize legacy metadata")).expect("reload legacy metadata");
assert!(
!loaded.versions[0].header.sorts_before(&loaded.versions[1].header),
"fixture must reach add_version_filemata non-canonically ordered"
);
loaded
.add_version_filemata(version_for_ordering(VersionType::Delete, Uuid::from_u128(90), mod_time, 90))
.expect("insert equal-time delete marker");
assert_eq!(loaded.versions.len(), 3);
assert!(
loaded
.versions
.windows(2)
.all(|pair| pair[0].header.sorts_before(&pair[1].header)),
"insert must leave the version list canonically ordered"
);
}
/// Regression for backlog#799 B16: replication reset state must be
/// persisted under both internal prefixes, never as a bare ARN. A bare-ARN
/// key (produced by `ObjectInfo::replication_state`) has no internal prefix,
+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
+362
View File
@@ -447,10 +447,19 @@ 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());
}
@@ -465,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
@@ -1136,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::{
+88 -88
View File
@@ -988,107 +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 AcquireSnapshotLease(SnapshotLeaseRequest) returns (SnapshotLeaseResponse) {};
rpc RenewSnapshotLease(SnapshotLeaseRenewRequest) returns (SnapshotLeaseResponse) {};
rpc ReleaseSnapshotLease(SnapshotLeaseReleaseRequest) returns (SnapshotLeaseMutationResponse) {};
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();
@@ -19,6 +19,7 @@ for later deletion.
- `disk-mutation-body-digest` internode mutating disk RPCs: servers temporarily accept mutating disk RPCs (RenameData, DeleteVersion, DeleteVersions, WriteMetadata, UpdateMetadata, WriteAll, Delete, DeletePaths, RenameFile, RenamePart, DeleteVolume, MakeVolume, MakeVolumes) that carry no signature-bound canonical body digest, so peers from releases that predate body-digest signing remain available during rolling upgrades. Accepted digestless mutations increment the internode body-digest fallback counter; that counter must read zero fleet-wide across a release window before RUSTFS_INTERNODE_RPC_BODY_DIGEST_STRICT is enabled. Because body-bound requests now consume replay-cache nonces on the receiver, deploy the raised RUSTFS_INTERNODE_RPC_REPLAY_CACHE_CAPACITY default fleet-wide before enabling strict mode, and watch the internode replay-cache overflow counter for undersized capacity during the rollout. Remove the digestless fallback after the minimum supported RustFS peer version body-binds every mutating disk RPC.
- `heal-status-rpc-v1` node heal status capability: new peers treat an unimplemented BackgroundHealStatus RPC as an explicitly incomplete rolling-upgrade response. Remove the fallback after the minimum supported RustFS peer version implements BackgroundHealStatus.
- `backlog-1316` legacy encrypted multipart range seek: the feature remains opt-in until every server that can initiate, write, or complete multipart uploads supports the candidate-to-final marker protocol and uploadId commit lock, and pre-upgrade multipart uploads have drained. Remove the RUSTFS_ENCRYPTED_RANGE_SEEK switch after the minimum supported release does so; keep the quorum marker and malformed-layout full-read guards permanently.
- `tonic-013-status-render` peer RPC failure classification: internode failures that reach a node only as text (a peer's error_info payload, a status flattened through format!) are classified by matching the rendering of an Unavailable gRPC status. Releases up to 1.0.0-alpha.38 shipped tonic 0.13, which rendered that status as "status: Unavailable, message: ..."; tonic 0.14 renders it as "code: 'The service is currently unavailable', message: ...". Both forms are matched so an older peer's relayed text still marks an unreachable peer offline. Remove the tonic 0.13 form after the minimum supported RustFS peer version ships tonic 0.14 or later.
- `rustfs-5063` pre-beta.9 Local KMS recovery: persisted Local KMS configs from beta.8 and earlier predate the explicit insecure-development flag, and encrypted key files use the legacy SHA-256 KDF. Remove the config fallback after supported upgrades have rewritten or explicitly resaved all pre-beta.9 configs with the development-default field, and remove the legacy KDF after supported upgrades have rewritten all pre-beta.9 Local KMS key files with explicit at-rest protection.
- `sse-local-dek-json-v1` legacy local SSE DEK decoding: releases before the JSON envelope wrote wrapped DEKs as `base64(nonce):base64(ciphertext)`, so readers retain that decoder while all new writes use the versioned JSON envelope. Remove the colon decoder after the minimum supported direct-upgrade release writes JSON envelopes and migration tooling has rewritten every retained legacy object.
+1
View File
@@ -47,6 +47,7 @@ io-scheduler-debug = [] # Enable debug information in I/O scheduler
tracing-chunk-debug = [] # Enable per-chunk tracing in data plane (high noise, for debugging only)
full = ["metrics-gpu", "ftps", "swift", "webdav", "sftp", "pyroscope"]
manual-test-runners = []
e2e-test-hooks = []
rio-v2 = ["rustfs-ecstore/rio-v2"]
pyroscope = ["rustfs-obs/pyroscope"]
# Tokio runtime telemetry. Requires `--cfg tokio_unstable`; use `make build-profiling`.
@@ -58,6 +58,8 @@ const LOG_SUBSYSTEM_ILM_TRANSITION: &str = "ilm_transition";
const EVENT_ADMIN_ILM_TRANSITION_STATE: &str = "admin_ilm_transition_state";
static ACTIVE_MANUAL_TRANSITION_SCOPES: OnceLock<Mutex<Vec<ManualTransitionRunScope>>> = OnceLock::new();
#[cfg(feature = "e2e-test-hooks")]
const E2E_MANUAL_TRANSITION_CANCEL_BARRIER_ENV: &str = "RUSTFS_E2E_MANUAL_TRANSITION_CANCEL_BARRIER";
static ACTIVE_MANUAL_TRANSITION_JOBS: OnceLock<Mutex<HashMap<Uuid, CancellationToken>>> = OnceLock::new();
static MANUAL_TRANSITION_OWNER_ID: OnceLock<String> = OnceLock::new();
@@ -748,6 +750,10 @@ async fn start_manual_transition_job(
let job_heartbeat_shutdown_token = heartbeat_shutdown_token.clone();
spawn_manual_transition_job_heartbeat(store, job_id, scan_cancel_token, heartbeat_shutdown_token);
tokio::spawn(async move {
#[cfg(feature = "e2e-test-hooks")]
if std::env::var_os(E2E_MANUAL_TRANSITION_CANCEL_BARRIER_ENV).is_some() {
job_scan_cancel_token.cancelled().await;
}
let result = enqueue_transition_for_existing_objects_scoped(run_store.clone(), &bucket, run_options).await;
if let Some(final_record) = finalize_manual_transition_job(run_store.clone(), job_id, result).await
&& final_record.is_terminal()
+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,
};
+312 -34
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!(
@@ -890,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
}
@@ -898,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
}
@@ -906,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
}
@@ -1195,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
}
@@ -1214,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
}
@@ -1221,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
}
@@ -1340,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
}
@@ -1347,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() {
@@ -1381,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() {
@@ -1413,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() {
@@ -1448,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() {
@@ -1480,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() {
@@ -1515,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;
@@ -1552,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() {
@@ -1583,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() {
@@ -1614,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,
@@ -1635,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,
@@ -1835,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,
@@ -1862,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,
@@ -1882,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);
@@ -1927,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,
@@ -1958,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,
@@ -1990,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,
@@ -2020,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,
@@ -2072,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,
@@ -2095,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]
@@ -2485,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
@@ -2603,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 {
@@ -2682,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]
@@ -4439,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() {
@@ -4749,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();
@@ -4813,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);
}
+7 -6
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>;
@@ -1365,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()
@@ -1441,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
}