Compare commits

...

8 Commits

Author SHA1 Message Date
Hauser 9601a7c4bc fix(admin): protect notification target edits (#9048)
* fix(admin): protect notification target edits

Use a durable configuration revision and conditional mutation to preserve concurrent changes and inherited target settings.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(admin): use the gateway error boundary

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* refactor(admin): limit target detail health probes

Reuse normalized source information and probe only runtime targets matching the requested instance.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* refactor(admin): streamline target config validation

Preserve duplicate-key order when comparing snapshots, avoid copying secret values, and reuse the validated target configuration.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>
2026-10-29 04:07:12 +08:00
Hauser d16e3cf7b6 perf(ecstore): streamline local GET batch reads (#9043)
* perf(ecstore): avoid eager parity reads for demand-bound GET

Reuse the existing batch setup and quorum targets to retain deferred parity
without materializing bytes that demand-bound reads immediately discard.
Keep legacy all-shard batching and verify degraded and ranged recovery.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* perf(ecstore): validate batch reads on the opened shard

Avoid repeated pathname traversal by reading metadata from the descriptor
used for positioned I/O. Reject nonregular shards without blocking and
preserve range, metrics, and cancellation behavior.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>
2026-10-29 02:05:01 +08:00
Chris An 3e056ffb4d fix: keep incomplete scanner measurements unknown (#9044)
* fix: keep missing scanner error rates unknown

* fix: keep incomplete pressure samples unknown
2026-10-29 01:23:44 +08:00
Chris An 72b4f6182f test(heal): keep rename-tail repair out of isolated root heal (#9041) 2026-10-29 00:22:19 +08:00
Hauser a9d3022f37 fix(ecstore): prove multipart part sizes before cleanup (#9037)
* fix(ecstore): prove multipart part sizes before cleanup

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* refactor(ecstore): merge strict multipart observation passes

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* test: respect lifecycle architecture boundaries

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* test: fix multipart E2E clippy warnings

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>
2026-10-29 00:14:32 +08:00
Chris An c232f1552d fix(metrics): reject missing histogram counts (#9039) 2026-10-28 22:25:09 +08:00
RustFS 7581a70412 feat(s3): serve Veeam SOSAPI system and capacity XML (#9007)
* feat(s3): serve Veeam SOSAPI system and capacity XML

Authenticated GET and HEAD of the two well-known SOSAPI keys return
generated XML, and read preconditions use that generated ETag and
Last-Modified. Mutations of those keys are rejected while the feature
is on.

Co-authored-by: RustFS <hello@rustfs.com>

* fix(s3): simplify Veeam quota metadata lookup

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(s3): harden Veeam SOSAPI capacity and reads

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(s3): bind Veeam capacity to bucket identity

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(s3): protect virtual objects and coalesce refreshes

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* test(e2e): refresh Veeam profile selections

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* test(e2e): bound pagination fixture writes

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

---------

Co-authored-by: Hauser <housemecn@gmail.com>
Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>
2026-10-28 22:04:49 +08:00
Ramakrishna Chilaka 8cb9296f20 perf(ecstore): reuse bucket incarnation fences across copy and metadata paths (#9034)
Eliminate redundant .bucket-incarnation/xl.meta quorum reads and distributed
bucket-metadata transaction read locks under held lifecycle read locks and
Object Lock configuration snapshots across put_object_metadata, CopyObject,
UploadPartCopy, RestoreObject, read_object_versions_for_lifecycle, and
get_bucket_incarnation_id_for_options_in.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: d85aaf47-2a4b-4c2c-97a3-dbe8c7717e2a
2026-10-28 21:03:46 +08:00
75 changed files with 6277 additions and 433 deletions
+2 -2
View File
@@ -1,2 +1,2 @@
sha256-darwin=a67a347f26be186c598b5beaf68e6bcbe1855350a35e1423dd5b8a81bcba06a5
sha256-linux=ad4fabe811c6cca1aba89fad34ccd404b5832a6a955875c2826f6e0fa1ea4f1f
sha256-darwin=67251cb7d9978d38e899268cac4d7898297ff7e5b5c48c9c1546dc760b93568d
sha256-linux=673da76c2e5710c4622fe2fa32d2b297730895769487fcf1152ee1c2366f32f4
+1 -1
View File
@@ -1 +1 @@
sha256=f065e7071aac9d48a7c1b301f328fa7cc14f11c4e7dc46da3e55057c4ea29bd5
sha256=e9dc896774122a304e2b8b90f28c82d4d4e0af8ded36a82b147b30945ec994e3
+1
View File
@@ -100,6 +100,7 @@ when restoring `/rustfs/console` for a custom OEM build.
## S3 API environment variables
- `RUSTFS_API_OBJECT_MAX_VERSIONS` caps the number of retained versions for a single object. It defaults to `9223372036854775807`, matching MinIO's practical-unlimited default. Set a positive integer to enforce a lower per-object metadata bound.
- `RUSTFS_VEEAM_SOSAPI_ENABLE` serves Veeam SOSAPI `system.xml` and `capacity.xml` for an authenticated GET or HEAD of those exact bucket-root keys. It defaults to `true`. Explicit version reads use the normal object path. Capacity requests return `ServiceUnavailable` until the scanner publishes a complete snapshot certified for the current bucket incarnation and authoritative quota metadata can be read. Legacy or mixed-version snapshots without an incarnation proof cannot serve capacity until a compatible scanner publishes a certified snapshot. Capacity also returns `ServiceUnavailable` when cluster or disk observations are missing, stale, or degraded. Retry after scanner convergence and storage or metadata recovery. Capacity assumes shard paths have independent backing space; shared physical storage can make the estimate optimistic. Set the flag to `false` to restore the normal object path, including deletion of a real object previously stored at that key.
- `MINIO_API_OBJECT_MAX_VERSIONS` is accepted as a compatibility alias when the canonical RustFS variable is not set.
## Distributed endpoint locality
+10
View File
@@ -222,6 +222,16 @@ pub const ENV_RUSTFS_TLS_PATH: &str = "RUSTFS_TLS_PATH";
/// Environment variable for server KMS enable.
pub const ENV_RUSTFS_KMS_ENABLE: &str = "RUSTFS_KMS_ENABLE";
/// Environment variable for Veeam SOSAPI virtual system objects.
///
/// When true (the default), authenticated GET and HEAD of the two Veeam
/// SOSAPI keys return generated XML. When false, those keys use the normal
/// object path so an operator can remove a real object stored there earlier.
pub const ENV_VEEAM_SOSAPI_ENABLE: &str = "RUSTFS_VEEAM_SOSAPI_ENABLE";
/// Default Veeam SOSAPI enablement.
pub const DEFAULT_VEEAM_SOSAPI_ENABLE: bool = true;
/// Default KMS enable for server-side encryption
/// This is the default value for enabling KMS encryption for server-side encryption.
/// Default value: false
+31
View File
@@ -466,6 +466,9 @@ pub struct BucketTargetUsageInfo {
#[derive(Debug, Default, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct BucketUsageInfo {
pub size: u64,
/// Bucket identity certified by every set contributing to this snapshot.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub bucket_incarnation: Option<String>,
// Following five fields suffixed with V1 are here for backward compatibility
// Total Size for objects that have not yet been replicated
pub replication_pending_size_v1: u64,
@@ -2316,6 +2319,9 @@ impl BucketUsageInfo {
/// Add size summary to this bucket usage
/// Merge another BucketUsageInfo into this one
pub fn merge(&mut self, other: &BucketUsageInfo) {
if self.bucket_incarnation != other.bucket_incarnation {
self.bucket_incarnation = None;
}
self.size += other.size;
self.objects_count += other.objects_count;
self.versions_count += other.versions_count;
@@ -3299,6 +3305,31 @@ mod tests {
assert_eq!(usage1.versions_count, 15);
}
#[test]
fn bucket_usage_merge_discards_incompatible_incarnation_proofs() {
let id = "00000000-0000-4000-8000-000000000001";
let mut same = BucketUsageInfo {
bucket_incarnation: Some(id.into()),
..Default::default()
};
same.merge(&BucketUsageInfo {
bucket_incarnation: Some(id.into()),
..Default::default()
});
assert_eq!(same.bucket_incarnation.as_deref(), Some(id));
same.merge(&BucketUsageInfo::default());
assert!(same.bucket_incarnation.is_none());
let mut different = BucketUsageInfo {
bucket_incarnation: Some(id.into()),
..Default::default()
};
different.merge(&BucketUsageInfo {
bucket_incarnation: Some("00000000-0000-4000-8000-000000000002".into()),
..Default::default()
});
assert!(different.bucket_incarnation.is_none());
}
#[test]
fn size_summary_add_saturates_instead_of_overflowing() {
// The scanner folds one summary per object into a per-prefix total, so a
@@ -1129,9 +1129,198 @@ async fn phase_quorum_matrix(
Ok(())
}
async fn list_parts_disk_loss_after_metadata_read(
cluster: &RustFSTestClusterEnvironment,
proxies: &[RpcFaultProxy],
clients: &[Client],
bucket: &str,
) -> TestResult {
let key = "list-parts-after-metadata-loss";
let body = payload_for("list-parts-acknowledged", 4096);
let upload = clients[0].create_multipart_upload().bucket(bucket).key(key).send().await?;
let upload_id = upload.upload_id().ok_or("missing upload ID")?.to_owned();
let part = clients[0]
.upload_part()
.bucket(bucket)
.key(key)
.upload_id(&upload_id)
.part_number(1)
.body(ByteStream::from(body.clone()))
.send()
.await?;
let etag = part.e_tag().ok_or("missing part ETag")?.to_owned();
let path = upload_path_scope(bucket, key, &upload_id)?;
let root = Path::new(&cluster.nodes[0].data_dirs[0])
.join(RUSTFS_META_MULTIPART_BUCKET)
.join(&path);
let data_dirs = std::fs::read_dir(&root)?
.filter_map(Result::ok)
.map(|entry| entry.path())
.filter(|entry| entry.join("part.1").is_file() && entry.join("part.1.meta").is_file())
.collect::<Vec<_>>();
assert_eq!(data_dirs.len(), 1, "the coordinator's actual part directory must be identified");
let data_dir = &data_dirs[0];
let dir_name = data_dir
.file_name()
.and_then(|name| name.to_str())
.ok_or("non-UTF8 data directory")?;
let scoped_path = format!("{path}/{dir_name}");
let held_dir = data_dir.with_extension("e2e-held");
std::fs::rename(data_dir, &held_dir)?;
let pauses = proxies[1..]
.iter()
.map(|proxy| proxy.pause(&["ListDir"], usize::MAX, Some(&scoped_path)))
.collect::<Result<Vec<_>, _>>()?;
let request = clients[0].list_parts().bucket(bucket).key(key).upload_id(&upload_id);
let task = tokio::spawn(async move { request.send().await });
// ListDir is dispatched only after the upload metadata read succeeded.
// Require all three scoped remote requests; unrelated failures do not count.
wait_for_faults(&pauses, &["ListDir"]).await?;
let result = tokio::time::timeout(DEADLINE, task).await??;
for pause in &pauses {
pause.discard_pending();
}
drop(pauses);
std::fs::rename(&held_dir, data_dir)?;
let error = result.expect_err("one missing directory and three unavailable participants cannot prove an empty upload");
let response = error.raw_response().expect("ListParts must return an S3 service response");
assert_eq!(response.status().as_u16(), 503);
assert_eq!(error.as_service_error().and_then(|err| err.meta().code()), Some("SlowDownRead"));
assert!(response.headers().get("x-rustfs-error-cause").is_none());
assert!(response.headers().get("retry-after").is_none());
assert!(
response
.headers()
.get("content-type")
.is_some_and(|value| value.starts_with("application/xml"))
);
assert!(
response
.headers()
.get("x-amz-request-id")
.is_some_and(|value| !value.is_empty())
);
let listed = recover_after_fault("ListParts after directory restoration", || {
clients[0].list_parts().bucket(bucket).key(key).upload_id(&upload_id).send()
})
.await??;
assert_eq!(listed.parts().len(), 1);
assert_eq!(listed.parts()[0].e_tag(), Some(etag.as_str()));
clients[0]
.complete_multipart_upload()
.bucket(bucket)
.key(key)
.upload_id(&upload_id)
.multipart_upload(
CompletedMultipartUpload::builder()
.parts(CompletedPart::builder().part_number(1).e_tag(&etag).build())
.build(),
)
.send()
.await?;
for client in clients {
assert_object_bytes_after_fault(client, bucket, key, &body).await?;
}
eprintln!(
"MPU list proof: metadata read completed, three scoped ListDir RPCs held, missing local directory rejected; restored part and all readers verified"
);
Ok(())
}
async fn concurrent_capped_parts_respect_limit_on_distinct_nodes(clients: &[Client], bucket: &str) -> TestResult {
let key = "concurrent-capped-parts";
let upload = clients[0]
.create_multipart_upload()
.bucket(bucket)
.key(key)
.customize()
.mutate_request(|request| {
let separator = if request.uri().contains('?') { "&" } else { "?" };
let uri = format!("{}{separator}x-rustfs-max-total-object-size=10", request.uri());
request
.set_uri(uri)
.expect("the signed multipart size query must be a valid URI");
})
.send()
.await?;
let upload_id = upload.upload_id().ok_or("missing capped upload ID")?.to_owned();
let barrier = std::sync::Arc::new(tokio::sync::Barrier::new(3));
let mut tasks = Vec::new();
for (index, number, byte) in [(0_usize, 1_i32, 0x41_u8), (1, 2, 0x42)] {
let client = clients[index].clone();
let bucket = bucket.to_owned();
let upload_id = upload_id.clone();
let barrier = std::sync::Arc::clone(&barrier);
tasks.push(tokio::spawn(async move {
barrier.wait().await;
let result = client
.upload_part()
.bucket(&bucket)
.key(key)
.upload_id(&upload_id)
.part_number(number)
.body(ByteStream::from(vec![byte; 6]))
.send()
.await;
(number, byte, result)
}));
}
barrier.wait().await;
let mut winner = None;
let mut rejected = 0;
for task in tasks {
let (number, byte, result) = tokio::time::timeout(DEADLINE, task).await??;
match result {
Ok(part) => {
assert!(winner.is_none(), "both six-byte parts cannot fit the ten-byte limit");
winner = Some((number, byte, part.e_tag().ok_or("missing accepted ETag")?.to_owned()));
}
Err(error) => {
assert_eq!(
error.as_service_error().and_then(|err| err.meta().code()),
Some("EntityTooLarge"),
"{error:?}"
);
rejected += 1;
}
}
}
let (number, byte, etag) = winner.ok_or("no part was admitted")?;
assert_eq!(rejected, 1);
let listed = clients[0]
.list_parts()
.bucket(bucket)
.key(key)
.upload_id(&upload_id)
.send()
.await?;
assert_eq!(listed.parts().len(), 1);
assert_eq!(listed.parts()[0].part_number(), Some(number));
assert_eq!(listed.parts()[0].e_tag(), Some(etag.as_str()));
clients[0]
.complete_multipart_upload()
.bucket(bucket)
.key(key)
.upload_id(&upload_id)
.multipart_upload(
CompletedMultipartUpload::builder()
.parts(CompletedPart::builder().part_number(number).e_tag(&etag).build())
.build(),
)
.send()
.await?;
for client in clients {
assert_object_bytes_after_fault(client, bucket, key, &[byte; 6]).await?;
}
eprintln!("MPU size limit: two coordinators raced; exactly one six-byte part admitted under ten-byte limit");
Ok(())
}
#[tokio::test]
async fn metadata_delay_preserves_write_quorum_and_rejects_quorum_minus_one() -> TestResult {
let (cluster, proxies, clients, bucket) = fixture(1, "EC:2").await?;
list_parts_disk_loss_after_metadata_read(&cluster, &proxies, &clients, &bucket).await?;
concurrent_capped_parts_respect_limit_on_distinct_nodes(&clients, &bucket).await?;
phase_quorum_matrix(&cluster, &proxies, &clients, &bucket, false, 1, 3).await
}
@@ -1214,6 +1403,22 @@ async fn mixed_version_failed_complete_requires_current_coordinator_for_abort()
cluster.start_node(0).await?;
crate::upgrade_compatibility_test::wait_for_upgrade_write_readiness(&clients, &bucket, "mixed-mpu-current-writer", DEADLINE)
.await?;
// Read-only ListParts remains available during the documented mutation pause.
// Compare the actual old and current coordinators on the same legacy upload.
for (version, client) in [("current", &clients[0]), ("previous", &clients[1])] {
let listed = client
.list_parts()
.bucket(&bucket)
.key(key)
.upload_id(upload_id)
.send()
.await?;
assert_eq!(listed.parts().len(), 1, "{version} coordinator must list the legacy part");
assert_eq!(listed.parts()[0].part_number(), Some(1));
assert_eq!(listed.parts()[0].size(), Some(9 * 1024 * 1024));
assert_eq!(listed.parts()[0].e_tag(), part.e_tag());
}
eprintln!("MPU mixed version: old and current coordinators agree on legacy ListParts identity and size");
let scope = upload_scope(upload_id)?;
let observers = proxies[1..]
.iter()
@@ -1344,7 +1549,7 @@ async fn native_certified_partial_abort_retry_reclaims_survivors_below_metadata_
capability_sha256: Option<[u8; 32]>,
abort: Option<AbortRef>,
}
let (cluster, proxies, clients, bucket) = fixture(4, "EC:4").await?;
let (mut cluster, proxies, clients, bucket) = fixture(4, "EC:4").await?;
let key = "partial-abort";
let committed = payload_for("committed-before-abort", 4096);
clients[0]
@@ -1475,6 +1680,48 @@ async fn native_certified_partial_abort_retry_reclaims_survivors_below_metadata_
.raw_response()
.and_then(|response| response.headers().get("x-amz-request-id"))
);
assert!(
retained.values().all(|source| source.participant.0 != 3),
"the selected offline node must have no surviving root"
);
cluster.stop_node(3)?;
let offline = tokio::time::timeout(
DEADLINE,
clients[0]
.abort_multipart_upload()
.bucket(&bucket)
.key(key)
.upload_id(upload_id)
.send(),
)
.await?
.expect_err("an unavailable original participant cannot be replaced by the certificate");
assert_unavailable(&offline, "storage_write_quorum", None);
assert!(
multipart_sources(&cluster, Some(&hash))? == retained,
"an unavailable participant must not authorize deletion or change retained evidence"
);
cluster.start_node(3).await?;
wait_for_storage_peer_recovery(&proxies, &clients[0], &bucket, upload_id, 4).await?;
eprintln!("MPU certified residual: unavailable original returned storage_write_quorum; all surviving bytes retained");
// Kill the coordinator and a node owning a real surviving source. The
// retained bytes and full-N certificate must survive both process restarts
// before a fresh coordinator can authorize the retry.
let survivor_node = retained.values().next().ok_or("missing surviving source")?.participant.0;
let mut restarted = vec![0, survivor_node];
restarted.sort_unstable();
restarted.dedup();
for node in restarted {
cluster.stop_node(node)?;
cluster.start_node(node).await?;
wait_for_storage_peer_recovery(&proxies, &clients[0], &bucket, upload_id, 4).await?;
assert!(
multipart_sources(&cluster, Some(&hash))? == retained,
"restart must preserve every surviving source and certificate byte"
);
}
eprintln!("MPU certified residual: coordinator and surviving owner restarted; all retained bytes unchanged");
let delete_observers = proxies
.iter()
.map(|proxy| proxy.pause(&["RecoverMultipartSource"], usize::MAX, Some(&path)))
@@ -47,6 +47,9 @@ struct UpgradeSeed {
history_bucket: String,
history_key: &'static str,
history_body: Vec<u8>,
multipart_upload: String,
multipart_etag: String,
multipart_body: Vec<u8>,
versioned_bucket: String,
versioned_key: &'static str,
version1: String,
@@ -241,6 +244,25 @@ async fn seed_history_and_iam(dist: &DistCluster) -> TestResult<UpgradeSeed> {
let history_body = b"written by the previous 4-node release".to_vec();
put_object(&root, &history_bucket, history_key, history_body.clone()).await?;
let multipart_body = vec![0x5a; 4096];
let multipart = root
.create_multipart_upload()
.bucket(&history_bucket)
.key("legacy-multipart.bin")
.send()
.await?;
let multipart_upload = multipart.upload_id().ok_or("legacy multipart upload ID missing")?.to_owned();
let part = root
.upload_part()
.bucket(&history_bucket)
.key("legacy-multipart.bin")
.upload_id(&multipart_upload)
.part_number(1)
.body(aws_sdk_s3::primitives::ByteStream::from(multipart_body.clone()))
.send()
.await?;
let multipart_etag = part.e_tag().ok_or("legacy multipart part ETag missing")?.to_owned();
let versioned_key = "versioned-history.txt";
let version1_body = b"version-one-before-upgrade".to_vec();
let version1 = root
@@ -278,6 +300,9 @@ async fn seed_history_and_iam(dist: &DistCluster) -> TestResult<UpgradeSeed> {
history_bucket,
history_key,
history_body,
multipart_upload,
multipart_etag,
multipart_body,
versioned_bucket,
versioned_key,
version1,
@@ -304,6 +329,21 @@ async fn assert_history_and_iam(dist: &DistCluster, seed: &UpgradeSeed, context:
)
.await?;
assert_object_bytes(&root_a, &seed.history_bucket, seed.history_key, &seed.history_body).await?;
// During mixed-version operation this is read-only; mutation resumes only
// after the whole fleet is upgraded.
for client in [&root_a, &root_b] {
let listed = client
.list_parts()
.bucket(&seed.history_bucket)
.key("legacy-multipart.bin")
.upload_id(&seed.multipart_upload)
.send()
.await?;
assert_eq!(listed.parts().len(), 1, "{context}: legacy part listing");
assert_eq!(listed.parts()[0].part_number(), Some(1));
assert_eq!(listed.parts()[0].size(), Some(4096));
assert_eq!(listed.parts()[0].e_tag(), Some(seed.multipart_etag.as_str()));
}
let v1 = root_b
.get_object()
@@ -387,17 +427,146 @@ async fn assert_history_and_iam(dist: &DistCluster, seed: &UpgradeSeed, context:
Ok(())
}
async fn compare_degraded_legacy_list_parts(dist: &mut DistCluster, seed: &UpgradeSeed, previous: &Path) -> TestResult {
let decoded = String::from_utf8(base64_simd::URL_SAFE_NO_PAD.decode_to_vec(seed.multipart_upload.as_bytes())?)?;
let (_, upload_dir) = decoded.split_once('.').ok_or("legacy upload ID lacks deployment delimiter")?;
let hash = sha256_hex(format!("{}/legacy-multipart.bin", seed.history_bucket).as_bytes());
let root = format!("{hash}/{upload_dir}");
let mut held = Vec::new();
let mut original_files = Vec::new();
for (index, node) in dist.cluster.nodes.iter().enumerate() {
let path = Path::new(&node.data_dirs[0])
.join(crate::storage_api::multipart_fence_acceptance::RUSTFS_META_MULTIPART_BUCKET)
.join(&root);
let entries = std::fs::read_dir(path)?.collect::<Result<Vec<_>, _>>()?;
let dirs = entries
.into_iter()
.map(|entry| entry.path())
.filter(|entry| entry.join("part.1").is_file() && entry.join("part.1.meta").is_file())
.collect::<Vec<_>>();
assert_eq!(dirs.len(), 1, "each configured disk must hold the original part before injection");
let source_root = dirs[0].parent().ok_or("part directory has no upload root")?;
for path in [
source_root.join("xl.meta"),
dirs[0].join("part.1"),
dirs[0].join("part.1.meta"),
] {
original_files.push((path.clone(), std::fs::read(path)?));
}
if index < 3 {
held.push((dirs[0].clone(), dirs[0].with_extension("held-for-mixed-test")));
}
}
dist.cluster.stop_node(3)?;
for (original, temporary) in &held {
std::fs::rename(original, temporary)?;
}
let clients = dist.clients()?;
let current = tokio::time::timeout(
CREDENTIAL_TIMEOUT,
clients[0]
.list_parts()
.bucket(&seed.history_bucket)
.key("legacy-multipart.bin")
.upload_id(&seed.multipart_upload)
.send(),
)
.await?;
let legacy = tokio::time::timeout(
CREDENTIAL_TIMEOUT,
clients[1]
.list_parts()
.bucket(&seed.history_bucket)
.key("legacy-multipart.bin")
.upload_id(&seed.multipart_upload)
.send(),
)
.await?;
for (original, temporary) in &held {
std::fs::rename(temporary, original)?;
}
// Three online disks retain metadata quorum but have no part directory;
// the fourth disk is offline. A legacy empty response is not a proof that
// the unavailable participant has no part.
let legacy = legacy.expect("previous coordinator should expose its legacy empty-list behavior");
assert!(legacy.parts().is_empty());
let current = current.expect_err("current coordinator must not infer empty from missing directories and offline disks");
assert_eq!(current.raw_response().expect("current S3 response").status().as_u16(), 503);
assert_eq!(current.as_service_error().and_then(|error| error.meta().code()), Some("SlowDownRead"));
dist.cluster.start_node_from_binary(3, previous).await?;
wait_for_upgrade_write_readiness(&clients, &seed.history_bucket, "mixed-list-restored", CREDENTIAL_TIMEOUT).await?;
for (path, before) in original_files {
assert_eq!(
std::fs::read(&path)?,
before,
"the fault must preserve every original source file: {path:?}"
);
}
for client in [&clients[0], &clients[1]] {
let listed = client
.list_parts()
.bucket(&seed.history_bucket)
.key("legacy-multipart.bin")
.upload_id(&seed.multipart_upload)
.send()
.await?;
assert_eq!(listed.parts().len(), 1);
assert_eq!(listed.parts()[0].e_tag(), Some(seed.multipart_etag.as_str()));
}
eprintln!(
"MPU mixed degraded: one offline disk, three missing part directories; previous returned empty, current returned 503 SlowDownRead; original part restored on both"
);
Ok(())
}
async fn complete_legacy_multipart(dist: &DistCluster, seed: &UpgradeSeed) -> TestResult {
// Native legacy-source upgrade needs every original participant, not just
// the ordinary three-of-four write quorum proven by the earlier probes.
let clients = dist.clients()?;
crate::upgrade_compatibility_test::wait_for_drive_readmission(
&clients,
&seed.history_bucket,
"legacy-multipart-all-drives",
CREDENTIAL_TIMEOUT,
)
.await?;
let root = &clients[0];
root.complete_multipart_upload()
.bucket(&seed.history_bucket)
.key("legacy-multipart.bin")
.upload_id(&seed.multipart_upload)
.multipart_upload(
aws_sdk_s3::types::CompletedMultipartUpload::builder()
.parts(
aws_sdk_s3::types::CompletedPart::builder()
.part_number(1)
.e_tag(&seed.multipart_etag)
.build(),
)
.build(),
)
.send()
.await?;
for client in &clients {
assert_object_bytes(client, &seed.history_bucket, "legacy-multipart.bin", &seed.multipart_body).await?;
}
Ok(())
}
#[tokio::test]
async fn four_node_direct_upgrade_preserves_history_and_iam_credentials() -> TestResult {
init_logging();
let previous = source_binary()?;
let mut dist = DistCluster::new_stopped(DistLayout::FourNodeFourDisk).await?;
dist.cluster.set_env("RUSTFS_STORAGE_CLASS_RRS", "EC:0");
capture_upgrade_logs(&mut dist, "direct-upgrade")?;
dist.start_from_binary(&previous).await?;
let seed = seed_history_and_iam(&dist).await?;
dist.restart_with_current_binary().await?;
assert_history_and_iam(&dist, &seed, "direct").await?;
complete_legacy_multipart(&dist, &seed).await?;
Ok(())
}
@@ -419,6 +588,7 @@ async fn four_node_rolling_upgrade_preserves_history_and_iam_credentials() -> Te
init_logging();
let previous = source_binary()?;
let mut dist = DistCluster::new_stopped(DistLayout::FourNodeFourDisk).await?;
dist.cluster.set_env("RUSTFS_STORAGE_CLASS_RRS", "EC:0");
capture_upgrade_logs(&mut dist, "rolling-upgrade")?;
dist.start_from_binary(&previous).await?;
@@ -426,6 +596,7 @@ async fn four_node_rolling_upgrade_preserves_history_and_iam_credentials() -> Te
replace_node_and_wait_for_write_readiness(&mut dist, &seed, 0).await?;
assert_history_and_iam(&dist, &seed, "one-current-node").await?;
compare_degraded_legacy_list_parts(&mut dist, &seed, &previous).await?;
for node_idx in [1, 2] {
replace_node_and_wait_for_write_readiness(&mut dist, &seed, node_idx).await?;
@@ -434,5 +605,6 @@ async fn four_node_rolling_upgrade_preserves_history_and_iam_credentials() -> Te
replace_node_and_wait_for_write_readiness(&mut dist, &seed, 3).await?;
assert_history_and_iam(&dist, &seed, "homogeneous-current").await?;
complete_legacy_multipart(&dist, &seed).await?;
Ok(())
}
@@ -1716,6 +1716,15 @@ mod tests {
cluster.set_env("RUSTFS_HEAL_AUTO_HEAL_ENABLE", background_enabled.to_string());
cluster.set_env("RUSTFS_HEAL_MRF_ENABLE", background_enabled.to_string());
cluster.set_env("RUSTFS_SCANNER_ENABLED", background_enabled.to_string());
let bucket = "heal-restart-during-rebuild";
let outage_key_prefix = "cluster/written-while-node-down-";
if !background_enabled {
// With MRF disabled, an outage PUT still hands its missing shard to a
// best-effort repair on the PUT coordinator. That repair retries until
// the target rejoins, so suppress it to keep the explicit root heal
// the only owner of the outage object.
cluster.set_env("RUSTFS_E2E_SUPPRESS_RENAME_TAIL_HEAL_PREFIX", format!("{bucket}/{outage_key_prefix}"));
}
if background_enabled {
// Only the scanner cadence is accelerated. Keep normal Heal
// concurrency and every automatic recovery owner enabled.
@@ -1748,7 +1757,6 @@ mod tests {
cluster.start_with_binary(&server_binary).await?;
let clients = cluster.create_all_clients()?;
let bucket = "heal-restart-during-rebuild";
clients[0].create_bucket().bucket(bucket).send().await?;
let ReplacementDriveSelection {
@@ -1876,7 +1884,7 @@ mod tests {
let mut replacement_outage_erasure_index = None;
let mut rejected_outage_keys = Vec::new();
for attempt in 0..max_outage_write_attempts {
let candidate_key = format!("cluster/written-while-node-down-{attempt:04}.bin");
let candidate_key = format!("{outage_key_prefix}{attempt:04}.bin");
let put_result = timeout(
Duration::from_secs(30),
clients[2]
@@ -1986,25 +1994,6 @@ mod tests {
}
let heal_body = r#"{"recursive":true,"dryRun":false,"remove":false,"recreate":true,"scanMode":2,"updateParity":false,"nolock":false}"#;
if !background_enabled {
// The PUT path may have admitted a direct Internal object repair while
// node 1 was offline. Cancel the isolated bucket path before the target
// returns; otherwise it could rebuild the outage object and invalidate
// the explicit-root ownership assertion below.
let cancel_outage_heal_path = format!("/rustfs/admin/v3/heal/{bucket}?forceStop=true");
let (cancel_status, cancel_body) = admin_request(
&cluster.nodes[0].url,
Method::POST,
&cancel_outage_heal_path,
Some(heal_body.to_string()),
&cluster.access_key,
&cluster.secret_key,
)
.await?;
if !cancel_status.is_success() {
return Err(format!("cancel outage heal failed: {cancel_status} {cancel_body}").into());
}
}
// Keep the partial-repair checkpoint stable across readiness and admin
// requests. Endpoint-blackhole tests must prove their own network stall.
@@ -38,6 +38,8 @@ mod tests {
use tracing::info;
const PAGINATION_TIMEOUT: Duration = Duration::from_secs(120);
// Bound queued durable commits on the single fixture disk under CI I/O pressure.
const FIXTURE_PUT_CONCURRENCY: usize = 4;
// Fixture population can require thousands of durable PUTs. Bound the
// listing traversal itself with one deadline shared by every page.
@@ -564,7 +566,7 @@ mod tests {
let object_count = 1002;
// Finish every durable PUT before listing, with the same bounded fanout as the delimiter fixtures.
stream::iter(0..object_count)
.for_each_concurrent(16, |i| {
.for_each_concurrent(FIXTURE_PUT_CONCURRENCY, |i| {
let client = &client;
async move {
let key = format!("object{i:04}.txt");
@@ -575,7 +577,7 @@ mod tests {
.body(ByteStream::from_static(b"test content"))
.send()
.await
.unwrap_or_else(|err| panic!("Failed to put fixture object {key}: {err}"));
.unwrap_or_else(|err| panic!("Failed to put fixture object {key}: {err:?}"));
}
})
.await;
@@ -822,7 +824,7 @@ mod tests {
.flat_map(|dir| (0..10).map(move |file| format!("dir-{dir:03}/file{file:02}.txt")))
.collect();
stream::iter(&all_keys)
.for_each_concurrent(16, |key| {
.for_each_concurrent(FIXTURE_PUT_CONCURRENCY, |key| {
let client = &client;
async move {
client
@@ -832,7 +834,7 @@ mod tests {
.body(ByteStream::from_static(b"x"))
.send()
.await
.unwrap_or_else(|err| panic!("Failed to put fixture object {key}: {err}"));
.unwrap_or_else(|err| panic!("Failed to put fixture object {key}: {err:?}"));
}
})
.await;
@@ -1055,7 +1057,7 @@ mod tests {
let files_per_dir = 100;
let fixture_started = std::time::Instant::now();
stream::iter((0..dir_count).flat_map(|d| (0..files_per_dir).map(move |f| (d, f))))
.for_each_concurrent(16, |(d, f)| {
.for_each_concurrent(FIXTURE_PUT_CONCURRENCY, |(d, f)| {
let client = &client;
async move {
let key = format!("dir-{d:02}/file{f:03}.txt");
@@ -1066,7 +1068,7 @@ mod tests {
.body(ByteStream::from_static(b"x"))
.send()
.await
.unwrap_or_else(|err| panic!("Failed to put fixture object {key}: {err}"));
.unwrap_or_else(|err| panic!("Failed to put fixture object {key}: {err:?}"));
}
})
.await;
@@ -261,6 +261,61 @@ mod tests {
archive
}
#[tokio::test]
async fn snowball_auto_extract_rejects_veeam_virtual_members() -> Result<(), Box<dyn Error + Send + Sync>> {
init_logging();
let mut env = RustFSTestEnvironment::new().await?;
env.start_rustfs_server_with_env(vec![], &[("RUSTFS_VEEAM_SOSAPI_ENABLE", "true")])
.await?;
let client = env.create_s3_client();
let bucket = "snowball-veeam-reserved";
client.create_bucket().bucket(bucket).send().await?;
let prefix = ".system-d26a9498-cb7c-4a87-a44a-8ae204f5ba6c";
for (member, destination, use_prefix) in [
(format!("{prefix}/system.xml"), format!("{prefix}/system.xml"), false),
("capacity.xml".to_string(), format!("{prefix}/capacity.xml"), true),
] {
let mut builder = tokio_tar::Builder::new(Cursor::new(Vec::new()));
append_typed_entry(&mut builder, &member, tokio_tar::EntryType::Regular, b"blocked").await?;
let archive = builder.into_inner().await?.into_inner();
let mut request = client
.put_object()
.bucket(bucket)
.key("fixture.tar")
.metadata("Snowball-Auto-Extract", "true")
.body(ByteStream::from(archive));
if use_prefix {
request = request
.metadata("Minio-Snowball-Prefix", prefix)
.metadata("Minio-Snowball-Ignore-Errors", "true");
}
let error = request.send().await.expect_err("virtual archive members must be rejected");
assert_eq!(error.code(), Some("MethodNotAllowed"));
let listed = client.list_objects_v2().bucket(bucket).prefix(&destination).send().await?;
assert!(listed.contents().is_empty(), "rejected virtual member must not be stored");
}
let mut builder = tokio_tar::Builder::new(Cursor::new(Vec::new()));
append_typed_entry(&mut builder, "ordinary.xml", tokio_tar::EntryType::Regular, b"ordinary").await?;
client
.put_object()
.bucket(bucket)
.key("fixture.tar")
.metadata("Snowball-Auto-Extract", "true")
.metadata("Minio-Snowball-Prefix", prefix)
.body(ByteStream::from(builder.into_inner().await?.into_inner()))
.send()
.await?;
let ordinary = client
.get_object()
.bucket(bucket)
.key(format!("{prefix}/ordinary.xml"))
.send()
.await?;
assert_eq!(ordinary.body.collect().await?.into_bytes().as_ref(), b"ordinary");
env.stop_server();
Ok(())
}
#[tokio::test]
async fn snowball_auto_extract_supports_minio_prefix_and_directory_markers() -> Result<(), Box<dyn Error + Send + Sync>> {
init_logging();
@@ -13,9 +13,13 @@
// limitations under the License.
use crate::common::{RustFSTestClusterEnvironment, init_logging};
use crate::storage_api::multipart_fence_acceptance::RUSTFS_META_MULTIPART_BUCKET;
use aws_sdk_s3::error::SdkError;
use aws_sdk_s3::primitives::ByteStream;
use tokio::time::{Duration, sleep};
use sha2::{Digest, Sha256};
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use tokio::time::{Duration, Instant, sleep};
use tracing::info;
use uuid::Uuid;
@@ -84,6 +88,126 @@ async fn wait_for_cleanup_on_all_nodes(
Err("stale multipart upload was not cleaned up on all nodes within timeout".into())
}
type PhysicalTree = BTreeMap<PathBuf, Option<Vec<u8>>>;
fn physical_tree(root: &Path) -> Result<PhysicalTree, Box<dyn std::error::Error + Send + Sync>> {
let mut result = BTreeMap::new();
let mut pending = vec![root.to_path_buf()];
while let Some(directory) = pending.pop() {
for entry in std::fs::read_dir(directory)? {
let entry = entry?;
let path = entry.path();
let relative = path.strip_prefix(root)?.to_path_buf();
let kind = entry.file_type()?;
if kind.is_dir() {
result.insert(relative, None);
pending.push(path);
} else if kind.is_file() {
result.insert(relative, Some(std::fs::read(path)?));
} else {
return Err(format!("unexpected staging entry type: {relative:?}").into());
}
}
}
Ok(result)
}
async fn assert_low_quorum_residuals_are_diagnosed_and_preserved(
cluster: &RustFSTestClusterEnvironment,
clients: &[aws_sdk_s3::Client],
log_paths: &[PathBuf],
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
let mut residuals = Vec::new();
for copies in [1_usize, 2] {
let key = format!("multipart/partial-{copies}-{}", Uuid::new_v4().simple());
let upload = clients[0]
.create_multipart_upload()
.bucket(CLEANUP_BUCKET)
.key(&key)
.send()
.await?;
let upload_id = upload.upload_id().ok_or("create response missing upload ID")?;
clients[0]
.upload_part()
.bucket(CLEANUP_BUCKET)
.key(&key)
.upload_id(upload_id)
.part_number(1)
.body(ByteStream::from(vec![0x41; 4096]))
.send()
.await?;
let decoded = String::from_utf8(base64_simd::URL_SAFE_NO_PAD.decode_to_vec(upload_id.as_bytes())?)?;
let (_, upload_dir) = decoded.split_once('.').ok_or("upload ID lacks deployment delimiter")?;
let digest = Sha256::digest(format!("{CLEANUP_BUCKET}/{key}").as_bytes());
let hash = hex_simd::encode_to_string(digest, hex_simd::AsciiCase::Lower);
let relative = format!("{hash}/{upload_dir}");
let roots = cluster
.nodes
.iter()
.map(|node| {
Path::new(&node.data_dirs[0])
.join(RUSTFS_META_MULTIPART_BUCKET)
.join(&relative)
})
.collect::<Vec<_>>();
assert_eq!(roots.len(), 4);
let mut preserved = Vec::new();
for root in &roots {
assert!(
root.join("xl.meta").is_file(),
"all four original metadata files must exist before injection"
);
preserved.push(physical_tree(root)?);
}
for root in roots.iter().skip(copies) {
std::fs::remove_dir_all(root)?;
}
residuals.push((relative, roots, preserved, copies));
}
let deadline = Instant::now() + Duration::from_secs(45);
loop {
let logs = log_paths
.iter()
.filter_map(|path| std::fs::read_to_string(path).ok())
.collect::<Vec<_>>();
let diagnosed = residuals.iter().all(|(path, _, _, _)| {
logs.iter().any(|log| {
log.lines()
.any(|line| line.contains(path) && line.contains("multipart_cleanup_lock_or_recheck_failed"))
})
});
if diagnosed {
break;
}
if Instant::now() >= deadline {
return Err("stale low-quorum residuals were not diagnosed before the deadline".into());
}
sleep(Duration::from_millis(200)).await;
}
for (path, roots, preserved, copies) in &residuals {
for slot in 0..*copies {
assert_eq!(
physical_tree(&roots[slot])?,
preserved[slot],
"retained source evidence changed: {path}, slot {slot}"
);
}
for root in roots.iter().skip(*copies) {
assert!(!root.exists(), "cleanup must not recreate missing source roots: {path}");
}
}
for path in log_paths {
let log = std::fs::read_to_string(path)?;
assert!(
!log.contains("op=\"upload_part\"") && !log.contains("\"op\":\"upload_part\""),
"background cleanup must not be logged as a client UploadPart"
);
}
eprintln!("Stale MPU low-quorum: EC:2 one/two-copy roots diagnosed; full surviving file inventories preserved");
Ok(())
}
#[tokio::test]
async fn test_stale_multipart_cleanup_removes_incomplete_upload_across_cluster()
-> Result<(), Box<dyn std::error::Error + Send + Sync>> {
@@ -92,6 +216,22 @@ async fn test_stale_multipart_cleanup_removes_incomplete_upload_across_cluster()
let mut cluster = RustFSTestClusterEnvironment::new(4).await?;
cluster.set_env("RUSTFS_API_STALE_UPLOADS_EXPIRY", "5s");
cluster.set_env("RUSTFS_API_STALE_UPLOADS_CLEANUP_INTERVAL", "1s");
cluster.set_env("RUSTFS_STORAGE_CLASS_STANDARD", "EC:2");
cluster.set_env(
"RUST_LOG",
concat!("warn,rustfs_ecstore", "::bucket::lifecycle::bucket_lifecycle_ops=debug"),
);
let log_dir = std::env::var_os("RUSTFS_E2E_LOG_DIR")
.map(PathBuf::from)
.map(|root| root.join(format!("stale-multipart-{}", Uuid::new_v4().simple())))
.unwrap_or_else(|| PathBuf::from(&cluster.temp_dir));
std::fs::create_dir_all(&log_dir)?;
let log_paths = (0..cluster.nodes.len())
.map(|index| log_dir.join(format!("stale-node-{index}.log")))
.collect::<Vec<_>>();
for (index, path) in log_paths.iter().enumerate() {
cluster.set_node_capture_log_path(index, path.to_string_lossy().into_owned())?;
}
cluster.start().await?;
cluster.create_test_bucket(CLEANUP_BUCKET).await?;
@@ -137,6 +277,7 @@ async fn test_stale_multipart_cleanup_removes_incomplete_upload_across_cluster()
);
wait_for_cleanup_on_all_nodes(&clients, CLEANUP_BUCKET, &key, &upload_id).await?;
assert_low_quorum_residuals_are_diagnosed_and_preserved(&cluster, &clients, &log_paths).await?;
Ok(())
}
@@ -1066,7 +1066,7 @@ pub(crate) async fn wait_for_upgrade_write_readiness(
/// reduced-redundancy class then fails with 503 `ServiceUnavailable` until the
/// serving node writes through every drive, on the previous release and on
/// the current build alike.
async fn wait_for_drive_readmission(clients: &[Client], bucket: &str, context: &str, budget: Duration) -> TestResult {
pub(crate) async fn wait_for_drive_readmission(clients: &[Client], bucket: &str, context: &str, budget: Duration) -> TestResult {
wait_for_upgrade_writes(
clients,
bucket,
@@ -600,6 +600,154 @@ mod tests {
info!("✅ PASSED: Veeam backup workflow simulation completed successfully");
}
/// Veeam B&R discovers SOSAPI by GET/HEAD of two well-known keys.
#[tokio::test]
async fn test_veeam_sosapi_system_and_capacity() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
use crate::common::local_http_client;
const SYSTEM_KEY: &str = ".system-d26a9498-cb7c-4a87-a44a-8ae204f5ba6c/system.xml";
const CAPACITY_KEY: &str = ".system-d26a9498-cb7c-4a87-a44a-8ae204f5ba6c/capacity.xml";
init_logging();
info!("🧪 TEST: Veeam SOSAPI system.xml and capacity.xml");
let mut env = RustFSTestEnvironment::new().await?;
env.start_rustfs_server_with_env(vec![], crate::common::FAST_DATA_USAGE_SCANNER_ENV)
.await?;
let client = create_s3_client(&env);
let bucket = "veeam-sosapi";
create_bucket(&client, bucket).await?;
let system = client.get_object().bucket(bucket).key(SYSTEM_KEY).send().await?;
let system_etag = system.e_tag().map(str::to_string);
let system_len = system.content_length();
assert_eq!(system.content_type(), Some("application/xml"));
let system_body = system.body.collect().await?.into_bytes();
let system_xml = std::str::from_utf8(&system_body)?;
assert!(system_xml.contains("<ProtocolVersion>\"1.0\"</ProtocolVersion>"), "{system_xml}");
assert!(
system_xml.contains("<ModelName>\"RustFS ") && system_xml.contains("\"</ModelName>"),
"{system_xml}"
);
assert!(system_xml.contains("<CapacityInfo>true</CapacityInfo>"));
assert!(system_xml.contains("<UploadSessions>false</UploadSessions>"));
assert!(system_xml.contains("<IAMSTS>false</IAMSTS>"));
assert!(!system_xml.contains("SystemRecommendations"));
assert!(!system_xml.contains("APIEndpoints"));
assert!(!system_xml.contains("&quot;"));
let head = client.head_object().bucket(bucket).key(SYSTEM_KEY).send().await?;
assert_eq!(head.content_type(), Some("application/xml"));
assert_eq!(head.content_length(), system_len);
assert_eq!(head.e_tag().map(str::to_string), system_etag);
assert!(head.last_modified().is_some());
let ranged = client
.get_object()
.bucket(bucket)
.key(SYSTEM_KEY)
.range("bytes=0-3")
.send()
.await?;
assert_eq!(ranged.content_length(), Some(4));
let ranged_header = ranged.content_range().unwrap_or("");
assert!(ranged_header.starts_with("bytes 0-3/"), "{ranged_header}");
let ranged_body = ranged.body.collect().await?.into_bytes();
assert_eq!(&ranged_body[..], &system_body[..4]);
let capacity = tokio::time::timeout(std::time::Duration::from_secs(180), async {
loop {
match client.get_object().bucket(bucket).key(CAPACITY_KEY).send().await {
Ok(capacity) => break Ok(capacity),
Err(err) if err.code() == Some("ServiceUnavailable") => {
tokio::time::sleep(std::time::Duration::from_secs(2)).await;
}
Err(err) => break Err(err),
}
}
})
.await??;
assert_eq!(capacity.content_type(), Some("application/xml"));
let capacity_body = capacity.body.collect().await?.into_bytes();
let capacity_xml = std::str::from_utf8(&capacity_body)?;
let capacity_value = xml_u64(capacity_xml, "Capacity")?;
let available_value = xml_u64(capacity_xml, "Available")?;
let used_value = xml_u64(capacity_xml, "Used")?;
assert!(used_value <= capacity_value, "{capacity_xml}");
assert!(available_value <= capacity_value.saturating_sub(used_value), "{capacity_xml}");
let put_err = client
.put_object()
.bucket(bucket)
.key(SYSTEM_KEY)
.body(ByteStream::from_static(b"overwrite"))
.send()
.await
.expect_err("SOSAPI put must fail");
assert_eq!(put_err.code(), Some("MethodNotAllowed"));
let delete_err = client
.delete_object()
.bucket(bucket)
.key(CAPACITY_KEY)
.send()
.await
.expect_err("SOSAPI delete must fail");
assert_eq!(delete_err.code(), Some("MethodNotAllowed"));
let normal_key = "Veeam/Backup/job.dat";
client
.put_object()
.bucket(bucket)
.key(normal_key)
.body(ByteStream::from_static(b"backup-bytes"))
.send()
.await?;
let normal = client.get_object().bucket(bucket).key(normal_key).send().await?;
let normal_body = normal.body.collect().await?.into_bytes();
assert_eq!(&normal_body[..], b"backup-bytes");
let anon = local_http_client()
.get(format!("{}/{bucket}/{SYSTEM_KEY}", env.url))
.send()
.await?;
assert_eq!(anon.status(), reqwest::StatusCode::FORBIDDEN);
let anon_body = anon.text().await?;
assert!(!anon_body.contains("ProtocolVersion"), "{anon_body}");
let now = time::OffsetDateTime::now_utc();
let scope_date = format!("{:04}{:02}{:02}", now.year(), u8::from(now.month()), now.day());
let amz_date = format!("{scope_date}T{:02}{:02}{:02}Z", now.hour(), now.minute(), now.second());
let bad = local_http_client()
.get(format!("{}/{bucket}/{SYSTEM_KEY}", env.url))
.header("x-amz-date", &amz_date)
.header("x-amz-content-sha256", "UNSIGNED-PAYLOAD")
.header(
"authorization",
format!(
"AWS4-HMAC-SHA256 Credential=bad/{scope_date}/us-east-1/s3/aws4_request, SignedHeaders=host;x-amz-content-sha256;x-amz-date, Signature=0000000000000000000000000000000000000000000000000000000000000000"
),
)
.send()
.await?;
assert_eq!(bad.status(), reqwest::StatusCode::FORBIDDEN);
let bad_body = bad.text().await?;
assert!(!bad_body.contains("ProtocolVersion"), "{bad_body}");
info!("✅ PASSED: Veeam SOSAPI system.xml and capacity.xml");
Ok(())
}
fn xml_u64(xml: &str, element: &str) -> Result<u64, Box<dyn std::error::Error + Send + Sync>> {
let start = format!("<{element}>");
let end = format!("</{element}>");
let start_idx = xml.find(&start).ok_or_else(|| format!("missing {element} in {xml}"))? + start.len();
let end_idx = xml[start_idx..]
.find(&end)
.ok_or_else(|| format!("unclosed {element} in {xml}"))?
+ start_idx;
Ok(xml[start_idx..end_idx].parse()?)
}
#[tokio::test]
async fn test_terraform_put_after_delete() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
init_logging();
+9 -8
View File
@@ -252,7 +252,7 @@ pub mod bucket {
pub mod quota {
pub use crate::bucket::quota::{
BucketQuota, QUOTA_RESERVATION_PROTOCOL_V1, QUOTA_RESERVATION_PROTOCOL_V2, QuotaCheckResult, QuotaError,
QuotaMigrationSkipReason, QuotaOperation, QuotaProtocolMigration,
QuotaMigrationSkipReason, QuotaOperation, QuotaProtocolMigration, QuotaType,
};
pub mod checker {
@@ -385,10 +385,10 @@ pub mod config {
ServerConfigCorruptError, ServerConfigSaveResult, ServerConfigSnapshot, delete_config, delete_config_no_lock,
is_server_config_corrupt_error, lookup_configs, read_config, read_config_no_lock, read_config_with_metadata,
read_config_without_migrate, read_config_without_migrate_no_lock, read_existing_server_config_no_lock,
read_server_config_snapshot, save_config, save_config_no_lock, save_config_with_opts, save_config_with_opts_quiet,
save_server_config, save_server_config_no_lock, save_server_config_snapshot,
save_server_config_snapshot_with_generation, server_config_path, try_migrate_server_config,
with_config_object_read_lock, with_config_object_write_lock, with_server_config_read_lock,
read_server_config_snapshot, read_server_config_with_generation, save_config, save_config_no_lock,
save_config_with_opts, save_config_with_opts_quiet, save_server_config, save_server_config_no_lock,
save_server_config_snapshot, save_server_config_snapshot_with_generation, server_config_path,
try_migrate_server_config, with_config_object_read_lock, with_config_object_write_lock, with_server_config_read_lock,
with_server_config_write_lock,
};
}
@@ -425,9 +425,10 @@ pub mod data_usage {
DATA_USAGE_CACHE_NAME, apply_bucket_usage_memory_overlay, compute_bucket_usage,
init_compression_total_memory_from_backend, invalidate_admin_data_usage_snapshot_cache,
invalidate_data_usage_snapshot_cache, live_bucket_usage_computations, load_admin_data_usage_from_backend_cached,
load_compression_total_from_memory, load_data_usage_from_backend, load_data_usage_from_backend_cached, quota_object_size,
record_bucket_delete_marker_memory, record_bucket_object_delete_memory, record_bucket_object_version_write_memory,
record_bucket_object_write_memory, record_bucket_object_write_unknown_previous_memory, record_compression_total_memory,
load_bucket_usage_from_backend_cached, load_compression_total_from_memory, load_data_usage_from_backend,
load_data_usage_from_backend_cached, quota_object_size, record_bucket_delete_marker_memory,
record_bucket_object_delete_memory, record_bucket_object_version_write_memory, record_bucket_object_write_memory,
record_bucket_object_write_unknown_previous_memory, record_compression_total_memory,
refresh_bucket_usage_from_object_layer, refresh_versioned_bucket_usage_from_object_layer,
remove_bucket_usage_from_backend, replace_bucket_usage_memory_from_info, store_compression_total_in_backend,
store_data_usage_in_backend,
@@ -63,6 +63,7 @@ use crate::services::tier::{
};
use crate::set_disk::{
MAX_PARTS_COUNT, RUSTFS_MULTIPART_BUCKET_KEY, RUSTFS_MULTIPART_OBJECT_KEY, SetDisks, get_lock_acquire_timeout,
multipart_part_logical_size, should_persist_encryption_original_size,
};
use crate::storage_api_contracts::{
lifecycle::ExpirationOptions,
@@ -119,6 +120,7 @@ const LOG_SUBSYSTEM_LIFECYCLE: &str = "lifecycle";
const EVENT_LIFECYCLE_WORKER_STATE: &str = "lifecycle_worker_state";
const EVENT_LIFECYCLE_TRANSITION_COMPENSATION: &str = "lifecycle_transition_compensation";
const EVENT_LIFECYCLE_STALE_MULTIPART_CLEANUP: &str = "lifecycle_stale_multipart_cleanup";
const METRIC_STALE_MULTIPART_CLEANUP_BLOCKED_TOTAL: &str = "rustfs_stale_multipart_cleanup_blocked_total";
const EVENT_LIFECYCLE_SCAN_SKIPPED: &str = "lifecycle_scan_skipped";
const EVENT_LIFECYCLE_EVALUATION_FAILED: &str = "lifecycle_evaluation_failed";
const EVENT_LIFECYCLE_EXPIRED_DETECTED: &str = "lifecycle_expired_detected";
@@ -3511,6 +3513,7 @@ async fn stale_upload_current_size_with_opts(
&ObjectOptions {
data_movement,
no_lock,
strict_multipart_part_sizes: true,
..Default::default()
},
)
@@ -3524,13 +3527,14 @@ fn stale_upload_size_from_parts(parts: &crate::storage_api_contracts::multipart:
if parts.is_truncated {
return None;
}
let transformed = rustfs_utils::http::contains_key_str(&parts.user_defined, rustfs_utils::http::SUFFIX_COMPRESSION)
|| should_persist_encryption_original_size(&parts.user_defined);
parts.parts.iter().try_fold(0usize, |total, part| {
// ListParts represents a quorum-confirmed missing part with an empty ETag.
if part.etag.as_deref().is_none_or(str::is_empty) {
return None;
}
let actual_size = usize::try_from(part.actual_size.max(0)).ok()?;
total.checked_add(part.size.max(actual_size))
total.checked_add(multipart_part_logical_size(part.size, part.actual_size, transformed)?)
})
}
@@ -3741,6 +3745,23 @@ async fn cleanup_empty_multipart_sha_dirs_on_local_disks(set: &Arc<SetDisks>) {
}
}
fn record_stale_multipart_cleanup_blocked(stage: &'static str) {
metrics::counter!(METRIC_STALE_MULTIPART_CLEANUP_BLOCKED_TOTAL, "stage" => stage).increment(1);
}
fn record_stale_multipart_cleanup_lock_lost(path: &str) {
record_stale_multipart_cleanup_blocked("lock_lost");
debug!(
event = EVENT_LIFECYCLE_STALE_MULTIPART_CLEANUP,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_LIFECYCLE,
state = "skipped",
path = %path,
reason = "multipart_cleanup_lock_lost",
"Skipped stale multipart cleanup"
);
}
async fn cleanup_stale_multipart_uploads_in_set(set: &Arc<SetDisks>, now: OffsetDateTime, default_expiry: StdDuration) -> usize {
let mut deleted = 0usize;
let mut candidates = HashMap::new();
@@ -3852,6 +3873,7 @@ async fn cleanup_stale_multipart_uploads_in_set(set: &Arc<SetDisks>, now: Offset
let cleanup_guard = match set.lock_stale_multipart_cleanup(&candidate.path).await {
Ok(guard) => guard,
Err(err) => {
record_stale_multipart_cleanup_blocked("recheck");
debug!(
event = EVENT_LIFECYCLE_STALE_MULTIPART_CLEANUP,
component = LOG_COMPONENT_ECSTORE,
@@ -3874,7 +3896,11 @@ async fn cleanup_stale_multipart_uploads_in_set(set: &Arc<SetDisks>, now: Offset
{
current_due = lifecycle_due;
}
if now < current_due || cleanup_guard.is_lock_lost() {
if now < current_due {
continue;
}
if cleanup_guard.is_lock_lost() {
record_stale_multipart_cleanup_lock_lost(&candidate.path);
continue;
}
@@ -3894,15 +3920,18 @@ async fn cleanup_stale_multipart_uploads_in_set(set: &Arc<SetDisks>, now: Offset
"Removed stale multipart upload"
);
}
Err(err) => debug!(
event = EVENT_LIFECYCLE_STALE_MULTIPART_CLEANUP,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_LIFECYCLE,
path = %candidate.path,
error = ?err,
reason = "multipart_remove_failed",
"Failed to remove stale multipart upload"
),
Err(err) => {
record_stale_multipart_cleanup_blocked("delete");
debug!(
event = EVENT_LIFECYCLE_STALE_MULTIPART_CLEANUP,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_LIFECYCLE,
path = %candidate.path,
error = ?err,
reason = "multipart_remove_failed",
"Failed to remove stale multipart upload"
);
}
}
}
@@ -15990,6 +16019,8 @@ mod tests {
.with_ansi(false)
.without_time()
.finish();
let recorder = crate::test_metrics::CapturingRecorder::default();
let recorder_guard = metrics::set_default_local_recorder(&recorder);
cleanup_stale_multipart_uploads_once_at(
ecstore.clone(),
OffsetDateTime::now_utc(),
@@ -15997,21 +16028,53 @@ mod tests {
)
.with_subscriber(subscriber)
.await;
drop(recorder_guard);
for (path, expected) in physical_paths.iter().zip(&original) {
assert_eq!(fs::read(path.join(STORAGE_FORMAT_FILE)).await.unwrap(), *expected);
}
let log_output = String::from_utf8(logs.0.lock().unwrap().clone()).unwrap();
let rechecks = log_output
.lines()
.filter(|line| line.contains("multipart_cleanup_lock_or_recheck_failed"));
let matching_path = rechecks.clone().any(|line| line.contains(&upload_path));
assert!(
log_output.contains(&upload_path),
"missing background diagnostic for {upload_path}: {log_output}"
matching_path,
"missing path-specific background diagnostic for {upload_path}: {log_output}"
);
assert_eq!(
recorder.counter_value(super::METRIC_STALE_MULTIPART_CLEANUP_BLOCKED_TOTAL, &[("stage", "recheck")]),
u64::try_from(rechecks.count()).expect("diagnostic count fits u64"),
"each observed recheck failure must increment the metric exactly once"
);
assert!(log_output.contains("multipart_cleanup_lock_or_recheck_failed"), "{log_output}");
assert!(!log_output.contains("op=\"upload_part\""), "unexpected upload-part warning: {log_output}");
for path in physical_paths.iter().take(copies) {
fs::remove_dir_all(path).await.unwrap();
}
}
let logs = CapturedLogs::default();
let subscriber = tracing_subscriber::fmt()
.with_max_level(tracing::Level::DEBUG)
.with_writer(logs.clone())
.with_ansi(false)
.without_time()
.finish();
let recorder = crate::test_metrics::CapturingRecorder::default();
metrics::with_local_recorder(&recorder, || {
tracing::subscriber::with_default(subscriber, || {
super::record_stale_multipart_cleanup_lock_lost("test-hash/test-upload");
})
});
let output = String::from_utf8(logs.0.lock().unwrap().clone()).unwrap();
assert!(
output.contains("multipart_cleanup_lock_lost") && output.contains("test-hash/test-upload"),
"{output}"
);
assert!(!output.contains("op=\"upload_part\""), "{output}");
assert_eq!(
recorder.counter_value(super::METRIC_STALE_MULTIPART_CLEANUP_BLOCKED_TOTAL, &[("stage", "lock_lost")]),
1
);
}
#[tokio::test(flavor = "current_thread")]
@@ -16021,7 +16084,34 @@ mod tests {
use tracing::instrument::WithSubscriber as _;
use tracing_subscriber::prelude::*;
struct PartListCounter(Arc<AtomicUsize>);
struct PartListCounter {
calls: Arc<AtomicUsize>,
bucket: String,
object: &'static str,
}
struct PartListFields<'a> {
bucket: &'a str,
object: &'a str,
matching_bucket: bool,
matching_object: bool,
}
impl tracing::field::Visit for PartListFields<'_> {
fn record_str(&mut self, field: &tracing::field::Field, value: &str) {
match field.name() {
"bucket" => self.matching_bucket = value == self.bucket,
"object" => self.matching_object = value == self.object,
_ => {}
}
}
fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
let value = format!("{value:?}");
match field.name() {
"bucket" => self.matching_bucket = value == format!("{:?}", self.bucket),
"object" => self.matching_object = value == format!("{:?}", self.object),
_ => {}
}
}
}
impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for PartListCounter {
fn on_new_span(
&self,
@@ -16029,8 +16119,18 @@ mod tests {
_id: &tracing::span::Id,
_ctx: tracing_subscriber::layer::Context<'_, S>,
) {
if attrs.metadata().name() == "list_object_parts" {
self.0.fetch_add(1, Ordering::Relaxed);
if attrs.metadata().name() != "list_object_parts" {
return;
}
let mut fields = PartListFields {
bucket: &self.bucket,
object: self.object,
matching_bucket: false,
matching_object: false,
};
attrs.record(&mut fields);
if fields.matching_bucket && fields.matching_object {
self.calls.fetch_add(1, Ordering::Relaxed);
}
}
}
@@ -16067,7 +16167,11 @@ mod tests {
.expect("upload a part for size filtering");
}
let calls = Arc::new(AtomicUsize::new(0));
let subscriber = tracing_subscriber::registry().with(PartListCounter(Arc::clone(&calls)));
let subscriber = tracing_subscriber::registry().with(PartListCounter {
calls: Arc::clone(&calls),
bucket: bucket.clone(),
object,
});
cleanup_stale_multipart_uploads_once_at(ecstore.clone(), now, StdDuration::from_secs(24 * 60 * 60))
.with_subscriber(subscriber)
.await;
@@ -16094,7 +16198,7 @@ mod tests {
..Default::default()
};
let mut parts = ListPartsInfo {
parts: vec![part(4, 6, "valid")],
parts: vec![part(6, 6, "valid")],
..Default::default()
};
assert_eq!(super::stale_upload_size_from_parts(&parts), Some(6));
@@ -16103,6 +16207,22 @@ mod tests {
parts.is_truncated = false;
parts.parts = vec![part(0, 0, "")];
assert_eq!(super::stale_upload_size_from_parts(&parts), None);
parts.parts = vec![part(1, -1, "valid")];
assert_eq!(super::stale_upload_size_from_parts(&parts), None);
parts.parts = vec![part(8, 6, "valid")];
assert_eq!(super::stale_upload_size_from_parts(&parts), None);
parts.parts = vec![part(4, 6, "valid")];
assert_eq!(super::stale_upload_size_from_parts(&parts), None);
rustfs_utils::http::insert_str(&mut parts.user_defined, rustfs_utils::http::SUFFIX_COMPRESSION, "test".to_owned());
assert_eq!(super::stale_upload_size_from_parts(&parts), Some(6));
parts.parts = vec![part(8, 6, "valid")];
assert_eq!(super::stale_upload_size_from_parts(&parts), Some(6));
rustfs_utils::http::remove_str(&mut parts.user_defined, rustfs_utils::http::SUFFIX_COMPRESSION);
parts
.user_defined
.insert("x-amz-server-side-encryption".to_owned(), "AES256".to_owned());
assert_eq!(super::stale_upload_size_from_parts(&parts), Some(6));
parts.user_defined.clear();
parts.parts = vec![part(usize::MAX, 0, "valid"), part(1, 1, "valid")];
assert_eq!(super::stale_upload_size_from_parts(&parts), None);
}
@@ -16157,6 +16277,184 @@ mod tests {
.is_some(),
"a size-independent rule remains eligible when size is unknown"
);
ecstore
.abort_multipart_upload(&bucket, object, &upload.upload_id, &ObjectOptions::default())
.await
.expect("remove this test's remaining upload from the shared fixture");
}
#[tokio::test]
#[serial]
async fn strict_multipart_lifecycle_size_skips_unavailable_participant_and_recovers() {
let (_, ecstore) = setup_test_env().await;
let bucket = format!("stale-strict-size-{}", Uuid::new_v4().simple());
let object = "logs/object";
create_test_bucket(&ecstore, &bucket).await;
let rule = br#"<LifecycleConfiguration><Rule><ID>size</ID><Status>Enabled</Status><Filter><ObjectSizeLessThan>5</ObjectSizeLessThan></Filter><AbortIncompleteMultipartUpload><DaysAfterInitiation>0</DaysAfterInitiation></AbortIncompleteMultipartUpload></Rule></LifecycleConfiguration>"#;
metadata_sys::update(&bucket, BUCKET_LIFECYCLE_CONFIG, rule.to_vec())
.await
.expect("install upper-size rule");
let upload = ecstore
.new_multipart_upload(&bucket, object, &ObjectOptions::default())
.await
.expect("create empty upload");
let decoded = base64_simd::URL_SAFE_NO_PAD
.decode_to_vec(upload.upload_id.as_bytes())
.expect("decode upload ID");
let decoded = String::from_utf8(decoded).expect("upload ID is UTF-8");
let (_, upload_dir) = decoded.split_once('.').expect("deployment-prefixed upload ID");
let metadata = HashMap::from([
(super::RUSTFS_MULTIPART_BUCKET_KEY.to_owned(), bucket.clone()),
(super::RUSTFS_MULTIPART_OBJECT_KEY.to_owned(), object.to_owned()),
]);
let initiated = OffsetDateTime::now_utc() - time::Duration::hours(1);
let set = &ecstore.pools[0].disk_set[0];
assert!(
super::stale_upload_lifecycle_due(set, &metadata, initiated, upload_dir, false)
.await
.is_some()
);
let disk = set.disks.write().await[3].take();
assert!(disk.is_some());
let due = super::stale_upload_lifecycle_due(set, &metadata, initiated, upload_dir, false).await;
set.disks.write().await[3] = disk;
assert!(due.is_none(), "unknown size must not match an upper-size rule");
assert!(
super::stale_upload_lifecycle_due(set, &metadata, initiated, upload_dir, false)
.await
.is_some()
);
ecstore
.abort_multipart_upload(&bucket, object, &upload.upload_id, &ObjectOptions::default())
.await
.expect("remove this test's remaining upload from the shared fixture");
}
#[tokio::test]
#[serial]
async fn strict_multipart_lifecycle_does_not_treat_a_minority_part_as_empty() {
let (paths, ecstore) = setup_test_env().await;
let bucket = format!("stale-part-minority-{}", Uuid::new_v4().simple());
let object = "logs/object";
create_test_bucket(&ecstore, &bucket).await;
let initiated = OffsetDateTime::now_utc() - time::Duration::hours(1);
let upload = ecstore
.new_multipart_upload(
&bucket,
object,
&ObjectOptions {
mod_time: Some(initiated),
max_parity: true,
..Default::default()
},
)
.await
.expect("create upload");
let mut body = PutObjReader::from_vec(vec![7; 6]);
ecstore
.put_object_part(&bucket, object, &upload.upload_id, 1, &mut body, &ObjectOptions::default())
.await
.expect("write a known six-byte part");
let decoded = base64_simd::URL_SAFE_NO_PAD
.decode_to_vec(upload.upload_id.as_bytes())
.expect("decode upload ID");
let decoded = String::from_utf8(decoded).expect("upload ID is UTF-8");
let (_, upload_dir) = decoded.split_once('.').expect("deployment-prefixed upload ID");
let upload_path = format!("{}/{upload_dir}", multipart_sha_dir(&bucket, object));
let root = paths[0].join(RUSTFS_META_MULTIPART_BUCKET).join(&upload_path);
let bytes = fs::read(root.join(STORAGE_FORMAT_FILE)).await.expect("read source metadata");
let info = super::get_file_info(
&bytes,
RUSTFS_META_MULTIPART_BUCKET,
&upload_path,
"",
super::FileInfoOpts {
data: false,
include_free_versions: false,
include_part_checksums: false,
},
)
.expect("decode source metadata");
let data_dir = info.data_dir.expect("upload data directory").to_string();
let metadata = HashMap::from([
(super::RUSTFS_MULTIPART_BUCKET_KEY.to_owned(), bucket.clone()),
(super::RUSTFS_MULTIPART_OBJECT_KEY.to_owned(), object.to_owned()),
]);
let set = &ecstore.pools[0].disk_set[0];
let rule = |filter: &str| {
format!("<LifecycleConfiguration><Rule><ID>size</ID><Status>Enabled</Status><Filter>{filter}</Filter><AbortIncompleteMultipartUpload><DaysAfterInitiation>0</DaysAfterInitiation></AbortIncompleteMultipartUpload></Rule></LifecycleConfiguration>").into_bytes()
};
let filters = [
"<ObjectSizeLessThan>10</ObjectSizeLessThan>",
"<ObjectSizeGreaterThan>5</ObjectSizeGreaterThan>",
];
for filter in filters {
metadata_sys::update(&bucket, BUCKET_LIFECYCLE_CONFIG, rule(filter))
.await
.expect("install baseline size filter");
assert!(
super::stale_upload_lifecycle_due(set, &metadata, initiated, upload_dir, false)
.await
.is_some(),
"known six-byte part must satisfy {filter} before the fault"
);
}
let mut saved = Vec::new();
for path in paths.iter().take(3) {
for file in ["part.1", "part.1.meta"] {
let path = path
.join(RUSTFS_META_MULTIPART_BUCKET)
.join(&upload_path)
.join(&data_dir)
.join(file);
let bytes = fs::read(&path).await.expect("snapshot original part");
fs::remove_file(&path).await.expect("simulate a missing part on this disk");
saved.push((path, bytes));
}
}
for filter in filters {
metadata_sys::update(&bucket, BUCKET_LIFECYCLE_CONFIG, rule(filter))
.await
.expect("install size filter");
super::metadata_boundary::get_lifecycle_config(&bucket)
.await
.expect("size filter must be readable during the fault");
assert!(
super::stale_upload_lifecycle_due(set, &metadata, initiated, upload_dir, false)
.await
.is_none(),
"a minority part must not satisfy {filter}"
);
}
cleanup_stale_multipart_uploads_once_at(
ecstore.clone(),
OffsetDateTime::now_utc(),
StdDuration::from_secs(7 * 24 * 60 * 60),
)
.await;
ecstore
.get_multipart_info(&bucket, object, &upload.upload_id, &ObjectOptions::default())
.await
.expect("unknown size must preserve the upload during cleanup");
for (path, bytes) in saved {
fs::write(path, bytes).await.expect("restore original part");
}
let size = super::stale_upload_current_size_with_opts(set, &metadata, upload_dir, false).await;
assert_eq!(size, Some(6), "restored part must have its original logical size");
for filter in filters {
metadata_sys::update(&bucket, BUCKET_LIFECYCLE_CONFIG, rule(filter))
.await
.expect("restore size filter");
let config = super::metadata_boundary::get_lifecycle_config(&bucket)
.await
.expect("read lifecycle");
let due = super::stale_upload_lifecycle_due(set, &metadata, initiated, upload_dir, false).await;
assert!(due.is_some(), "restored part must satisfy {filter}; size={size:?}, config={config:?}");
}
ecstore
.abort_multipart_upload(&bucket, object, &upload.upload_id, &ObjectOptions::default())
.await
.expect("remove the restored test upload from the shared fixture");
}
#[tokio::test]
@@ -63,14 +63,7 @@ pub(crate) async fn get_expiry_configs_for_options(
bucket: &str,
opts: &crate::object_api::ObjectOptions,
) -> Result<LifecycleExpiryConfigs> {
let bucket_incarnation_id = match opts
.object_lock_config_snapshot
.as_ref()
.and_then(|snapshot| snapshot.validated_bucket_incarnation_for(api.id, bucket, opts.expected_bucket_incarnation_id))
{
Some(incarnation_id) => incarnation_id,
None => metadata_sys::get_bucket_incarnation_id_for_options_in(&api.ctx, bucket, opts).await?,
};
let bucket_incarnation_id = metadata_sys::get_bucket_incarnation_id_for_options_in(&api.ctx, bucket, opts).await?;
let metadata = get_authoritative_metadata(api, bucket, bucket_incarnation_id).await?;
let table_bucket_enabled = metadata.table_bucket_enabled();
+28 -1
View File
@@ -564,12 +564,39 @@ pub(crate) async fn validate_bucket_reconstruction_in(
Ok(())
}
/// Validate against disk while retaining an already-held Object Lock fence.
/// Resolve the authoritative bucket incarnation id for `opts`:
/// 1. When namespace locking is enabled and `opts.object_lock_config_snapshot` is
/// present, validate and return the snapshot's incarnation under its held
/// lifecycle and metadata-transaction fences.
/// 2. When namespace locking is enabled and `opts.bucket_lifecycle_lock_fence` is
/// present and not lost, reuse the store's `BucketFenceRegistry` memo if a
/// registered lifecycle read guard is currently held without lock loss.
/// 3. Otherwise, acquire the bucket metadata transaction read lock (unless
/// already held on `opts`) and read the incarnation from disk.
pub(crate) async fn get_bucket_incarnation_id_for_options_in(
ctx: &crate::runtime::instance::InstanceContext,
bucket: &str,
opts: &ObjectOptions,
) -> Result<Uuid> {
if !ctx.lock_manager().is_disabled() {
if let Some(snapshot) = opts.object_lock_config_snapshot.as_ref() {
let store = object_store_in(ctx).await?;
return snapshot
.validated_bucket_incarnation_for(store.id, bucket, opts.expected_bucket_incarnation_id)
.ok_or_else(|| {
Error::other("Object Lock snapshot does not hold a valid metadata transaction fence for this bucket")
});
}
if opts
.bucket_lifecycle_lock_fence
.as_ref()
.is_some_and(|fence| !fence.is_lock_lost())
&& let Some(store) = object_store_if_initialized_in(ctx).await
&& let Some(incarnation) = store.bucket_fence_registry.validated_incarnation(bucket)
{
return Ok(incarnation);
}
}
let sys = bucket_metadata_sys_of(ctx)?;
let sys = sys.read().await.clone();
let guard = acquire_bucket_metadata_transaction_read_lock_for_options_in(ctx, bucket, opts).await?;
+9
View File
@@ -2617,6 +2617,15 @@ where
save_server_config_snapshot(api, cfg, &snapshot).await.map(|_| ())
}
/// Read config and generation from the same object without taking the transaction write lock.
pub async fn read_server_config_with_generation(api: Arc<ECStore>) -> Result<(Config, Option<Uuid>)> {
match read_config_with_metadata_inner(api.clone(), &server_config_path(), &ObjectOptions::default(), false, None).await {
Ok((raw, info)) => Ok((decode_persisted_server_config(&raw)?.merge(), info.data_dir.filter(|id| !id.is_nil()))),
Err(Error::ConfigNotFound) => read_config_without_migrate(api).await.map(|config| (config, None)),
Err(error) => Err(error),
}
}
/// One serialized read of the persisted server config used as the seed and
/// compare-and-swap fence for an admin update.
#[derive(Debug)]
+42
View File
@@ -1493,6 +1493,48 @@ pub async fn load_data_usage_from_backend_cached(store: Arc<ECStore>) -> Result<
project_data_usage_from_backend_cached(store, DataUsageInfo::clone).await
}
/// Read one bucket's persisted usage without cloning the complete snapshot.
pub async fn load_bucket_usage_from_backend_cached(
store: Arc<ECStore>,
bucket: &str,
incarnation: uuid::Uuid,
) -> Result<Option<u64>, Error> {
project_data_usage_from_backend_cached(store, |info| certified_bucket_usage(info, bucket, incarnation)).await
}
fn certified_bucket_usage(info: &DataUsageInfo, bucket: &str, incarnation: uuid::Uuid) -> Option<u64> {
if incarnation.is_nil()
|| !info.is_complete_bucket_usage_snapshot()
|| info.usage_snapshot_observational
|| info.scanner_cycle.is_none()
|| info.scanner_epoch.is_none()
{
return None;
}
let pending_count = u64::try_from(info.usage_snapshot_pending_buckets.len()).ok()?;
if pending_count != info.usage_snapshot_pending_bucket_count {
return None;
}
match info.usage_snapshot_converged {
Some(true) if pending_count == 0 => {}
Some(false)
if pending_count > 0
&& info.usage_snapshot_pending_buckets.windows(2).all(|pair| pair[0] < pair[1])
&& info
.usage_snapshot_pending_buckets
.binary_search_by(|pending| pending.as_str().cmp(bucket))
.is_err() => {}
_ => return None,
}
let usage = info.buckets_usage.get(bucket)?;
let certified = usage
.bucket_incarnation
.as_deref()
.and_then(|id| uuid::Uuid::parse_str(id).ok())
.filter(|id| !id.is_nil())?;
(certified == incarnation).then_some(usage.size)
}
// Project under the cache guard so scalar readers never copy the snapshot's
// bucket maps. The owned API shares the same TTL, refresh and invalidation path.
async fn project_data_usage_from_backend_cached<T>(
@@ -129,6 +129,41 @@ fn projected_publication_moves_bucket_maps_into_cache() {
);
}
#[test]
fn certified_bucket_usage_rejects_unproven_and_pending_data() {
let id = uuid::Uuid::from_u128(7);
let mut info = data_usage_info_for_test("known", 1, 42, SystemTime::UNIX_EPOCH);
info.usage_snapshot_converged = Some(true);
info.scanner_cycle = Some(1);
info.scanner_epoch = Some(1);
info.buckets_usage.get_mut("known").unwrap().bucket_incarnation = Some(id.to_string());
assert_eq!(certified_bucket_usage(&info, "known", id), Some(42));
assert_eq!(certified_bucket_usage(&info, "known", uuid::Uuid::from_u128(8)), None);
assert_eq!(certified_bucket_usage(&info, "known", uuid::Uuid::nil()), None);
let mut legacy = info.clone();
legacy.buckets_usage.get_mut("known").unwrap().bucket_incarnation = None;
assert_eq!(certified_bucket_usage(&legacy, "known", id), None);
legacy.buckets_usage.get_mut("known").unwrap().bucket_incarnation = Some("corrupt".into());
assert_eq!(certified_bucket_usage(&legacy, "known", id), None);
let mut pending = info.clone();
pending.usage_snapshot_converged = Some(false);
pending.usage_snapshot_pending_buckets = vec!["other".into()];
pending.usage_snapshot_pending_bucket_count = 1;
assert_eq!(certified_bucket_usage(&pending, "known", id), Some(42));
pending.usage_snapshot_pending_buckets = vec!["known".into()];
assert_eq!(certified_bucket_usage(&pending, "known", id), None);
pending.usage_snapshot_pending_bucket_count = 0;
assert_eq!(certified_bucket_usage(&pending, "known", id), None);
pending.usage_snapshot_pending_buckets = vec!["z".into(), "a".into()];
pending.usage_snapshot_pending_bucket_count = 2;
assert_eq!(certified_bucket_usage(&pending, "known", id), None);
let mut observational = info.clone();
observational.usage_snapshot_observational = true;
assert_eq!(certified_bucket_usage(&observational, "known", id), None);
info.usage_snapshot_converged = None;
assert_eq!(certified_bucket_usage(&info, "known", id), None);
}
#[tokio::test]
#[serial]
async fn persisted_scalar_projection_shares_cache_refresh_and_invalidation() {
@@ -140,6 +175,13 @@ async fn persisted_scalar_projection_shares_cache_refresh_and_invalidation() {
info.buckets_usage.insert("empty".to_string(), BucketUsageInfo::default());
info.bucket_sizes.insert("empty".to_string(), 0);
info.buckets_count = 2;
info.usage_snapshot_converged = Some(true);
info.scanner_cycle = Some(1);
info.scanner_epoch = Some(1);
let incarnation = uuid::Uuid::from_u128(7);
for usage in info.buckets_usage.values_mut() {
usage.bucket_incarnation = Some(incarnation.to_string());
}
save_config(
store.clone(),
DATA_USAGE_OBJ_NAME_PATH.as_str(),
@@ -151,11 +193,29 @@ async fn persisted_scalar_projection_shares_cache_refresh_and_invalidation() {
for (bucket, expected) in [("known", Some(42)), ("empty", Some(0)), ("missing", None)] {
assert_eq!(persisted_bucket_usage(store.clone(), bucket).await, expected);
assert_eq!(
load_bucket_usage_from_backend_cached(store.clone(), bucket, incarnation)
.await
.expect("project bucket usage"),
expected
);
let owned = load_data_usage_from_backend_cached(store.clone())
.await
.expect("owned cached snapshot");
assert_eq!(expected, owned.buckets_usage.get(bucket).map(|usage| usage.size));
}
assert_eq!(
load_bucket_usage_from_backend_cached(store.clone(), "known", uuid::Uuid::from_u128(8))
.await
.unwrap(),
None
);
assert_eq!(
load_bucket_usage_from_backend_cached(store.clone(), "known", uuid::Uuid::nil())
.await
.unwrap(),
None
);
info.buckets_usage.get_mut("known").expect("known bucket").size = 84;
info.bucket_sizes.insert("known".to_string(), 84);
save_config(
+80 -9
View File
@@ -11330,7 +11330,7 @@ pub(crate) async fn batch_shard_pread(requests: Vec<(std::path::PathBuf, usize,
let submitted = metrics.as_ref().map(|_| std::time::Instant::now());
tokio::task::spawn_blocking(move || {
use rustfs_io_metrics::LocalBatchReadStage;
use std::os::unix::fs::FileExt;
use std::os::unix::fs::{FileExt, OpenOptionsExt};
let execution_started = metrics.as_ref().map(|_| std::time::Instant::now());
if let (Some(metrics), Some(submitted)) = (&metrics, submitted) {
@@ -11341,17 +11341,24 @@ pub(crate) async fn batch_shard_pread(requests: Vec<(std::path::PathBuf, usize,
for (file_path, offset, length) in requests {
let mut bytes_read = 0;
let r = (|| -> Result<Bytes> {
let meta = measure_local_batch_stage(metrics.as_deref(), LocalBatchReadStage::Metadata, || {
std::fs::metadata(&file_path).map_err(DiskError::from)
})?;
let end = offset.checked_add(length).ok_or(DiskError::FileCorrupt)?;
if meta.len() < u64::try_from(end).unwrap_or(u64::MAX) {
let end = u64::try_from(end).map_err(|_| DiskError::FileCorrupt)?;
// Bind validation and reads to one descriptor. Nonblocking open
// lets corrupt non-regular entries fail without waiting on a FIFO.
let file = measure_local_batch_stage(metrics.as_deref(), LocalBatchReadStage::Open, || {
std::fs::OpenOptions::new()
.read(true)
.custom_flags(libc::O_NONBLOCK)
.open(&file_path)
.map_err(DiskError::from)
})?;
let meta = measure_local_batch_stage(metrics.as_deref(), LocalBatchReadStage::Metadata, || {
file.metadata().map_err(DiskError::from)
})?;
if !meta.is_file() || meta.len() < end {
return Err(DiskError::FileCorrupt);
}
let file = measure_local_batch_stage(metrics.as_deref(), LocalBatchReadStage::Open, || {
std::fs::File::open(&file_path).map_err(DiskError::from)
})?;
let mut buf = vec![0u8; length];
if length > 0 {
measure_local_batch_stage(metrics.as_deref(), LocalBatchReadStage::Read, || {
@@ -31045,7 +31052,7 @@ mod test {
assert_eq!(recorder.counter_value(counter, &[("outcome", "error")]), 2);
assert_eq!(recorder.counter_value("rustfs_io_get_object_local_batch_read_bytes_total", &[]), 9);
let stages = "rustfs_io_get_object_local_batch_stage_seconds";
for (stage, success, errors) in [("metadata", 3, 1), ("open", 2, 0), ("read", 1, 0), ("execution", 0, 1)] {
for (stage, success, errors) in [("open", 3, 1), ("metadata", 3, 0), ("read", 1, 0), ("execution", 0, 1)] {
assert_eq!(
recorder
.histogram_values(stages, &[("stage", stage), ("outcome", "success")])
@@ -31114,6 +31121,70 @@ mod test {
assert_eq!(recorder.counter_value("rustfs_io_get_object_local_batch_read_bytes_total", &[]), 4);
}
#[cfg(unix)]
#[tokio::test]
async fn test_batch_shard_pread_rejects_nonregular_files_without_blocking() {
use std::os::unix::fs::OpenOptionsExt;
let dir = tempfile::tempdir().expect("nonregular shard fixture");
let fifo = dir.path().join("fifo");
assert!(
std::process::Command::new("mkfifo")
.arg(&fifo)
.status()
.expect("run POSIX mkfifo for the shard fixture")
.success()
);
let good = dir.path().join("good");
std::fs::write(&good, b"data").expect("write regular shard");
let mut batch = Box::pin(batch_shard_pread(vec![
(dir.path().to_owned(), 0, 0),
(fifo.clone(), 0, 0),
(fifo.clone(), 0, 4),
(good, 0, 4),
]));
let results = match tokio::time::timeout(Duration::from_secs(2), &mut batch).await {
Ok(results) => results,
Err(_) => {
// Release a regressed blocking open before failing, so runtime
// shutdown does not leave the test waiting on its worker.
let _release = std::fs::OpenOptions::new()
.read(true)
.write(true)
.custom_flags(libc::O_NONBLOCK)
.open(&fifo)
.expect("release blocked FIFO reader");
batch.await;
panic!("a nonregular shard must not block its batch");
}
};
for result in &results[..3] {
assert!(matches!(result, Err(DiskError::FileCorrupt)), "nonregular shard: {result:?}");
}
assert_eq!(results[3].as_ref().expect("later regular shard remains readable").as_ref(), b"data");
}
#[cfg(unix)]
#[tokio::test]
async fn test_batch_shard_pread_validates_descriptor_ranges() {
let dir = tempfile::tempdir().expect("range validation fixture");
let path = dir.path().join("shard");
std::fs::write(&path, b"01234567").expect("write range fixture");
let results = batch_shard_pread(vec![
(path.clone(), 3, 5),
(path.clone(), 8, 0),
(path.clone(), 8, 1),
(path.clone(), 9, 0),
(path, usize::MAX, 1),
])
.await;
assert_eq!(results[0].as_ref().expect("range ending at EOF").as_ref(), b"34567");
assert!(results[1].as_ref().expect("empty range at EOF").is_empty());
for result in &results[2..] {
assert!(matches!(result, Err(DiskError::FileCorrupt)), "invalid range: {result:?}");
}
}
#[cfg(unix)]
#[tokio::test]
async fn test_batch_shard_pread_basic() {
+4
View File
@@ -959,6 +959,10 @@ pub struct ObjectOptions {
/// Persisted bucket incarnation observed before authorization.
pub expected_bucket_incarnation_id: Option<Uuid>,
pub no_lock: bool,
/// Require complete participant observations for internal multipart size decisions.
/// This transient flag is never derived from client input or stored on disk.
#[doc(hidden)]
pub strict_multipart_part_sizes: bool,
/// A genuine target write guard; loss fences alone do not authorize a borrowed write.
#[doc(hidden)]
pub write_lock_context: Option<WriteLockContext>,
+431 -38
View File
@@ -280,7 +280,7 @@ pub(in crate::set_disk) fn resolve_read_part_from_responses(
continue;
}
if !parts[part_idx].etag.is_empty() {
if !parts[part_idx].etag.is_empty() && parts[part_idx].error.is_none() {
present_count += 1;
let part = &parts[part_idx];
let key = (part.etag.as_str(), part.number, part.size, part.actual_size, part.integrity.as_ref());
@@ -1118,9 +1118,9 @@ pub(in crate::set_disk) fn fill_deferred_bitrot_readers(
// offset can be advanced to the failing stripe. A parity reader that was
// opened eagerly during setup is pinned at the stripe-0 stream position
// and could never be engaged mid-object, so swap it for an unopened
// deferred reader carrying a stripe handle. The eager open already proved
// the shard is reachable; no shard bytes were read from it, and the
// ready/error bookkeeping that quorum decisions rely on is left untouched.
// deferred reader carrying a stripe handle. The eager setup already proved
// the shard is reachable; the ready/error bookkeeping that quorum decisions
// rely on is left untouched.
// Gate off (default): keep the eagerly opened parity readers exactly as
// before — the lockstep path reads them on every stripe.
if !demand_bound_lockstep {
@@ -1217,8 +1217,8 @@ pub(in crate::set_disk) fn record_bitrot_reader_setup_strategy(
}
/// When all online shards are local and mmap-read is enabled, materialize
/// shard bytes with one `batch_shard_pread` instead of per-shard
/// `spawn_blocking` via `open_disk_reader`.
/// shard bytes in batches instead of per-shard `spawn_blocking` via
/// `open_disk_reader`. Demand-bound reads add parity batches only as needed.
#[cfg(unix)]
#[allow(clippy::too_many_arguments)]
async fn try_create_bitrot_readers_via_batch_pread(
@@ -1232,6 +1232,10 @@ async fn try_create_bitrot_readers_via_batch_pread(
shard_size: usize,
checksum_algo: HashAlgorithm,
skip_verify_bitrot: bool,
data_shards: usize,
parity_shards: usize,
mode: BitrotReaderSetupMode,
strategy: BitrotReaderSetupStrategy,
) -> Option<BitrotReaderSetup> {
use crate::disk::local::batch_shard_pread;
use rustfs_io_metrics::{LocalBatchReadDecision, record_local_batch_read_decision};
@@ -1275,42 +1279,62 @@ async fn try_create_bitrot_readers_via_batch_pread(
return None;
}
let requests: Vec<_> = batch_items.iter().map(|(_, p, off, len)| (p.clone(), *off, *len)).collect();
#[cfg(test)]
for (disk_index, _, _, _) in &batch_items {
disk_call_counters::record(object, disk_call_counters::KIND_DISK_SHARD_OPEN, *disk_index);
}
record_decision(LocalBatchReadDecision::Selected);
let batch_results = batch_shard_pread(requests).await;
// Legacy lockstep consumes every shard, so keep its coalesced batch even
// when setup prefers data blocks. Only demand-bound decode can leave the
// parity readers unopened; preading them here would discard their bytes
// when fill_deferred_bitrot_readers installs stripe-aligned reserves.
let data_first = strategy != BitrotReaderSetupStrategy::AllShards
&& crate::erasure::coding::decode::get_lockstep_data_shards_only_enabled();
let mut setup = BitrotReaderSetup::new(disks.len());
for (i, (idx, _, _, _)) in batch_items.iter().enumerate() {
setup.mark_scheduled(*idx);
match &batch_results[i] {
Ok(bytes) => {
let reader = BitrotReader::new(
ShardReader::InMemory(Cursor::new(bytes.clone())),
shard_size,
checksum_algo.clone(),
skip_verify_bitrot,
);
setup.apply_reader_result(*idx, Ok(Some(reader)));
}
Err(e) => {
setup.apply_reader_result(*idx, Err(e.clone()));
}
}
}
for (idx, disk_op) in disks.iter().enumerate() {
if setup.scheduled[idx] {
continue;
}
setup.mark_scheduled(idx);
if disk_op.is_none() {
for (idx, disk) in disks.iter().enumerate() {
if disk.is_none() {
setup.mark_scheduled(idx);
setup.apply_reader_result(idx, Ok(None));
}
}
let mut batch_size = if data_first {
batch_items.iter().take_while(|(idx, _, _, _)| *idx < data_shards).count()
} else {
batch_items.len()
};
let mut pending = batch_items.into_iter();
record_decision(LocalBatchReadDecision::Selected);
loop {
let mut indices = Vec::with_capacity(batch_size.min(pending.len()));
let requests: Vec<_> = pending
.by_ref()
.take(batch_size)
.map(|(idx, path, offset, length)| {
setup.mark_scheduled(idx);
indices.push(idx);
#[cfg(test)]
disk_call_counters::record(object, disk_call_counters::KIND_DISK_SHARD_OPEN, idx);
(path, offset, length)
})
.collect();
if !requests.is_empty() {
for (idx, result) in indices.into_iter().zip(batch_shard_pread(requests).await) {
let result = result.map(|bytes| {
Some(BitrotReader::new(
ShardReader::InMemory(Cursor::new(bytes)),
shard_size,
checksum_algo.clone(),
skip_verify_bitrot,
))
});
setup.apply_reader_result(idx, result);
}
}
if !data_first || setup.has_setup_quorum(data_shards, parity_shards, mode) || pending.len() == 0 {
break;
}
// A missing/corrupt data file needs replacement parity. Reuse the
// canonical target, including the extra verification source required
// for reconstruction, and retry only the outstanding deficit.
batch_size = setup
.scheduling_target(data_shards, parity_shards, mode)
.saturating_sub(setup.available_shards());
}
Some(setup)
}
@@ -1328,6 +1352,10 @@ async fn try_create_bitrot_readers_via_batch_pread(
_shard_size: usize,
_checksum_algo: HashAlgorithm,
_skip_verify_bitrot: bool,
_data_shards: usize,
_parity_shards: usize,
_mode: BitrotReaderSetupMode,
_strategy: BitrotReaderSetupStrategy,
) -> Option<BitrotReaderSetup> {
rustfs_io_metrics::record_local_batch_read_decision(rustfs_io_metrics::LocalBatchReadDecision::UnsupportedPlatform, None);
None
@@ -1517,6 +1545,10 @@ pub(in crate::set_disk) async fn create_bitrot_readers_until_quorum_with_prefere
shard_size,
checksum_algo.clone(),
skip_verify_bitrot,
data_shards,
parity_shards,
mode,
strategy,
)
.await
{
@@ -7889,6 +7921,329 @@ mod tests {
(dirs, disks)
}
#[cfg(unix)]
async fn local_batch_quorum_fixture(
bucket: &str,
object: &str,
) -> (Vec<TempDir>, Vec<Option<DiskStore>>, Vec<FileInfo>, Vec<u8>) {
let payload: Vec<_> = (0..1024 * 1024)
.map(|index| u8::try_from(index % 251).expect("bounded byte"))
.collect();
let (dirs, disks) = call_counter_local_disks(bucket, 8).await;
let mut files = inline_metadata_fanout_fileinfos_with_geometry(bucket, object, &payload, false, 4, 4).await;
for (file, disk) in files.iter_mut().zip(&disks) {
let path = format!("{object}/{}/part.1", file.data_dir.expect("fixture data directory"));
disk.as_ref()
.expect("fixture disk")
.write_all(bucket, &path, file.data.take().expect("encoded fixture shard"))
.await
.expect("external shard fixture should write");
}
(dirs, disks, files, payload)
}
#[cfg(unix)]
async fn local_batch_quorum_setup(
files: &[FileInfo],
disks: &[Option<DiskStore>],
bucket: &str,
object: &str,
mode: BitrotReaderSetupMode,
) -> BitrotReaderSetup {
let erasure = coding::Erasure::new_with_options(4, 4, files[0].erasure.block_size, false);
create_bitrot_readers_until_quorum_with_preference(
files,
disks,
bucket,
object,
1,
0,
erasure.shard_file_offset(0, 1024 * 1024, 1024 * 1024),
erasure.shard_size(),
HashAlgorithm::HighwayHash256S,
false,
true,
4,
4,
mode,
false,
None,
None,
)
.await
}
#[cfg(unix)]
async fn local_batch_quorum_decode(setup: BitrotReaderSetup, block_size: usize) -> (Vec<u8>, Option<std::io::Error>) {
let erasure = coding::Erasure::new_with_options(4, 4, block_size, false);
let mut output = Vec::new();
let (_, error) = erasure
.decode_with_stripe_handles_and_reopeners(
&mut output,
setup.readers,
0,
1024 * 1024,
1024 * 1024,
None,
setup.deferred_stripe_handles,
setup.deferred_reopeners,
)
.await;
(output, error)
}
#[cfg(unix)]
#[tokio::test]
#[serial_test::serial]
async fn local_batch_quorum_healthy_reads_only_data_when_demand_bound() {
temp_env::async_with_vars(
[
(ENV_RUSTFS_GET_DATA_BLOCKS_FIRST_READER_SETUP, Some("true")),
("RUSTFS_GET_LOCKSTEP_DATA_SHARDS_ONLY_ENABLE", Some("false")),
],
crate::erasure::coding::decode::with_decode_read_policy(
crate::erasure::coding::decode::DecodeReadPolicy::DemandBound,
async {
let bucket = "batch-data-only";
let object = "healthy-one-mib";
let (_dirs, disks, files, payload) = local_batch_quorum_fixture(bucket, object).await;
let calls = disk_call_counters::observe(object);
let setup = local_batch_quorum_setup(&files, &disks, bucket, object, BitrotReaderSetupMode::ReadQuorum).await;
assert_eq!(calls.total(disk_call_counters::KIND_DISK_SHARD_OPEN), 4);
assert_eq!(setup.available_shards(), 4);
assert_eq!(setup.deferred_shards(), 4);
for index in 4..8 {
assert_eq!(calls.for_disk(disk_call_counters::KIND_DISK_SHARD_OPEN, index), 0);
assert!(!setup.attempted[index], "parity must stay unopened until needed");
}
let (output, error) = local_batch_quorum_decode(setup, files[0].erasure.block_size).await;
assert!(error.is_none(), "healthy decode failed: {error:?}");
assert_eq!(output, payload);
assert_eq!(calls.total(disk_call_counters::KIND_DISK_SHARD_OPEN), 4);
},
),
)
.await;
}
#[cfg(unix)]
#[tokio::test]
#[serial_test::serial]
async fn local_batch_quorum_preserves_legacy_and_all_shards_override() {
for (data_first, demand_bound) in [("true", "false"), ("false", "true")] {
temp_env::async_with_vars(
[
(ENV_RUSTFS_GET_DATA_BLOCKS_FIRST_READER_SETUP, Some(data_first)),
("RUSTFS_GET_LOCKSTEP_DATA_SHARDS_ONLY_ENABLE", Some(demand_bound)),
],
async {
let bucket = "batch-legacy";
let object = "all-shards";
let (_dirs, disks, files, payload) = local_batch_quorum_fixture(bucket, object).await;
let calls = disk_call_counters::observe(object);
let setup = local_batch_quorum_setup(&files, &disks, bucket, object, BitrotReaderSetupMode::ReadQuorum).await;
assert_eq!(calls.total(disk_call_counters::KIND_DISK_SHARD_OPEN), 8);
assert_eq!(setup.available_shards(), 8);
let (output, error) = local_batch_quorum_decode(setup, files[0].erasure.block_size).await;
assert!(error.is_none(), "legacy decode failed: {error:?}");
assert_eq!(output, payload);
},
)
.await;
}
}
#[cfg(unix)]
#[tokio::test]
#[serial_test::serial]
async fn local_batch_quorum_recovers_missing_truncated_and_bitrot_data() {
temp_env::async_with_vars(
[
(ENV_RUSTFS_GET_DATA_BLOCKS_FIRST_READER_SETUP, Some("true")),
("RUSTFS_GET_LOCKSTEP_DATA_SHARDS_ONLY_ENABLE", Some("true")),
],
async {
for fault in ["missing", "truncated", "bitrot"] {
let bucket = "batch-degraded";
let object = format!("degraded-{fault}");
let (dirs, disks, files, payload) = local_batch_quorum_fixture(bucket, &object).await;
let path = dirs[0]
.path()
.join(bucket)
.join(&object)
.join(files[0].data_dir.expect("fixture directory").to_string())
.join("part.1");
match fault {
"missing" => tokio::fs::remove_file(&path).await.expect("remove data shard"),
"truncated" => tokio::fs::write(&path, b"short").await.expect("truncate data shard"),
_ => {
let mut bytes = tokio::fs::read(&path).await.expect("read bitrot fixture");
bytes[32] ^= 0xff;
tokio::fs::write(&path, bytes)
.await
.expect("corrupt data without changing length");
}
}
let setup =
local_batch_quorum_setup(&files, &disks, bucket, &object, BitrotReaderSetupMode::ReadQuorum).await;
assert_eq!(setup.available_shards(), 4);
if fault == "bitrot" {
assert_eq!(setup.attempted_shards(), 4, "bitrot is detected on decode, not pread");
} else {
assert_eq!(setup.attempted_shards(), 5, "setup reads only one replacement parity shard");
}
let (output, error) = local_batch_quorum_decode(setup, files[0].erasure.block_size).await;
assert!(error.is_none(), "{fault} should recover: {error:?}");
assert_eq!(output, payload, "{fault} must reconstruct the exact body");
}
},
)
.await;
}
#[cfg(unix)]
#[tokio::test]
#[serial_test::serial]
async fn local_batch_quorum_keeps_extra_reconstruction_verification_source() {
temp_env::async_with_vars(
[
(ENV_RUSTFS_GET_CODEC_STREAMING_DATA_BLOCKS_FIRST_READER_SETUP, Some("true")),
("RUSTFS_GET_LOCKSTEP_DATA_SHARDS_ONLY_ENABLE", Some("true")),
],
async {
let bucket = "batch-verification";
let object = "missing-data";
let (_dirs, mut disks, files, payload) = local_batch_quorum_fixture(bucket, object).await;
disks[0] = None;
let calls = disk_call_counters::observe(object);
let setup =
local_batch_quorum_setup(&files, &disks, bucket, object, BitrotReaderSetupMode::VerifyReconstruction).await;
assert_eq!(setup.available_shards(), 5, "degraded reconstruction needs quorum plus verification");
assert_eq!(calls.total(disk_call_counters::KIND_DISK_SHARD_OPEN), 5);
assert!(setup.ready[4] && setup.ready[5], "two parity sources must be available");
assert!(!setup.attempted[6] && !setup.attempted[7], "unneeded parity stays deferred");
let (output, error) = local_batch_quorum_decode(setup, files[0].erasure.block_size).await;
assert!(error.is_none(), "verified reconstruction failed: {error:?}");
assert_eq!(output, payload);
},
)
.await;
}
#[cfg(unix)]
#[tokio::test]
#[serial_test::serial]
async fn local_batch_quorum_fails_without_read_quorum() {
temp_env::async_with_vars(
[
(ENV_RUSTFS_GET_DATA_BLOCKS_FIRST_READER_SETUP, Some("true")),
("RUSTFS_GET_LOCKSTEP_DATA_SHARDS_ONLY_ENABLE", Some("true")),
],
async {
let bucket = "batch-no-quorum";
let object = "all-data-offline";
let (_dirs, mut disks, files, _payload) = local_batch_quorum_fixture(bucket, object).await;
disks[..5].fill(None);
let setup = local_batch_quorum_setup(&files, &disks, bucket, object, BitrotReaderSetupMode::ReadQuorum).await;
assert_eq!(setup.available_shards(), 3);
assert!(!setup.has_setup_quorum(4, 4, BitrotReaderSetupMode::ReadQuorum));
assert_eq!(setup.deferred_shards(), 0);
let (output, error) = local_batch_quorum_decode(setup, files[0].erasure.block_size).await;
assert!(error.is_some(), "insufficient shards must fail closed");
assert!(output.is_empty(), "no unverified bytes may be emitted");
},
)
.await;
}
#[cfg(unix)]
#[tokio::test]
#[serial_test::serial]
async fn local_batch_quorum_recovers_late_bitrot_after_range_offset() {
temp_env::async_with_vars(
[
(ENV_RUSTFS_GET_DATA_BLOCKS_FIRST_READER_SETUP, Some("true")),
("RUSTFS_GET_LOCKSTEP_DATA_SHARDS_ONLY_ENABLE", Some("true")),
],
async {
let bucket = "batch-ranged";
let object = "late-bitrot";
let (_dirs, disks) = call_counter_local_disks(bucket, 8).await;
let erasure = coding::Erasure::new_with_options(4, 4, 64, false);
let payload: Vec<u8> = (0..192).collect();
let mut shards = vec![Vec::new(); 8];
for block in payload.chunks(64) {
for (index, shard) in erasure
.encode_data(block)
.expect("encode range stripe")
.into_iter()
.enumerate()
{
let mut writer = coding::BitrotWriterWrapper::new(
coding::CustomWriter::new_inline_buffer(),
erasure.shard_size(),
HashAlgorithm::HighwayHash256S,
);
writer.write(&shard).await.expect("write range stripe");
writer.shutdown().await.expect("finish range stripe");
shards[index].extend(writer.into_inline_data().expect("range stripe bytes"));
}
}
// Corrupt the third stripe; the first requested stripe is healthy.
shards[0][2 * (32 + erasure.shard_size()) + 32] ^= 0xff;
let mut files = vec![FileInfo::new(object, 4, 4); 8];
let data_dir = Uuid::new_v4();
for ((file, disk), bytes) in files.iter_mut().zip(&disks).zip(shards) {
file.data_dir = Some(data_dir);
disk.as_ref()
.expect("fixture disk")
.write_all(bucket, &format!("{object}/{data_dir}/part.1"), Bytes::from(bytes))
.await
.expect("write multi-stripe shard");
}
let calls = disk_call_counters::observe(object);
let setup = create_bitrot_readers_until_quorum_with_preference(
&files,
&disks,
bucket,
object,
1,
erasure.shard_size(),
2 * erasure.shard_size(),
erasure.shard_size(),
HashAlgorithm::HighwayHash256S,
false,
true,
4,
4,
BitrotReaderSetupMode::ReadQuorum,
false,
None,
None,
)
.await;
assert_eq!(calls.total(disk_call_counters::KIND_DISK_SHARD_OPEN), 4);
let mut output = Vec::new();
let (written, error) = erasure
.decode_with_stripe_handles_and_reopeners(
&mut output,
setup.readers,
64,
128,
payload.len(),
None,
setup.deferred_stripe_handles,
setup.deferred_reopeners,
)
.await;
assert!(error.is_none(), "late bitrot must recover from stripe-aligned parity: {error:?}");
assert_eq!(written, 128);
assert_eq!(output, payload[64..]);
},
)
.await;
}
#[cfg(unix)]
#[test]
fn local_batch_read_decisions_follow_the_actual_reader_setup_entry() {
@@ -7957,6 +8312,10 @@ mod tests {
4,
HashAlgorithm::None,
false,
1,
1,
BitrotReaderSetupMode::ReadQuorum,
BitrotReaderSetupStrategy::AllShards,
)
.await
.is_none(),
@@ -7974,6 +8333,10 @@ mod tests {
4,
HashAlgorithm::HighwayHash256S,
false,
1,
1,
BitrotReaderSetupMode::ReadQuorum,
BitrotReaderSetupStrategy::AllShards,
)
.await
.is_none(),
@@ -7993,6 +8356,10 @@ mod tests {
4,
HashAlgorithm::None,
false,
1,
1,
BitrotReaderSetupMode::ReadQuorum,
BitrotReaderSetupStrategy::AllShards,
)
.await
.is_none()
@@ -8020,6 +8387,10 @@ mod tests {
4,
HashAlgorithm::None,
false,
1,
1,
BitrotReaderSetupMode::ReadQuorum,
BitrotReaderSetupStrategy::AllShards,
)
.await
.is_none(),
@@ -8037,6 +8408,10 @@ mod tests {
4,
HashAlgorithm::None,
false,
1,
1,
BitrotReaderSetupMode::ReadQuorum,
BitrotReaderSetupStrategy::AllShards,
)
.await
.is_none()
@@ -8054,6 +8429,10 @@ mod tests {
4,
HashAlgorithm::None,
false,
1,
1,
BitrotReaderSetupMode::ReadQuorum,
BitrotReaderSetupStrategy::AllShards,
)
.await
.is_none()
@@ -12823,6 +13202,20 @@ mod tests {
assert_eq!(err, DiskError::ErasureReadQuorum);
}
#[test]
fn resolve_read_part_does_not_count_error_response_with_etag_as_a_vote() {
let healthy = read_part_test_part(1, "same-etag");
let mut failed = healthy.clone();
failed.error = Some("metadata read failed".to_owned());
let responses = vec![Some(vec![healthy.clone()]), Some(vec![healthy]), Some(vec![failed])];
let err = resolve_read_part_from_responses("bucket", "upload/part.1.meta", 1, 0, 1, &responses, 3)
.expect_err("an error-bearing response must not supply the third vote");
assert_eq!(err, DiskError::ErasureReadQuorum);
let part = resolve_read_part_from_responses("bucket", "upload/part.1.meta", 1, 0, 1, &responses, 2)
.expect("the two healthy matching responses remain a valid quorum");
assert!(part.error.is_none());
}
#[test]
fn resolve_read_part_accepts_alternate_missing_error_markers() {
let responses = vec![
@@ -338,17 +338,26 @@ fn abort_slot(
Ok(())
}
fn abort_certificate_cohort(
pub(in crate::set_disk) fn abort_certificate_cohort(
certificate: &MultipartAbortCertificate,
ids: &[Option<Uuid>],
authority: &ProofVerificationContext,
) -> Result<()> {
certificate.verify_with_authority(authority)?;
if ids.len() != certificate.disk_ids.len()
|| ids.iter().copied().collect::<Option<Vec<_>>>().as_ref() != Some(&certificate.disk_ids)
{
if ids.len() != certificate.disk_ids.len() {
return Err(DiskError::FileCorrupt);
}
let mut missing = false;
for (actual, expected) in ids.iter().zip(&certificate.disk_ids) {
match actual {
Some(actual) if actual != expected => return Err(DiskError::FileCorrupt),
Some(_) => {}
None => missing = true,
}
}
if missing {
return Err(DiskError::ErasureWriteQuorum);
}
Ok(())
}
+29 -1
View File
@@ -283,6 +283,23 @@ fn ensure_decommission_tier_free_version_commit_fence(bucket: &str, object: &str
Ok(())
}
/// Names a `bucket/object` key prefix whose best-effort rename-tail repairs
/// are not submitted. A root-heal fixture arms it so the explicit admin heal
/// remains the only owner of objects written while their target was offline.
#[cfg(feature = "e2e-test-hooks")]
const RENAME_TAIL_HEAL_SUPPRESS_PREFIX_ENV: &str = "RUSTFS_E2E_SUPPRESS_RENAME_TAIL_HEAL_PREFIX";
#[cfg(feature = "e2e-test-hooks")]
fn rename_tail_heal_suppressed_for_e2e(bucket: &str, object: Option<&str>) -> bool {
let Some(prefix) = std::env::var(RENAME_TAIL_HEAL_SUPPRESS_PREFIX_ENV)
.ok()
.filter(|prefix| !prefix.is_empty())
else {
return false;
};
object.is_some_and(|object| format!("{bucket}/{object}").starts_with(&prefix))
}
impl SetDisks {
pub(super) async fn require_current_restore_operation_id(
&self,
@@ -486,10 +503,16 @@ pub(crate) fn strip_internal_multipart_metadata(metadata: &mut HashMap<String, S
rustfs_utils::http::metadata_compat::remove_str(metadata, SUFFIX_BUCKET_INCARNATION_ID);
}
fn should_persist_encryption_original_size(metadata: &HashMap<String, String>) -> bool {
pub(crate) fn should_persist_encryption_original_size(metadata: &HashMap<String, String>) -> bool {
metadata.keys().any(|key| is_object_encryption_marker(key))
}
/// Reject ambiguous part sizes unless the upload records a transformation.
pub(crate) fn multipart_part_logical_size(stored_size: usize, actual_size: i64, transformed: bool) -> Option<usize> {
let actual_size = usize::try_from(actual_size).ok()?;
(transformed || actual_size == stored_size).then_some(actual_size)
}
/// Per-set memoized capacity dirty scope.
///
/// The disk endpoint/path identity of each slot in a set is immutable for the
@@ -5143,6 +5166,11 @@ impl SetDisks {
}
}
#[cfg(feature = "e2e-test-hooks")]
if rename_tail_heal_suppressed_for_e2e(&request.bucket, request.object_prefix.as_deref()) {
return Ok(());
}
let _ = rustfs_heal_contracts::heal_channel::send_heal_request(request).await;
Ok(())
}
+518 -54
View File
@@ -43,10 +43,11 @@ use super::super::{
complete_multipart_part_error, complete_multipart_part_error_result, complete_part_checksum, contains_key_str,
create_bitrot_writer, debug, disk, error, file_info_is_valid_for_metadata, get_complete_multipart_md5, get_header_map,
get_str, insert_str, is_err_object_not_found, is_err_version_not_found, is_min_allowed_part_size,
log_multipart_write_quorum_failure, parts_after_marker, path_join_buf, prepare_complete_multipart_parts,
record_compression_total_memory, record_multipart_write_quorum_failure, reduce_read_quorum_errs, reduce_write_quorum_errs,
remove_header_map, resolve_write_layout, restore_commit_operation_id_from_metadata, should_persist_encryption_original_size,
strip_internal_multipart_metadata, to_object_err, warn,
log_multipart_write_quorum_failure, multipart_part_logical_size, parts_after_marker, path_join_buf,
prepare_complete_multipart_parts, record_compression_total_memory, record_multipart_write_quorum_failure,
reduce_read_quorum_errs, reduce_write_quorum_errs, remove_header_map, resolve_write_layout,
restore_commit_operation_id_from_metadata, should_persist_encryption_original_size, strip_internal_multipart_metadata,
to_object_err, warn,
};
use super::bitrot_self_verify::{BitrotSelfVerifyTarget, drop_failed_writer_disks, verify_written_bitrot_shards};
#[cfg(test)]
@@ -423,6 +424,8 @@ impl StaleMultipartCleanupGuard {
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum MultipartCommitPause {
NewUploadBeforeLockLost,
#[cfg(test)]
ListPartsAfterMetadata,
PutPartBeforeLockAcquire,
PutPartBeforeLockLost,
PutPartAfterCapacityAdmission,
@@ -837,7 +840,8 @@ where
}
if successful_responses < read_quorum {
if empty_upload_fallback_possible(successful_responses, &errs) {
// A single missing directory cannot speak for unavailable disks.
if !errs.is_empty() && errs.iter().all(|err| matches!(err, Some(DiskError::FileNotFound))) {
return Err(DiskError::FileNotFound);
}
@@ -847,6 +851,51 @@ where
Ok((errs, object_parts))
}
// Size-dependent decisions must not discard a part merely because only a
// minority of its files was observed. All participants must answer, and each
// observed part must have a complete file/metadata pair on a read quorum.
fn validate_part_size_observations(
errs: &[Option<DiskError>],
object_parts: &[Vec<String>],
read_quorum: usize,
) -> disk::error::Result<Vec<usize>> {
if errs.iter().any(Option::is_some) {
return Err(DiskError::ErasureReadQuorum);
}
let mut counts: HashMap<usize, usize> = HashMap::new();
for entries in object_parts {
let mut observed: HashMap<usize, u8> = HashMap::new();
for entry in entries {
let Some(name) = entry.strip_prefix("part.") else {
continue;
};
let (number, flag) = match name.strip_suffix(".meta") {
Some(number) => (number, 2),
None => (name, 1),
};
let parsed = number.parse::<usize>().map_err(|_| DiskError::FileCorrupt)?;
if !(1..=MAX_PARTS_COUNT).contains(&parsed)
|| !number.as_bytes().first().is_some_and(|digit| (b'1'..=b'9').contains(digit))
{
return Err(DiskError::FileCorrupt);
}
*observed.entry(parsed).or_default() |= flag;
}
for (number, flags) in observed {
if flags != 3 {
return Err(DiskError::FileCorrupt);
}
*counts.entry(number).or_default() += 1;
}
}
if counts.values().any(|&count| count < read_quorum) {
return Err(DiskError::ErasureReadQuorum);
}
let mut part_numbers: Vec<_> = counts.into_keys().collect();
part_numbers.sort_unstable();
Ok(part_numbers)
}
fn reduce_quorum_part_numbers(object_parts: Vec<Vec<String>>, read_quorum: usize) -> Vec<usize> {
let mut part_quorum_map: HashMap<usize, usize> = HashMap::new();
@@ -1017,7 +1066,7 @@ impl SetDisks {
]),
SLASH_SEPARATOR
);
let part_numbers = match Self::list_parts(&online_disks, &part_path, read_quorum).await {
let part_numbers = match Self::list_parts(&online_disks, &part_path, read_quorum, true).await {
Ok(parts) => parts,
Err(DiskError::FileNotFound) => return Ok(0),
Err(err) => return Err(to_object_err(err.into(), vec![bucket, object, upload_id])),
@@ -1029,28 +1078,45 @@ impl SetDisks {
.iter()
.map(|number| format!("{part_path}part.{number}.meta"))
.collect::<Vec<_>>();
let existing_parts =
Self::read_parts(&online_disks, RUSTFS_META_MULTIPART_BUCKET, &part_meta_paths, &part_numbers, read_quorum)
.await
.map_err(|err| to_object_err(err.into(), vec![bucket, object, upload_id]))?;
let existing_parts = Self::read_parts(
&online_disks,
RUSTFS_META_MULTIPART_BUCKET,
&part_meta_paths,
&part_numbers,
fi.write_quorum(self.default_write_quorum()),
)
.await
.map_err(|err| to_object_err(err.into(), vec![bucket, object, upload_id]))?;
existing_parts.into_iter().try_fold(0_u64, |total, part| {
if part.error.is_some() || part.number == replacing_part {
return if part.error.is_some() {
Err(Error::PartMissingOrCorrupt)
} else {
Ok(total)
};
}
let part_size = u64::try_from(part.actual_size).map_err(|_| Error::PartMissingOrCorrupt)?;
total.checked_add(part_size).ok_or_else(|| {
Error::InvalidArgument(
"multipart upload".to_string(),
SUFFIX_MAX_TOTAL_OBJECT_SIZE.to_string(),
"logical size overflow".to_string(),
)
if existing_parts.len() != part_numbers.len() {
return Err(Error::PartMissingOrCorrupt);
}
let transformed = fi.is_compressed() || should_persist_encryption_original_size(&fi.metadata);
existing_parts
.into_iter()
.zip(part_numbers)
.try_fold(0_u64, |total, (part, number)| {
if part.number != number {
return Err(Error::PartMissingOrCorrupt);
}
if part.error.is_some() || part.number == replacing_part {
return if part.error.is_some() {
Err(Error::PartMissingOrCorrupt)
} else {
Ok(total)
};
}
let part_size = multipart_part_logical_size(part.size, part.actual_size, transformed)
.and_then(|size| u64::try_from(size).ok())
.ok_or(Error::PartMissingOrCorrupt)?;
total.checked_add(part_size).ok_or_else(|| {
Error::InvalidArgument(
"multipart upload".to_string(),
SUFFIX_MAX_TOTAL_OBJECT_SIZE.to_string(),
"logical size overflow".to_string(),
)
})
})
})
}
async fn discover_multipart_upload_paths(
@@ -1282,6 +1348,7 @@ impl SetDisks {
disks: &[Option<DiskStore>],
part_path: &str,
read_quorum: usize,
strict_size: bool,
) -> disk::error::Result<Vec<usize>> {
let mut futures = Vec::with_capacity(disks.len());
let part_path = part_path.to_string();
@@ -1290,21 +1357,24 @@ impl SetDisks {
let part_path = part_path.clone();
futures.push(async move {
if let Some(disk) = disk {
disk.list_dir(RUSTFS_META_MULTIPART_BUCKET, RUSTFS_META_MULTIPART_BUCKET, part_path.as_str(), -1)
match disk
.list_dir(RUSTFS_META_MULTIPART_BUCKET, RUSTFS_META_MULTIPART_BUCKET, part_path.as_str(), -1)
.await
{
Err(DiskError::FileNotFound) if strict_size => Ok(Vec::new()),
result => result,
}
} else {
Err(DiskError::DiskNotFound)
}
});
}
let mut errs = Vec::with_capacity(disks.len());
let mut object_parts = Vec::with_capacity(disks.len());
let (collected_errs, collected_parts) = collect_list_parts_results(futures, read_quorum).await?;
errs.extend(collected_errs);
object_parts.extend(collected_parts);
let (errs, object_parts) = collect_list_parts_results(futures, read_quorum).await?;
if strict_size {
return validate_part_size_observations(&errs, &object_parts, read_quorum);
}
if let Some(err) = reduce_read_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, read_quorum) {
return Err(err);
}
@@ -2651,6 +2721,8 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
ensure_data_movement_upload_access(&fi, bucket, object, upload_id, opts)?;
ensure_multipart_bucket_incarnation(&self.ctx, &fi, bucket, object, upload_id, opts.expected_bucket_incarnation_id)
.await?;
#[cfg(test)]
pause_multipart_commit(bucket, object, MultipartCommitPause::ListPartsAfterMetadata).await;
let upload_id_path = Self::validated_multipart_upload_path(bucket, object, upload_id, opts.data_movement)?;
@@ -2699,16 +2771,17 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
SLASH_SEPARATOR
);
let mut part_numbers = match Self::list_parts(&online_disks, &part_path, read_quorum).await {
Ok(parts) => parts,
Err(err) => {
if err == DiskError::FileNotFound {
return Ok(ret);
}
let mut part_numbers =
match Self::list_parts(&online_disks, &part_path, read_quorum, opts.strict_multipart_part_sizes).await {
Ok(parts) => parts,
Err(err) => {
if err == DiskError::FileNotFound {
return Ok(ret);
}
return Err(to_object_err(err.into(), vec![bucket, object]));
}
};
return Err(to_object_err(err.into(), vec![bucket, object]));
}
};
if part_numbers.is_empty() {
return Ok(ret);
@@ -2725,11 +2798,29 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
.map(|v| format!("{part_path}part.{v}.meta"))
.collect::<Vec<String>>();
let object_parts =
Self::read_parts(&online_disks, RUSTFS_META_MULTIPART_BUCKET, &part_meta_paths, &part_numbers, read_quorum)
.await
.map_err(|e| to_object_err(e.into(), vec![bucket, object, upload_id]))?;
let object_parts = Self::read_parts(
&online_disks,
RUSTFS_META_MULTIPART_BUCKET,
&part_meta_paths,
&part_numbers,
if opts.strict_multipart_part_sizes {
fi.write_quorum(self.default_write_quorum())
} else {
read_quorum
},
)
.await
.map_err(|e| to_object_err(e.into(), vec![bucket, object, upload_id]))?;
if opts.strict_multipart_part_sizes
&& (object_parts.len() != part_numbers.len()
|| object_parts
.iter()
.zip(&part_numbers)
.any(|(part, number)| part.number != *number || part.error.is_some() || part.etag.is_empty()))
{
return Err(Error::PartMissingOrCorrupt);
}
let mut count = max_parts;
for part in object_parts.iter() {
@@ -7516,7 +7607,7 @@ mod tests {
}
#[tokio::test]
async fn collect_list_parts_results_returns_file_not_found_for_empty_upload_dirs() {
async fn collect_list_parts_results_rejects_unavailable_empty_upload_dirs() {
let tasks: Vec<_> = vec![
(5_u64, Err(DiskError::FileNotFound)),
(10, Err(DiskError::DiskNotFound)),
@@ -7531,11 +7622,372 @@ mod tests {
let err = collect_list_parts_results(tasks, 2)
.await
.expect_err("missing multipart directories should be treated as empty uploads");
.expect_err("an unavailable disk cannot prove an empty upload");
assert_eq!(err, DiskError::ErasureReadQuorum);
}
#[tokio::test(flavor = "multi_thread")]
#[serial]
async fn list_parts_rejects_disk_loss_after_upload_metadata_was_read() {
let (dirs, disks, set) = hermetic_set_disks(4).await;
let bucket = "multipart-list-loss-after-metadata";
let object = "object";
make_bucket_on_all(&disks, bucket).await;
let opts = ObjectOptions {
max_parity: true,
..Default::default()
};
let (upload_id, _) = stage_upload_with_create_opts(&set, bucket, object, &[1; 32], &opts).await;
let (fi, _) = set
.check_upload_id_exists(bucket, object, &upload_id, false)
.await
.expect("read initial metadata");
let upload_path = SetDisks::get_upload_id_dir(bucket, object, &upload_id);
let part_dir = dirs[0]
.path()
.join(RUSTFS_META_MULTIPART_BUCKET)
.join(upload_path)
.join(fi.data_dir.expect("upload data directory").to_string());
let held_dir = part_dir.with_extension("held-for-test");
let barrier = MultipartCommitBarrier::install(bucket, object, MultipartCommitPause::ListPartsAfterMetadata);
let query_set = Arc::clone(&set);
let query_id = upload_id.clone();
let query = tokio::spawn(async move {
query_set
.list_object_parts(bucket, object, &query_id, None, 100, &ObjectOptions::default())
.await
});
barrier.wait_until_paused().await;
tokio::fs::rename(&part_dir, &held_dir)
.await
.expect("simulate the surviving disk missing the part directory");
{
let mut live = set.disks.write().await;
for slot in 1..4 {
live[slot] = None;
}
}
barrier.release();
let result = query.await.expect("list task must finish");
drop(barrier);
tokio::fs::rename(&held_dir, &part_dir)
.await
.expect("restore the original directory");
{
let mut live = set.disks.write().await;
for slot in 1..4 {
live[slot] = Some(disks[slot].clone());
}
}
assert!(
matches!(result, Err(StorageError::InsufficientReadQuorum(..))),
"disk loss must fail rather than return an empty part list: {result:?}"
);
let recovered = set
.list_object_parts(bucket, object, &upload_id, None, 100, &ObjectOptions::default())
.await
.expect("listing resumes after the original disks return");
assert_eq!(recovered.parts.len(), 1);
}
#[tokio::test]
async fn collect_list_parts_results_proves_all_directories_missing() {
let tasks = (0..4)
.map(|_| async { Err::<Vec<String>, _>(DiskError::FileNotFound) })
.collect();
let err = collect_list_parts_results(tasks, 2)
.await
.expect_err("all missing directories are empty");
assert_eq!(err, DiskError::FileNotFound);
}
#[test]
fn strict_multipart_part_sizes_reject_incomplete_observations() {
let pair = || vec!["part.1".to_string(), "part.1.meta".to_string()];
let all_online = vec![None; 4];
assert!(validate_part_size_observations(&all_online, &[vec![], vec![], vec![], vec![]], 2).is_ok());
assert_eq!(
validate_part_size_observations(&all_online, &[pair(), pair(), vec![], vec![]], 2),
Ok(vec![1])
);
let unordered = vec!["part.9.meta", "part.1", "part.9", "part.1.meta", "part.1"]
.into_iter()
.map(str::to_owned)
.collect::<Vec<_>>();
assert_eq!(
validate_part_size_observations(&all_online, &[unordered.clone(), unordered, vec![], vec![]], 2),
Ok(vec![1, 9])
);
assert_eq!(
validate_part_size_observations(&all_online, &[pair(), vec![], vec![], vec![]], 2),
Err(DiskError::ErasureReadQuorum)
);
assert_eq!(
validate_part_size_observations(&all_online, &[vec!["part.1.meta".to_string()], pair(), pair(), vec![]], 2),
Err(DiskError::FileCorrupt)
);
for invalid in ["part.0", "part.10001", "part.01", "part.+1", "part.1x"] {
let entries = vec![invalid.to_string(), format!("{invalid}.meta")];
assert_eq!(
validate_part_size_observations(&all_online, &[entries.clone(), entries.clone(), entries.clone(), entries], 2),
Err(DiskError::FileCorrupt),
"noncanonical part names are not valid size evidence even on every disk"
);
}
assert_eq!(
validate_part_size_observations(
&[None, None, None, Some(DiskError::DiskNotFound)],
&[pair(), pair(), pair(), vec![]],
2
),
Err(DiskError::ErasureReadQuorum)
);
}
#[tokio::test]
async fn strict_multipart_capped_upload_rejects_unavailable_participant_and_recovers() {
let (_dirs, disks, set) = hermetic_set_disks(4).await;
let bucket = "strict-multipart-size-bucket";
let object = "object";
make_bucket_on_all(&disks, bucket).await;
let mut metadata = HashMap::new();
insert_str(&mut metadata, SUFFIX_MAX_TOTAL_OBJECT_SIZE, "10".to_string());
let opts = ObjectOptions {
max_parity: true,
user_defined: metadata,
..Default::default()
};
let (upload_id, _) = stage_upload_with_create_opts(&set, bucket, object, &[1; 6], &opts).await;
set.disks.write().await[3] = None;
let mut part = PutObjReader::from_vec(vec![2; 4]);
let err = set
.put_object_part(bucket, object, &upload_id, 2, &mut part, &ObjectOptions::default())
.await;
assert!(err.is_err(), "an unavailable disk must not produce a size admission: {err:?}");
set.disks.write().await[3] = Some(disks[3].clone());
// Two distinct part identities at read quorum cannot prove which size
// belongs to the acknowledged write; require the write quorum instead.
let (other_id, _) = stage_upload_with_create_opts(&set, bucket, "other", &[9; 2], &ObjectOptions::default()).await;
let (fi, _) = set
.check_upload_id_exists(bucket, object, &upload_id, false)
.await
.expect("read capped upload metadata");
let (other_fi, _) = set
.check_upload_id_exists(bucket, "other", &other_id, false)
.await
.expect("read other upload metadata");
let path = format!(
"{}/{}/part.1.meta",
SetDisks::get_upload_id_dir(bucket, object, &upload_id),
fi.data_dir.expect("data dir")
);
let other_path = format!(
"{}/{}/part.1.meta",
SetDisks::get_upload_id_dir(bucket, "other", &other_id),
other_fi.data_dir.expect("data dir")
);
let mut originals = Vec::new();
for disk in disks.iter().take(2) {
originals.push(
disk.read_all(RUSTFS_META_MULTIPART_BUCKET, &path)
.await
.expect("save original part metadata"),
);
let conflicting = disk
.read_all(RUSTFS_META_MULTIPART_BUCKET, &other_path)
.await
.expect("read conflicting part metadata");
disk.write_all(RUSTFS_META_MULTIPART_BUCKET, &path, conflicting)
.await
.expect("install split part metadata");
}
let mut split = PutObjReader::from_vec(vec![5; 4]);
let split_result = set
.put_object_part(bucket, object, &upload_id, 2, &mut split, &ObjectOptions::default())
.await;
for (disk, original) in disks.iter().zip(originals) {
disk.write_all(RUSTFS_META_MULTIPART_BUCKET, &path, original)
.await
.expect("restore original part metadata");
}
assert!(
split_result.is_err(),
"split part identities must not authorize a size decision: {split_result:?}"
);
let mut originals = Vec::new();
for disk in disks.iter().take(3) {
let original = disk
.read_all(RUSTFS_META_MULTIPART_BUCKET, &path)
.await
.expect("save original part metadata");
let mut stale = ObjectPartInfo::unmarshal(&original).expect("decode original part metadata");
stale.actual_size = 1;
disk.write_all(
RUSTFS_META_MULTIPART_BUCKET,
&path,
Bytes::from(stale.marshal_msg().expect("encode stale part")),
)
.await
.expect("install stale size on write quorum");
originals.push(original);
}
let mut undercounted = PutObjReader::from_vec(vec![6; 4]);
let result = set
.put_object_part(bucket, object, &upload_id, 2, &mut undercounted, &ObjectOptions::default())
.await;
for (disk, original) in disks.iter().zip(originals) {
disk.write_all(RUSTFS_META_MULTIPART_BUCKET, &path, original)
.await
.expect("restore original part metadata");
}
assert!(
matches!(result, Err(StorageError::PartMissingOrCorrupt)),
"smaller logical size must fail closed: {result:?}"
);
let mut originals = Vec::new();
for disk in disks.iter().take(3) {
let original = disk
.read_all(RUSTFS_META_MULTIPART_BUCKET, &path)
.await
.expect("save original part metadata");
let mut wrong_identity = ObjectPartInfo::unmarshal(&original).expect("decode original part metadata");
wrong_identity.number = 2;
disk.write_all(
RUSTFS_META_MULTIPART_BUCKET,
&path,
Bytes::from(wrong_identity.marshal_msg().expect("encode wrong identity")),
)
.await
.expect("install mismatched part identity on write quorum");
originals.push(original);
}
let list_result = set
.list_object_parts(
bucket,
object,
&upload_id,
None,
100,
&ObjectOptions {
strict_multipart_part_sizes: true,
..Default::default()
},
)
.await;
let mut replacement = PutObjReader::from_vec(vec![7; 4]);
let result = set
.put_object_part(bucket, object, &upload_id, 2, &mut replacement, &ObjectOptions::default())
.await;
for (disk, original) in disks.iter().zip(originals) {
disk.write_all(RUSTFS_META_MULTIPART_BUCKET, &path, original)
.await
.expect("restore original part metadata");
}
assert!(
matches!(list_result, Err(StorageError::PartMissingOrCorrupt)),
"strict listing must reject mismatched part identity: {list_result:?}"
);
assert!(
matches!(result, Err(StorageError::PartMissingOrCorrupt)),
"mismatched part identity must not bypass the cap: {result:?}"
);
let mut too_large = PutObjReader::from_vec(vec![3; 5]);
let err = set
.put_object_part(bucket, object, &upload_id, 2, &mut too_large, &ObjectOptions::default())
.await;
assert!(
matches!(err, Err(StorageError::EntityTooLarge(11, 10))),
"existing part must count: {err:?}"
);
let mut fits = PutObjReader::from_vec(vec![4; 4]);
set.put_object_part(bucket, object, &upload_id, 2, &mut fits, &ObjectOptions::default())
.await
.expect("the upload should resume at the exact limit after all disks recover");
}
#[tokio::test(flavor = "multi_thread")]
async fn capped_multipart_concurrent_parts_cannot_exceed_logical_limit() {
let (_dirs, disks, set) = hermetic_set_disks(4).await;
let bucket = "concurrent-capped-multipart";
let object = "object";
make_bucket_on_all(&disks, bucket).await;
let mut metadata = HashMap::new();
insert_str(&mut metadata, SUFFIX_MAX_TOTAL_OBJECT_SIZE, "10".to_owned());
let opts = ObjectOptions {
max_parity: true,
user_defined: metadata,
..Default::default()
};
let upload = set
.new_multipart_upload(bucket, object, &opts)
.await
.expect("create capped upload");
let barrier = Arc::new(tokio::sync::Barrier::new(3));
let mut tasks = Vec::new();
for (number, byte) in [(1_usize, 0x41_u8), (2, 0x42)] {
let set = Arc::clone(&set);
let barrier = Arc::clone(&barrier);
let upload_id = upload.upload_id.clone();
tasks.push(tokio::spawn(async move {
let mut body = PutObjReader::from_vec(vec![byte; 6]);
barrier.wait().await;
set.put_object_part(bucket, object, &upload_id, number, &mut body, &ObjectOptions::default())
.await
}));
}
barrier.wait().await;
let mut accepted = Vec::new();
let mut rejected = 0;
for task in tasks {
match tokio::time::timeout(Duration::from_secs(30), task)
.await
.expect("concurrent part deadline")
.expect("part task")
{
Ok(part) => accepted.push(part),
Err(StorageError::EntityTooLarge(12, 10)) => rejected += 1,
Err(other) => panic!("unexpected concurrent admission result: {other:?}"),
}
}
assert_eq!(accepted.len(), 1, "only one six-byte part fits the ten-byte limit");
assert_eq!(rejected, 1);
let listed = set
.list_object_parts(bucket, object, &upload.upload_id, None, 100, &ObjectOptions::default())
.await
.expect("read final part listing");
assert_eq!(listed.parts.len(), 1);
assert_eq!(listed.parts[0].part_num, accepted[0].part_num);
let winner = accepted.remove(0);
let part_number = winner.part_num;
set.clone()
.complete_multipart_upload(
bucket,
object,
&upload.upload_id,
vec![CompletePart {
part_num: part_number,
etag: winner.etag,
..Default::default()
}],
&ObjectOptions::default(),
)
.await
.expect("complete the admitted part");
let mut reader = set
.get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
.await
.expect("read completed object");
let mut actual = Vec::new();
reader.stream.read_to_end(&mut actual).await.expect("read whole object");
assert_eq!(actual, vec![if part_number == 1 { 0x41 } else { 0x42 }; 6]);
}
#[tokio::test]
async fn collect_list_parts_results_fails_early_when_file_not_found_fallback_is_impossible() {
let started = Instant::now();
@@ -8155,9 +8607,7 @@ mod tests {
assert!(listed.parts.is_empty(), "part lock loss before rename must not publish the part");
}
#[tokio::test]
#[serial]
async fn put_object_part_fences_upload_lock_loss_before_rename() {
async fn put_object_part_fences_upload_lock_loss_before_rename_case(capped: bool) {
let target = Arc::new(std::sync::RwLock::new(None));
let refresh_calls = Arc::new(AtomicUsize::new(0));
let lockers: Vec<Arc<dyn LockClient>> = (0..4)
@@ -8169,10 +8619,12 @@ mod tests {
let bucket = "multipart-put-part-lock-loss-bucket";
let object = "object";
make_bucket_on_all(&disk_stores, bucket).await;
let mut create_opts = ObjectOptions::default();
if capped {
insert_str(&mut create_opts.user_defined, SUFFIX_MAX_TOTAL_OBJECT_SIZE, "8192".to_owned());
}
let upload = temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("false"))], async {
set_disks
.new_multipart_upload(bucket, object, &ObjectOptions::default())
.await
set_disks.new_multipart_upload(bucket, object, &create_opts).await
})
.await
.expect("multipart upload should be created");
@@ -8223,6 +8675,18 @@ mod tests {
assert!(listed.parts.is_empty(), "lock loss before rename must not publish the part");
}
#[tokio::test]
#[serial]
async fn put_object_part_fences_upload_lock_loss_before_rename() {
put_object_part_fences_upload_lock_loss_before_rename_case(false).await;
}
#[tokio::test]
#[serial]
async fn capped_multipart_part_fences_upload_lock_loss_before_rename() {
put_object_part_fences_upload_lock_loss_before_rename_case(true).await;
}
#[tokio::test]
async fn complete_encrypted_multipart_marks_quorum_validated_layout() {
temp_env::async_with_vars([(crate::object_api::ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, Some("true"))], async {
@@ -363,6 +363,202 @@ mod native_abort_closure_tests {
}
}
#[tokio::test]
async fn stale_cleanup_reclaims_certified_one_and_two_disk_residuals_after_reopen() {
for survivors in [1, 2] {
let (dirs, disks, set) = hermetic_set_disks(4).await;
let bucket = "native-stale-certified";
let object = format!("survivors-{survivors}");
make_bucket_on_all(&disks, bucket).await;
let opts = ObjectOptions {
max_parity: true,
..Default::default()
};
let (upload, _) = stage_upload_with_create_opts(&set, bucket, &object, &[0x61; 4096], &opts).await;
let path = SetDisks::get_upload_id_dir(bucket, &object, &upload);
let before = originals(&disks, &path).await;
{
let physical = disks.iter().cloned().map(Some).collect::<Vec<_>>();
let SourceCleanupAuthority::Native { certificate, snapshots } =
prepare_source_cleanup(&physical, bucket, &object, &path, true, &set.multipart_proof_verification_context())
.await
.expect("close every original and persist the certificate")
else {
panic!("native upload must retain a native certificate")
};
for slot in 0..(4 - survivors) {
let response = operation(
&disks[slot],
&path,
MultipartSourceOperation::DeleteAborted {
expected: snapshots[slot].clone().expect("actual certified source"),
certificate: certificate.as_ref().clone(),
},
)
.await;
assert!(matches!(response, MultipartSourceResponse::AbortDeleted(_)));
}
}
let (reopened, cold) = cold_reopen(disks, set).await;
for slot in (4 - survivors)..4 {
assert_eq!(
retained_native_multipart_original(&reopened[slot], slot, bucket, &object, &path).await,
before[slot]
);
}
cold.lock_stale_multipart_cleanup(&path)
.await
.expect("complete certificate authorizes the background retry")
.delete(&cold)
.await
.expect("background retry removes certified survivors");
for dir in &dirs {
assert!(!dir.path().join(RUSTFS_META_MULTIPART_BUCKET).join(&path).exists());
}
assert!(
cold.lock_stale_multipart_cleanup(&path).await.is_err(),
"a repeated pass must not recreate the retired upload"
);
}
}
#[tokio::test]
#[serial]
async fn stale_certified_residual_survives_lock_loss_then_converges() {
let target = Arc::new(std::sync::RwLock::new(None));
let refresh_calls = Arc::new(AtomicUsize::new(0));
let lockers: Vec<Arc<dyn LockClient>> = (0..4)
.map(|_| {
Arc::new(SelectiveLockLossClient::new(Arc::clone(&target), Arc::clone(&refresh_calls))) as Arc<dyn LockClient>
})
.collect();
let (dirs, disks, set) = hermetic_set_disks_with_lockers(4, 0, 2, lockers).await;
let bucket = "stale-certified-lock-loss";
let object = "object";
make_bucket_on_all(&disks, bucket).await;
let opts = ObjectOptions {
max_parity: true,
..Default::default()
};
let (upload, _) = stage_upload_with_create_opts(&set, bucket, object, &[0x63; 4096], &opts).await;
let path = SetDisks::get_upload_id_dir(bucket, object, &upload);
let (fi, _) = set
.check_upload_id_exists(bucket, object, &upload, false)
.await
.expect("source metadata before closure");
let part_path = format!("{path}/{}/part.1", fi.data_dir.expect("source data directory"));
let original = originals(&disks, &path).await;
let surviving_part = disks[3]
.read_all(RUSTFS_META_MULTIPART_BUCKET, &part_path)
.await
.expect("snapshot surviving shard");
{
let physical = disks.iter().cloned().map(Some).collect::<Vec<_>>();
let SourceCleanupAuthority::Native { certificate, snapshots } =
prepare_source_cleanup(&physical, bucket, object, &path, true, &set.multipart_proof_verification_context())
.await
.expect("all original participants close and persist the certificate")
else {
panic!("native certificate required")
};
let authority = set.multipart_proof_verification_context();
let mut ids = certificate.disk_ids.iter().copied().map(Some).collect::<Vec<_>>();
ids[0] = None;
assert_eq!(
crate::set_disk::core::multipart_source::abort_certificate_cohort(&certificate, &ids, &authority),
Err(DiskError::ErasureWriteQuorum),
"an unavailable participant is a retryable proof gap"
);
ids[1] = Some(Uuid::new_v4());
assert_eq!(
crate::set_disk::core::multipart_source::abort_certificate_cohort(&certificate, &ids, &authority),
Err(DiskError::FileCorrupt),
"a missing participant cannot hide a known identity conflict"
);
ids[0] = Some(Uuid::new_v4());
ids[1] = Some(certificate.disk_ids[1]);
assert_eq!(
crate::set_disk::core::multipart_source::abort_certificate_cohort(&certificate, &ids, &authority),
Err(DiskError::FileCorrupt),
"a known different identity remains a certificate conflict"
);
for slot in 0..3 {
let response = operation(
&disks[slot],
&path,
MultipartSourceOperation::DeleteAborted {
expected: snapshots[slot].clone().expect("certified source"),
certificate: certificate.as_ref().clone(),
},
)
.await;
assert!(matches!(response, MultipartSourceResponse::AbortDeleted(_)));
}
}
assert!(dirs[3].path().join(RUSTFS_META_MULTIPART_BUCKET).join(&path).exists());
let offline = set.disks.write().await[0]
.take()
.expect("take an original participant offline");
let rejected = set.lock_stale_multipart_cleanup(&path).await;
set.disks.write().await[0] = Some(offline);
let err = rejected
.err()
.expect("a certificate must not substitute for an unavailable participant");
assert!(
matches!(err, StorageError::ErasureWriteQuorum | StorageError::InsufficientWriteQuorum(..)),
"the missing participant must be reported as a quorum failure: {err:?}"
);
assert_eq!(retained_native_multipart_original(&disks[3], 3, bucket, object, &path).await, original[3]);
assert_eq!(
disks[3]
.read_all(RUSTFS_META_MULTIPART_BUCKET, &part_path)
.await
.expect("surviving shard while a participant was offline"),
surviving_part
);
tokio::time::pause();
*target.write().expect("lock-loss target") =
Some(rustfs_lock::ObjectKey::new(RUSTFS_META_MULTIPART_BUCKET, path.clone()));
let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await;
let guard = set
.lock_stale_multipart_cleanup(&path)
.await
.expect("certificate admits the one-disk residual");
tokio::time::advance(Duration::from_secs(11)).await;
for _ in 0..20 {
if guard.is_lock_lost() {
break;
}
tokio::task::yield_now().await;
}
assert!(refresh_calls.load(Ordering::Acquire) > 0, "the real lock refresh must fail");
assert!(guard.is_lock_lost(), "the cleanup guard must observe lock loss");
// Real disk I/O must not run under Tokio's auto-advancing paused clock:
// a pending blocking operation could otherwise exhaust its timeout.
tokio::time::resume();
let err = guard.delete(&set).await.expect_err("lock loss must fence physical deletion");
assert!(matches!(err, StorageError::NamespaceLockQuorumUnavailable { .. }), "{err:?}");
assert_eq!(retained_native_multipart_original(&disks[3], 3, bucket, object, &path).await, original[3]);
assert_eq!(
disks[3]
.read_all(RUSTFS_META_MULTIPART_BUCKET, &part_path)
.await
.expect("surviving shard after loss"),
surviving_part
);
*target.write().expect("restore lock service") = None;
set.lock_stale_multipart_cleanup(&path)
.await
.expect("reacquire the cleanup lock")
.delete(&set)
.await
.expect("the certified residual converges after lock recovery");
for dir in &dirs {
assert!(!dir.path().join(RUSTFS_META_MULTIPART_BUCKET).join(&path).exists());
}
}
#[tokio::test]
async fn native_abort_partial_delete_surviving_producer_confirms_current_and_historical_exact_target() {
use crate::set_disk::core::io_primitives::{rename_fault_injection, rollback_fault_injection};
+6 -7
View File
@@ -11958,13 +11958,12 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
let bucket_lifecycle_guard = if let Some(expected_incarnation_id) = opts.expected_bucket_incarnation_id
&& opts.bucket_lifecycle_lock_fence.is_none()
{
let guard = metadata_sys::object_store_in(&self.ctx)
.await?
.acquire_bucket_lifecycle_read_lock(bucket)
.await?;
self.validate_bucket_incarnation(bucket, expected_incarnation_id, opts)
.await?;
Some(guard)
Some(
metadata_sys::object_store_in(&self.ctx)
.await?
.acquire_bucket_incarnation_fence_with_options(bucket, expected_incarnation_id, Some(opts))
.await?,
)
} else {
None
};
+81 -3
View File
@@ -448,6 +448,14 @@ impl ECStore {
await_orphan_bucket_recovery(bucket, &publication_guard, &lifecycle_guard, &metadata_guard, &ns_guard, recover).await
}
/// Read quota and bucket identity from one authoritative metadata snapshot.
pub async fn bucket_quota_and_incarnation_from_disk(
&self,
bucket: &str,
) -> Result<(Option<crate::bucket::quota::BucketQuota>, Uuid, OffsetDateTime)> {
metadata_sys::get_quota_config_and_incarnation_from_disk_in(&self.ctx, bucket).await
}
pub async fn get_bucket_metadata(&self, bucket: &str) -> Result<Arc<BucketMetadata>> {
let sys = metadata_sys::require_bucket_metadata_sys_in(&self.ctx)?;
sys.read().await.get(bucket).await
@@ -564,13 +572,83 @@ impl ECStore {
.await
}
pub(super) async fn acquire_bucket_incarnation_fence_with_options(
pub(crate) async fn acquire_bucket_incarnation_fence_with_options(
&self,
bucket: &str,
expected: Uuid,
opts: Option<&ObjectOptions>,
) -> Result<super::BucketIncarnationFenceGuard> {
Ok(self
.acquire_bucket_incarnation_fence_for_options(bucket, Some(expected), opts)
.await?
.0)
}
pub(crate) async fn acquire_bucket_incarnation_fence_for_options(
&self,
bucket: &str,
expected: Option<Uuid>,
opts: Option<&ObjectOptions>,
) -> Result<(super::BucketIncarnationFenceGuard, Uuid)> {
let inner = self.acquire_bucket_lifecycle_read_lock(bucket).await?;
self.validate_bucket_incarnation_fence_guard_with_options(bucket, inner, expected, opts)
.await
}
/// Acquire lexicographically ordered bucket lifecycle read locks for a copy
/// operation and validate both bucket incarnations through [`super::BucketFenceRegistry`].
pub async fn acquire_copy_bucket_incarnation_fences(
&self,
source_bucket: &str,
expected_source: Option<Uuid>,
destination_bucket: &str,
expected_destination: Option<Uuid>,
) -> Result<(super::BucketIncarnationFenceGuard, Uuid, Option<super::BucketIncarnationFenceGuard>, Uuid)> {
let (source_guard, destination_guard) = if source_bucket == destination_bucket {
(self.acquire_bucket_lifecycle_read_lock(source_bucket).await?, None)
} else if source_bucket < destination_bucket {
let source_guard = self.acquire_bucket_lifecycle_read_lock(source_bucket).await?;
let destination_guard = self.acquire_bucket_lifecycle_read_lock(destination_bucket).await?;
(source_guard, Some(destination_guard))
} else {
let destination_guard = self.acquire_bucket_lifecycle_read_lock(destination_bucket).await?;
let source_guard = self.acquire_bucket_lifecycle_read_lock(source_bucket).await?;
(source_guard, Some(destination_guard))
};
let (source_guard, current_source_incarnation_id) = self
.validate_bucket_incarnation_fence_guard_with_options(source_bucket, source_guard, expected_source, None)
.await?;
let (destination_guard, current_destination_incarnation_id) = if let Some(destination_guard) = destination_guard {
let (destination_guard, destination_incarnation_id) = self
.validate_bucket_incarnation_fence_guard_with_options(
destination_bucket,
destination_guard,
expected_destination,
None,
)
.await?;
(Some(destination_guard), destination_incarnation_id)
} else {
if expected_destination.is_some_and(|expected| expected != current_source_incarnation_id) {
return Err(StorageError::BucketNotFound(destination_bucket.to_owned()));
}
(None, current_source_incarnation_id)
};
Ok((
source_guard,
current_source_incarnation_id,
destination_guard,
current_destination_incarnation_id,
))
}
async fn validate_bucket_incarnation_fence_guard_with_options(
&self,
bucket: &str,
inner: rustfs_lock::NamespaceLockGuard,
expected: Option<Uuid>,
opts: Option<&ObjectOptions>,
) -> Result<(super::BucketIncarnationFenceGuard, Uuid)> {
let mut guard = super::BucketIncarnationFenceGuard::new(
Arc::clone(&self.bucket_fence_registry),
bucket,
@@ -586,10 +664,10 @@ impl ECStore {
}
})
.await?;
if current != expected {
if expected.is_some_and(|expected| current != expected) {
return Err(StorageError::BucketNotFound(bucket.to_owned()));
}
Ok(guard)
Ok((guard, current))
}
/// Hold this guard through recursive-delete authorization and mutation so
+22 -1
View File
@@ -149,6 +149,18 @@ impl BucketFenceRegistry {
}
}
}
/// Return the memoized incarnation id for `bucket` when at least one
/// registered lifecycle read guard is currently held without lock loss.
pub(crate) fn validated_incarnation(&self, bucket: &str) -> Option<Uuid> {
let mut entries = self.entries.lock().expect("bucket fence registry poisoned");
let entry = entries.get_mut(bucket)?;
if entry.guards.iter().any(|guard| guard.loss_probe.is_lost()) {
entry.validated = None;
return None;
}
entry.validated
}
}
/// Owns continuous lifecycle coverage and its incarnation validation memo.
@@ -221,7 +233,7 @@ impl BucketIncarnationFenceGuard {
}
}
pub(crate) fn is_lock_lost(&self) -> bool {
pub fn is_lock_lost(&self) -> bool {
self.inner.as_ref().is_some_and(NamespaceLockGuard::is_lock_lost)
}
@@ -310,9 +322,16 @@ mod tests {
let (first_probe, first_lost) = controllable_probe();
let first = reg.enter("b", first_probe);
assert_eq!(first.memoized, None);
assert_eq!(reg.validated_incarnation("b"), None);
reg.memoize("b", uuid(7));
assert_eq!(reg.validated_incarnation("b"), Some(uuid(7)));
first_lost.store(true, std::sync::atomic::Ordering::SeqCst);
assert_eq!(
reg.validated_incarnation("b"),
None,
"live lost guard must clear and block validated_incarnation"
);
let second = reg.enter("b", live_probe());
assert_eq!(second.memoized, None, "live lost guard must force disk revalidation");
@@ -320,9 +339,11 @@ mod tests {
let third = reg.enter("b", live_probe());
assert_eq!(third.memoized, None, "memo remains blocked while the lost guard is live");
assert_eq!(reg.validated_incarnation("b"), None);
reg.exit("b", first.token, true);
reg.memoize("b", uuid(8));
assert_eq!(reg.validated_incarnation("b"), Some(uuid(8)));
let fourth = reg.enter("b", live_probe());
assert_eq!(fourth.memoized, Some(uuid(8)), "memo resumes after lost coverage leaves");
+150
View File
@@ -21384,6 +21384,141 @@ mod tests {
shutdown.cancel();
}
#[tokio::test]
#[serial_test::serial(storage_class_env)]
async fn put_object_metadata_and_copy_object_reuse_overlapping_incarnation_fence() {
let temp = tempfile::tempdir().expect("create metadata/copy fence store");
let (_ctx, store, shutdown) =
without_storage_class_env(build_isolated_test_store(temp.path(), "metadata-copy-fence-reuse", &[4])).await;
crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
let bucket = format!("metadata-copy-fence-{}", Uuid::new_v4());
store
.make_bucket(&bucket, &MakeBucketOptions::default())
.await
.expect("create bucket");
let mut seed_reader = PutObjReader::from_vec(b"seed payload".to_vec());
let versioned_opts = ObjectOptions {
versioned: true,
..Default::default()
};
let mut src_info = store
.put_object(&bucket, "src.bin", &mut seed_reader, &versioned_opts)
.await
.expect("seed source object");
let src_version_id = src_info
.version_id
.expect("versioned PUT should return a version ID")
.to_string();
let expected = store.bucket_incarnation_id(&bucket).await.expect("read incarnation");
let coverage = store
.acquire_bucket_incarnation_fence(&bucket, expected)
.await
.expect("validate continuous coverage");
let metadata_writer = crate::bucket::metadata_sys::acquire_bucket_metadata_transaction_lock_in(&store.ctx, &bucket)
.await
.expect("hold metadata writer during covered metadata and copy operations");
let updated = tokio::time::timeout(
Duration::from_secs(10),
store.put_object_metadata(
&bucket,
"src.bin",
&ObjectOptions {
version_id: Some(src_version_id.clone()),
versioned: true,
eval_metadata: Some(HashMap::from([("x-amz-meta-fenced".to_string(), "true".to_string())])),
..Default::default()
},
),
)
.await
.expect("covered put_object_metadata must not reacquire the metadata transaction lock")
.expect("update object metadata using authoritative memo");
assert_eq!(updated.user_defined.get("x-amz-meta-fenced").map(String::as_str), Some("true"));
Arc::make_mut(&mut src_info.user_defined).insert("x-amz-meta-copied".to_string(), "true".to_string());
let src_copy_opts = ObjectOptions {
version_id: Some(src_version_id),
versioned: true,
..Default::default()
};
let copied = tokio::time::timeout(
Duration::from_secs(10),
store.copy_object(&bucket, "src.bin", &bucket, "src.bin", &mut src_info, &src_copy_opts, &versioned_opts),
)
.await
.expect("covered versioned copy_object must not reacquire the metadata transaction lock")
.expect("versioned copy using authoritative memo");
assert_eq!(copied.user_defined.get("x-amz-meta-copied").map(String::as_str), Some("true"));
let (src_guard, src_incarnation, dst_guard, dst_incarnation) = tokio::time::timeout(
Duration::from_secs(10),
store.acquire_copy_bucket_incarnation_fences(&bucket, Some(expected), &bucket, Some(expected)),
)
.await
.expect("covered acquire_copy_bucket_incarnation_fences must not reacquire the metadata transaction lock")
.expect("validate same-bucket copy fences using authoritative memo");
assert!(dst_guard.is_none());
assert_eq!(src_incarnation, expected);
assert_eq!(dst_incarnation, expected);
let mut borrowed_fence_opts = ObjectOptions {
expected_bucket_incarnation_id: Some(expected),
..Default::default()
};
src_guard.attach_to_object_options(&mut borrowed_fence_opts);
let fenced_incarnation = tokio::time::timeout(
Duration::from_secs(10),
crate::bucket::metadata_sys::get_bucket_incarnation_id_for_options_in(&store.ctx, &bucket, &borrowed_fence_opts),
)
.await
.expect("borrowed fence under active coverage must not reacquire the metadata transaction lock")
.expect("resolve incarnation under borrowed fence");
assert_eq!(fenced_incarnation, expected);
assert!(
matches!(
store
.acquire_copy_bucket_incarnation_fences(&bucket, Some(expected), &bucket, Some(Uuid::new_v4()))
.await,
Err(StorageError::BucketNotFound(_))
),
"same-bucket copy fence must reject a mismatched destination incarnation"
);
drop(src_guard);
let stale_opts = ObjectOptions {
expected_bucket_incarnation_id: Some(Uuid::new_v4()),
versioned: true,
..Default::default()
};
assert!(
matches!(
store.put_object_metadata(&bucket, "src.bin", &stale_opts).await,
Err(StorageError::BucketNotFound(_))
),
"put_object_metadata must reject a stale expected incarnation even under warm coverage"
);
assert!(
matches!(
store
.copy_object(
&bucket,
"src.bin",
&bucket,
"src.bin",
&mut src_info,
&ObjectOptions::default(),
&stale_opts,
)
.await,
Err(StorageError::BucketNotFound(_))
),
"copy_object must reject a stale expected incarnation even under warm coverage"
);
drop(metadata_writer);
drop(coverage);
shutdown.cancel();
}
#[tokio::test]
#[serial_test::serial(storage_class_env)]
async fn multipart_snapshot_bypasses_initializer_queued_behind_metadata_writer() {
@@ -21469,10 +21604,25 @@ mod tests {
.acquire_bucket_incarnation_fence(&bucket, old)
.await
.expect("validate with no-op lock");
let snapshot = store
.object_lock_config_snapshot_under_lifecycle_fence(&bucket, &crate::object_api::NamespaceLockFence::new())
.await
.expect("capture snapshot under no-op lock");
let current = Uuid::new_v4();
crate::bucket::metadata::save_bucket_incarnation(store.clone(), &bucket, current)
.await
.expect("replace authority while no-op guard provides no exclusion");
let snapshot_opts = ObjectOptions {
expected_bucket_incarnation_id: Some(old),
object_lock_config_snapshot: Some(snapshot),
..Default::default()
};
assert_eq!(
crate::bucket::metadata_sys::get_bucket_incarnation_id_for_options_in(&store.ctx, &bucket, &snapshot_opts)
.await
.expect("disabled lock manager must re-read incarnation from disk instead of trusting snapshot"),
current
);
assert!(
matches!(
store.acquire_bucket_incarnation_fence(&bucket, old).await,
+9 -6
View File
@@ -1030,16 +1030,19 @@ impl ECStore {
// Lock order matches writes: bucket lifecycle, metadata validation,
// then object namespace. Borrow fences supplied by an outer operation.
let bucket_guard = if opts.bucket_lifecycle_lock_fence.is_none() {
let guard = self.acquire_bucket_lifecycle_read_lock(bucket).await?;
opts.add_bucket_lifecycle_lock_guard(&guard);
let guard = self
.acquire_bucket_incarnation_fence_with_options(bucket, bucket_incarnation_id, Some(&opts))
.await?;
guard.attach_to_object_options(&mut opts);
Some(guard)
} else {
let incarnation =
crate::bucket::metadata_sys::get_bucket_incarnation_id_for_options_in(&self.ctx, bucket, &opts).await?;
if incarnation != bucket_incarnation_id {
return Err(Error::BucketNotFound(bucket.to_owned()));
}
None
};
let incarnation = crate::bucket::metadata_sys::get_bucket_incarnation_id_for_options_in(&self.ctx, bucket, &opts).await?;
if incarnation != bucket_incarnation_id {
return Err(Error::BucketNotFound(bucket.to_owned()));
}
let metadata_sys = crate::bucket::metadata_sys::bucket_metadata_sys_of(&self.ctx)?;
let metadata_sys = metadata_sys.read().await.clone();
let (versioning, _) = metadata_sys.get_versioning_config(bucket).await?;
+1 -1
View File
@@ -412,7 +412,7 @@ impl ECStore {
/// Multipart lock order is bucket lifecycle, generation validation, then
/// object/upload locks in the selected set.
async fn guard_multipart_bucket_incarnation(
pub(super) async fn guard_multipart_bucket_incarnation(
&self,
bucket: &str,
opts: &ObjectOptions,
+22 -47
View File
@@ -30,8 +30,8 @@ use crate::bucket::lifecycle::{
},
};
use crate::bucket::metadata_sys::{
acquire_bucket_metadata_transaction_read_lock_in, get_bucket_incarnation_id_for_options_in,
get_cached_bucket_incarnation_id_in, get_delete_config_from_disk_in, get_object_lock_config_and_incarnation_from_disk_in,
acquire_bucket_metadata_transaction_read_lock_in, get_cached_bucket_incarnation_id_in, get_delete_config_from_disk_in,
get_object_lock_config_and_incarnation_from_disk_in,
};
use crate::bucket::object_lock::objectlock_sys::{
check_object_lock_for_deletion_with_state, ensure_recursive_force_delete_allowed_for_state,
@@ -4883,24 +4883,20 @@ impl ECStore {
if !is_meta_bucketname(dst_bucket) && dst_opts.expected_bucket_incarnation_id.is_none() {
dst_opts.expected_bucket_incarnation_id = Some(self.bucket_incarnation_id(dst_bucket).await?);
}
let _bucket_lifecycle_guard = if is_meta_bucketname(dst_bucket) || dst_opts.bucket_lifecycle_lock_fence.is_some() {
None
} else {
Some(self.acquire_bucket_lifecycle_read_lock(dst_bucket).await?)
};
let current_bucket_incarnation_id = if let Some(guard) = _bucket_lifecycle_guard.as_ref() {
dst_opts.add_bucket_lifecycle_lock_guard(guard);
let current_incarnation_id = get_bucket_incarnation_id_for_options_in(&self.ctx, dst_bucket, &dst_opts).await?;
if dst_opts
.expected_bucket_incarnation_id
.is_some_and(|expected| expected != current_incarnation_id)
{
return Err(StorageError::BucketNotFound(dst_bucket.to_string()));
}
Some(current_incarnation_id)
} else {
dst_opts.expected_bucket_incarnation_id
};
let (_bucket_lifecycle_guard, current_bucket_incarnation_id) =
if is_meta_bucketname(dst_bucket) || dst_opts.bucket_lifecycle_lock_fence.is_some() {
(None, dst_opts.expected_bucket_incarnation_id)
} else {
let (guard, current_incarnation_id) = self
.acquire_bucket_incarnation_fence_for_options(
dst_bucket,
dst_opts.expected_bucket_incarnation_id,
Some(&dst_opts),
)
.await?;
guard.attach_to_object_options(&mut dst_opts);
(Some(guard), Some(current_incarnation_id))
};
if dst_opts
.bucket_lifecycle_lock_fence
.as_ref()
@@ -6504,24 +6500,8 @@ impl ECStore {
opts: &ObjectOptions,
) -> Result<()> {
let object = encode_dir_object(object);
let mut opts = transition_restore_pool_opts(opts);
if !is_meta_bucketname(bucket) && opts.expected_bucket_incarnation_id.is_none() {
opts.expected_bucket_incarnation_id = Some(self.bucket_incarnation_id(bucket).await?);
}
let bucket_lifecycle_guard = if is_meta_bucketname(bucket) {
None
} else {
Some(self.acquire_bucket_lifecycle_read_lock(bucket).await?)
};
if let Some(guard) = bucket_lifecycle_guard.as_ref() {
opts.add_bucket_lifecycle_lock_guard(guard);
}
if !is_meta_bucketname(bucket) {
let current_incarnation_id = get_bucket_incarnation_id_for_options_in(&self.ctx, bucket, &opts).await?;
if opts.expected_bucket_incarnation_id != Some(current_incarnation_id) {
return Err(StorageError::BucketNotFound(bucket.to_string()));
}
}
let opts = transition_restore_pool_opts(opts);
let (mut opts, _bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, &opts).await?;
if opts.overwrites_existing_version() && !is_meta_bucketname(bucket) {
let expected_incarnation_id = opts
.expected_bucket_incarnation_id
@@ -6589,16 +6569,11 @@ impl ECStore {
let bucket_lifecycle_guard = if is_meta_bucketname(bucket) {
None
} else {
let guard = self.acquire_bucket_lifecycle_read_lock(bucket).await?;
let current_incarnation_id = get_bucket_incarnation_id_for_options_in(&self.ctx, bucket, &opts).await?;
if opts
.expected_bucket_incarnation_id
.is_some_and(|expected| expected != current_incarnation_id)
{
return Err(StorageError::BucketNotFound(bucket.to_string()));
}
let (guard, current_incarnation_id) = self
.acquire_bucket_incarnation_fence_for_options(bucket, opts.expected_bucket_incarnation_id, Some(&opts))
.await?;
opts.expected_bucket_incarnation_id = Some(current_incarnation_id);
opts.add_bucket_lifecycle_lock_guard(&guard);
guard.attach_to_object_options(&mut opts);
if guard.is_lock_lost() {
return Err(Error::other("bucket lifecycle lock was lost before the metadata update"));
}
+214 -13
View File
@@ -42,13 +42,14 @@ enum NotifyConfigStoreError {
Read(String),
Save(String),
Converge { persisted: bool, error: String },
Conflict,
}
fn notification_convergence_error(persisted: bool, error: impl std::fmt::Display) -> NotificationError {
let durable_state = if persisted { "persisted" } else { "unchanged" };
NotificationError::Configuration(format!(
"configuration was {durable_state} but notification runtime convergence failed: {error}"
))
NotificationError::ConfigurationConvergence {
persisted,
detail: error.to_string(),
}
}
async fn supervise_config_update<T>(
@@ -63,12 +64,62 @@ where
})?
}
fn configs_equal(first: Option<&KVS>, second: Option<&KVS>) -> bool {
match (first, second) {
(None, None) => true,
(Some(first), Some(second)) => {
if first == second {
return true;
}
let mut first: Vec<_> = first.0.iter().collect();
let mut second: Vec<_> = second.0.iter().collect();
// Lookup uses the first occurrence of a key, so preserve duplicate-key order.
first.sort_by(|a, b| a.key.cmp(&b.key));
second.sort_by(|a, b| a.key.cmp(&b.key));
first == second
}
_ => false,
}
}
fn replace_target_if_unchanged(
config: &mut Config,
target_type: &str,
target_name: &str,
expected: Option<&KVS>,
expected_default: Option<&KVS>,
kvs: &KVS,
) -> Result<bool, NotifyConfigStoreError> {
let targets = config.0.get(target_type);
let mut matching = targets
.into_iter()
.flat_map(|targets| targets.iter())
.filter(|(name, _)| name.eq_ignore_ascii_case(target_name));
let current = matching.next();
if matching.next().is_some() {
return Err(NotifyConfigStoreError::Conflict);
}
let stored_name = current
.map(|(name, _)| name.clone())
.unwrap_or_else(|| target_name.to_owned());
let current_default = targets.and_then(|targets| targets.get(rustfs_config::DEFAULT_DELIMITER));
if !configs_equal(current.map(|(_, kvs)| kvs), expected) || !configs_equal(current_default, expected_default) {
return Err(NotifyConfigStoreError::Conflict);
}
config
.0
.entry(target_type.to_owned())
.or_default()
.insert(stored_name, kvs.clone());
Ok(true)
}
async fn update_server_config<F>(
mut modifier: F,
lifecycle: NotifyLifecycleCoordinator,
) -> Result<Option<(crate::lifecycle::NotificationLifecycleTransition, bool)>, NotifyConfigStoreError>
where
F: FnMut(&mut Config) -> bool + Send + 'static,
F: FnMut(&mut Config, Option<String>) -> Result<bool, NotifyConfigStoreError> + Send + 'static,
{
let Some(store) = resolve_notify_object_store_handle() else {
return Err(NotifyConfigStoreError::StorageNotAvailable);
@@ -79,7 +130,8 @@ where
.await
.map_err(NotifyConfigStoreError::Read)?;
let mut config = snapshot.config.clone();
if !modifier(&mut config) {
let generation = snapshot.generation().map(|value| value.to_string());
if !modifier(&mut config, generation)? {
return Ok(None);
}
@@ -245,6 +297,30 @@ impl NotifyConfigManager {
.await
}
/// Replace a target only when its persisted configuration is unchanged since validation.
pub async fn set_target_config_if_unchanged(
&self,
target_type: &str,
target_name: &str,
expected: Option<KVS>,
expected_default: Option<KVS>,
expected_generation: Option<String>,
kvs: KVS,
) -> Result<(), NotificationError> {
let ttype = target_type.to_lowercase();
let tname = target_name.to_lowercase();
self.update_config_and_reload_checked(move |config, generation| {
if expected_generation
.as_ref()
.is_some_and(|expected| generation.as_ref() != Some(expected))
{
return Err(NotifyConfigStoreError::Conflict);
}
replace_target_if_unchanged(config, &ttype, &tname, expected.as_ref(), expected_default.as_ref(), &kvs)
})
.await
}
pub async fn remove_target_config(&self, target_type: &str, target_name: &str) -> Result<(), NotificationError> {
debug!(
event = EVENT_NOTIFY_CONFIG_UPDATE,
@@ -338,9 +414,17 @@ impl NotifyConfigManager {
Ok(())
}
async fn update_config_and_reload<F>(&self, modifier: F) -> Result<(), NotificationError>
async fn update_config_and_reload<F>(&self, mut modifier: F) -> Result<(), NotificationError>
where
F: FnMut(&mut Config) -> bool + Send + 'static,
{
self.update_config_and_reload_checked(move |config, _generation| Ok(modifier(config)))
.await
}
async fn update_config_and_reload_checked<F>(&self, modifier: F) -> Result<(), NotificationError>
where
F: FnMut(&mut Config, Option<String>) -> Result<bool, NotifyConfigStoreError> + Send + 'static,
{
if self.lifecycle.state() == NotificationRuntimeState::Terminated {
return Err(NotificationError::Initialization("Notification runtime has terminated".to_string()));
@@ -356,6 +440,7 @@ impl NotifyConfigManager {
NotifyConfigStoreError::Read(err) => NotificationError::ReadConfig(err),
NotifyConfigStoreError::Save(err) => NotificationError::SaveConfig(err),
NotifyConfigStoreError::Converge { persisted, error } => notification_convergence_error(persisted, error),
NotifyConfigStoreError::Conflict => NotificationError::ConfigurationConflict,
})?
else {
debug!(
@@ -394,6 +479,126 @@ mod tests {
use crate::rules::RulesMap;
use crate::{NotificationError, NotificationRuntimeState};
#[test]
fn config_comparison_preserves_duplicates_and_hidden_flags() {
use rustfs_config::server_config::{KV, KVS};
let field = |key: &str, value: &str, hidden_if_empty: bool| KV {
key: key.into(),
value: value.into(),
hidden_if_empty,
};
let original = KVS(vec![
field("password", "secret", false),
field("token", "", true),
field("password", "secret", false),
]);
let reordered = KVS(vec![
field("password", "secret", false),
field("password", "secret", false),
field("token", "", true),
]);
let missing_duplicate = KVS(vec![field("password", "secret", false), field("token", "", true)]);
let changed_flag = KVS(vec![
field("password", "secret", false),
field("password", "secret", false),
field("token", "", false),
]);
assert!(super::configs_equal(Some(&original), Some(&reordered)));
assert!(!super::configs_equal(Some(&original), Some(&missing_duplicate)));
assert!(!super::configs_equal(Some(&original), Some(&changed_flag)));
assert!(!super::configs_equal(Some(&original), None));
let first_value = KVS(vec![
field("password", "old", false),
field("token", "", true),
field("password", "new", false),
]);
let reordered_keys = KVS(vec![
field("token", "", true),
field("password", "old", false),
field("password", "new", false),
]);
let reordered_duplicates = KVS(vec![
field("token", "", true),
field("password", "new", false),
field("password", "old", false),
]);
assert!(super::configs_equal(Some(&first_value), Some(&reordered_keys)));
assert!(!super::configs_equal(Some(&first_value), Some(&reordered_duplicates)));
}
#[test]
fn conditional_target_update_rejects_rotation_and_deletion() {
use rustfs_config::server_config::{Config, KVS};
let mut old = KVS::new();
old.insert("password".into(), "old".into());
let mut rotated = KVS::new();
rotated.insert("password".into(), "new".into());
let mut config = Config::new();
config
.0
.entry("notify_nats".into())
.or_default()
.insert("primary".into(), rotated.clone());
let result = super::replace_target_if_unchanged(&mut config, "notify_nats", "primary", Some(&old), None, &old);
assert!(matches!(result, Err(super::NotifyConfigStoreError::Conflict)));
assert_eq!(config.0["notify_nats"]["primary"], rotated);
config.0.get_mut("notify_nats").expect("subsystem exists").remove("primary");
assert!(matches!(
super::replace_target_if_unchanged(&mut config, "notify_nats", "primary", Some(&old), None, &old),
Err(super::NotifyConfigStoreError::Conflict)
));
assert!(
super::replace_target_if_unchanged(&mut config, "notify_nats", "primary", None, None, &old)
.expect("create unchanged target")
);
assert!(
super::replace_target_if_unchanged(&mut config, "notify_nats", "primary", Some(&old), None, &rotated)
.expect("replace unchanged target")
);
assert_eq!(config.0["notify_nats"]["primary"], rotated);
let mut first = KVS::new();
first.insert("url".into(), "nats://example.com".into());
first.insert("token".into(), "secret".into());
let mut reordered = KVS::new();
reordered.insert("token".into(), "secret".into());
reordered.insert("url".into(), "nats://example.com".into());
config
.0
.get_mut("notify_nats")
.expect("subsystem")
.insert("primary".into(), first);
assert!(
super::replace_target_if_unchanged(&mut config, "notify_nats", "primary", Some(&reordered), None, &rotated)
.expect("key order does not signal a change")
);
let targets = config.0.get_mut("notify_nats").expect("subsystem");
targets.remove("primary");
targets.insert("Primary".into(), rotated.clone());
assert!(
super::replace_target_if_unchanged(&mut config, "notify_nats", "primary", Some(&rotated), None, &old)
.expect("case-insensitive match")
);
let targets = &config.0["notify_nats"];
assert!(targets.contains_key("Primary"));
assert!(!targets.contains_key("primary"));
let mut old_default = KVS::new();
old_default.insert("password".into(), "old-default".into());
let mut new_default = KVS::new();
new_default.insert("password".into(), "new-default".into());
config
.0
.get_mut("notify_nats")
.expect("subsystem")
.insert("_".into(), new_default.clone());
assert!(matches!(
super::replace_target_if_unchanged(&mut config, "notify_nats", "primary", Some(&old), Some(&old_default), &rotated),
Err(super::NotifyConfigStoreError::Conflict)
));
assert_eq!(config.0["notify_nats"]["Primary"], old);
assert_eq!(config.0["notify_nats"]["_"], new_default);
}
#[tokio::test]
async fn supervised_config_update_survives_waiter_cancellation() {
let (started_tx, started_rx) = tokio::sync::oneshot::channel();
@@ -550,13 +755,9 @@ mod tests {
#[test]
fn convergence_error_reports_durable_write_state() {
for (persisted, expected) in [(true, "configuration was persisted"), (false, "configuration was unchanged")] {
for persisted in [true, false] {
let error = notification_convergence_error(persisted, "injected failure");
let NotificationError::Configuration(message) = error else {
panic!("convergence error should be a configuration error");
};
assert!(message.contains(expected), "unexpected convergence error: {message}");
assert!(message.contains("injected failure"));
assert!(matches!(error, NotificationError::ConfigurationConvergence { persisted: value, .. } if value == persisted));
}
}
}
+6
View File
@@ -38,6 +38,12 @@ pub enum NotificationError {
#[error("Configuration error: {0}")]
Configuration(String),
#[error("Target configuration changed during validation")]
ConfigurationConflict,
#[error("Notification configuration convergence failed (persisted: {persisted}): {detail}")]
ConfigurationConvergence { persisted: bool, detail: String },
#[error("ARN not found: {0}")]
ARNNotFound(String),
+16
View File
@@ -298,6 +298,22 @@ impl NotificationSystem {
.await
}
/// Replace a target only if its persisted configuration still matches the validated snapshot.
pub async fn set_target_config_if_unchanged(
&self,
target_type: &str,
target_name: &str,
expected: Option<KVS>,
expected_default: Option<KVS>,
expected_generation: Option<String>,
kvs: KVS,
) -> Result<(), NotificationError> {
self.services
.config_manager
.set_target_config_if_unchanged(target_type, target_name, expected, expected_default, expected_generation, kvs)
.await
}
/// Removes all notification configurations for a bucket.
/// If the configuration is successfully removed, the entire notification system will be automatically reloaded.
///
+14 -1
View File
@@ -740,6 +740,15 @@ pub(super) fn completed_data_usage_info(
.map(|candidate| (candidate.data, candidate.last_update))
}
/// Certify an incarnation only when every contributing set reports the same non-nil identity.
fn certified_bucket_incarnation(results: &[DataUsageCache], bucket: &str) -> Option<uuid::Uuid> {
let mut identities = results
.iter()
.map(|result| result.info.scan_bucket_incarnations.get(bucket).copied());
let first = identities.next().flatten().filter(|id| !id.is_nil())?;
identities.all(|id| id == Some(first)).then_some(first)
}
pub(super) fn completed_usage_candidate(
results: &[DataUsageCache],
scope: &ScannerSnapshotScope<'_>,
@@ -792,7 +801,11 @@ pub(super) fn completed_usage_candidate(
let merged_last_update = validated.last_update;
let buckets_usage = bucket_entries
.into_iter()
.map(|(bucket, entry)| Ok((bucket, checked_bucket_usage_info(entry).ok_or(CandidateOutcome::UsageCounters)?)))
.map(|(bucket, entry)| {
let mut usage = checked_bucket_usage_info(entry).ok_or(CandidateOutcome::UsageCounters)?;
usage.bucket_incarnation = certified_bucket_incarnation(results, &bucket).map(|id| id.to_string());
Ok((bucket, usage))
})
.collect::<std::result::Result<HashMap<_, _>, CandidateOutcome>>()?;
let bucket_sizes = buckets_usage
.iter()
@@ -996,6 +996,73 @@ fn completed_data_usage_info_adds_same_bucket_across_unique_sets() {
assert_eq!(bucket.delete_markers_count, 1);
}
#[test]
fn bucket_usage_incarnation_is_optional_and_old_readers_ignore_it() {
#[derive(serde::Deserialize)]
struct LegacyBucketUsage {
size: u64,
}
let legacy = BucketUsageInfo {
size: 42,
..Default::default()
};
let without = serde_json::to_value(&legacy).expect("encode legacy usage");
assert!(without.get("bucket_incarnation").is_none());
let restored: BucketUsageInfo = serde_json::from_value(without).expect("decode old usage");
assert_eq!(restored.bucket_incarnation, None);
let certified = BucketUsageInfo {
size: 42,
bucket_incarnation: Some(uuid::Uuid::from_u128(7).to_string()),
..Default::default()
};
let encoded = serde_json::to_vec(&certified).expect("encode certified usage");
let old: LegacyBucketUsage = serde_json::from_slice(&encoded).expect("old schema ignores additive field");
assert_eq!(old.size, 42);
let named = rmp_serde::to_vec_named(&certified).expect("encode certified named msgpack");
let old: LegacyBucketUsage = rmp_serde::from_slice(&named).expect("old named reader ignores additive field");
assert_eq!(old.size, 42);
let old_bytes = rmp_serde::to_vec_named(&legacy).expect("encode old shape");
let new: BucketUsageInfo = rmp_serde::from_slice(&old_bytes).expect("new reader accepts old shape");
assert_eq!(new.bucket_incarnation, None);
}
#[test]
fn completed_usage_certifies_only_matching_non_nil_bucket_incarnations() {
let buckets = vec!["bucket".to_string()];
let id = uuid::Uuid::from_u128(7);
let mut a = completed_root_cache("bucket", 2, 10, DataUsageCacheSource::new(0, 0));
let mut b = completed_root_cache("bucket", 3, 20, DataUsageCacheSource::new(1, 0));
a.info.scan_bucket_incarnations.insert("bucket".to_string(), id);
b.info.scan_bucket_incarnations.insert("bucket".to_string(), id);
let published = completed_data_usage_info_for_test(&[a.clone(), b.clone()], &buckets, false, false)
.unwrap()
.0;
assert_eq!(
published.buckets_usage["bucket"].bucket_incarnation.as_deref(),
Some(id.to_string().as_str())
);
let encoded = serde_json::to_vec(&published).expect("serialize certified usage");
let decoded: DataUsageInfo = serde_json::from_slice(&encoded).expect("read certified usage");
assert_eq!(decoded.buckets_usage["bucket"].bucket_incarnation, Some(id.to_string()));
b.info
.scan_bucket_incarnations
.insert("bucket".to_string(), uuid::Uuid::from_u128(8));
let mismatch = completed_data_usage_info_for_test(&[a.clone(), b.clone()], &buckets, false, false)
.unwrap()
.0;
assert_eq!(mismatch.buckets_usage["bucket"].bucket_incarnation, None);
b.info.scan_bucket_incarnations.remove("bucket");
let missing = completed_data_usage_info_for_test(&[a.clone(), b.clone()], &buckets, false, false)
.unwrap()
.0;
assert_eq!(missing.buckets_usage["bucket"].bucket_incarnation, None);
b.info
.scan_bucket_incarnations
.insert("bucket".to_string(), uuid::Uuid::nil());
let nil = completed_data_usage_info_for_test(&[a, b], &buckets, false, false).unwrap().0;
assert_eq!(nil.buckets_usage["bucket"].bucket_incarnation, None);
}
#[test]
fn checked_bucket_usage_info_moves_replication_keys_and_preserves_the_snapshot() {
let first_target = "arn:rustfs:replication:local::target-a".to_string();
@@ -1057,6 +1124,7 @@ fn checked_bucket_usage_info_moves_replication_keys_and_preserves_the_snapshot()
}
let expected = BucketUsageInfo {
size: 32768,
bucket_incarnation: None,
replication_pending_size_v1: 0,
replication_failed_size_v1: 0,
replicated_size_v1: 0,
+1 -1
View File
@@ -100,7 +100,7 @@ Standard S3 areas that must not be described as complete:
| Bucket access logging | Handlers exist (`get_bucket_logging`, `put_bucket_logging` in `rustfs/src/storage/ecfs.rs`); the `test_*bucket_logging*` s3tests cases are still listed as unimplemented | `unimplemented_tests.txt` |
| POST Object form upload checksum handling | Not yet passing | `unimplemented_tests.txt` |
| Bucket ownership controls | No handler | `unimplemented_tests.txt` |
| Multipart upload listing and part lookup compatibility edge cases | Not part of default gate | `excluded_tests.txt` |
| Multipart upload listing and part lookup compatibility edge cases | Not part of default gate; ListParts returns HTTP 503 `SlowDownRead` when no directory was listed and any participant is unavailable, instead of treating one missing directory as an empty upload | `excluded_tests.txt`; `collect_list_parts_results_rejects_unavailable_empty_upload_dirs` |
| IAM-account or multi-storage-class dependent cases | Not part of default gate | `unimplemented_tests.txt` |
| Tenanted bucket policy edge cases | Needs investigation | `unimplemented_tests.txt` |
@@ -182,6 +182,18 @@ requires a compatibility window and a dedicated cleanup entry.
| `.scanner-cycle.lock` | scanner runtime lock | Serializes cycle-level work. A missing lock object is not itself usage evidence. |
| `.scanner-pause-backlog.json` | scanner pause and catch-up ledger | Tracks dirty usage, discovered lifecycle work, and full-scan catch-up while authoritative publication is fenced by data movement. It never grants publication admission. |
The authoritative snapshot may include an optional `bucket_incarnation` in each
bucket usage entry. The coordinator writes it only when every contributing set
reports the same non-nil bucket UUID. Missing or conflicting identities leave
that entry uncertified; they do not reject the usage publication or change quota
admission. Veeam SOSAPI capacity reads require this proof to match the bucket
incarnation read from authoritative storage. An old snapshot, a mixed-version
writer that omits the field, a pending target bucket, or an unavailable identity
therefore yields a retryable capacity error until a compatible scanner publishes
a certified entry. The optional field is omitted when unavailable so older
snapshot readers continue to see the original shape. This proof identifies the
bucket generation; it does not make scanner usage a real-time quota ledger.
## Observed Snapshots
Observed snapshots are a diagnostic and availability layer. They may be served
+2 -1
View File
@@ -54,4 +54,5 @@ Properties of a capped upload:
- Unknown-length or negative-length parts are rejected with `UnexpectedContent` rather than buffered without a bound.
- Capped parts are admitted under an upload-wide write lock before temporary shards are created, and hold a per-upload staging permit that bounds local in-flight data. The lock is released while the body is read and reacquired for the final check and rename, so `Complete` and `Abort` are not blocked behind a slow upload. The request-body stall timeout releases the staging permit when a client stops sending.
- An upload created without the parameter stays unlimited.
- Enforcement runs in the multipart data plane on every node. During a rolling upgrade, route capped uploads only to nodes that carry the V2 implementation; a node without it treats the internal metadata as unknown and cannot enforce the limit.
- Size admission requires a response from every configured disk, refuses incomplete or sub-quorum part observations, and requires matching part metadata at write quorum. An unknown logical size, or one differing from its stored part size without a recorded transformation, is refused. A missing part directory on a responding disk is an empty observation; an unavailable disk is not. Capped UploadPart may therefore return a retryable storage error during disk outages even when the write quorum is available. Retry after the disks recover.
- Enforcement runs in the multipart data plane on every node. During a rolling upgrade, route capped uploads only to nodes that carry both V2 and the strict part-size proof; older nodes can treat an unavailable part observation as empty. A node without V2 does not enforce the limit at all. Fleet-wide protection requires upgrading every node that may receive these writes.
+5 -1
View File
@@ -729,7 +729,11 @@ Only two MinIO scanner variables are recognized. `apply_external_env_compat` (`c
| `scanner alert_excess_folders` (50000) | `scanner.alert_excess_folders` (65538) | Not mapped; see [Scanner Excess Alerts](scanner-excess-alerts.md). |
| Any other `MINIO_SCANNER_*` | Corresponding `RUSTFS_SCANNER_*` from the tables above | Not mapped; rename explicitly. |
Stale-upload cleanup differs in one crash-recovery detail. RustFS's stale multipart cleanup (`cleanup_stale_multipart_uploads_in_set`) takes a namespace write lock, re-checks the upload under write quorum (`check_multipart_upload_path_exists`), and fans the delete out to every disk, where the local recursive delete renames the directory into `.rustfs.sys/tmp/.trash/<uuid>` (`move_to_trash`). If the process dies mid fan-out after more disks than the parity count have already moved the upload directory, the next cleanup pass's quorum re-check fails (`FileNotFound` is not in `OBJECT_OP_IGNORED_ERRS`) and the candidate is skipped, so the remaining per-disk residue is not reclaimed by that job. The residue is invisible to the S3 API and only consumes disk space; the window is milliseconds wide. MinIO processes each disk independently and converges in the same scenario.
Stale-upload cleanup takes the upload namespace write lock and rechecks cleanup authority before deletion. An incomplete legacy upload without sufficient identity or deletion quorum remains on disk; age and absence from an S3 listing do not establish that its recovery or publication authority has ended. A native Abort with a complete, validated closure certificate can recover a smaller residual after partial deletion, subject to the original-participant checks. Missing, unavailable or corrupt proof keeps the operation conservative. For an otherwise valid native Abort certificate, an unavailable participant identity is a retryable write-quorum failure; a known conflicting identity remains a corruption error, including when another participant is unavailable. See [the multipart authority contract](../architecture/erasure-coding.md#7-write-path-and-write-quorum).
Size-filtered incomplete-upload rules also require every participant to answer the part listing and matching part metadata at write quorum. Unknown size skips only size-dependent rules. During a mixed-version upgrade, an older node can still evaluate those rules using the older empty-list behavior; fleet-wide protection requires every cleanup worker to run the updated version. Plan the rollout so older workers cannot perform size-filtered aborts while mixed versions are active.
The `rustfs_stale_multipart_cleanup_blocked_total` counter has a bounded `stage` label: `recheck` for lock or authority recheck failures, `lock_lost` when a held cleanup lock is lost, and `delete` for a failed authorized deletion. It counts attempts, not unique uploads or bytes, and includes transient failures; it carries no bucket, key or upload labels. The `multipart_cleanup_lock_or_recheck_failed` debug event identifies a candidate path and the failure; `multipart_cleanup_lock_lost` identifies the path when a previously acquired cleanup lock is lost. Restore unavailable disks and retry through supported S3 or background recovery paths when their prerequisites hold. Preserve the disk state and seek an evidence-based recovery procedure when authority remains uncertain; do not recursively remove a residual solely because it is old or a destination object exists. The default expiry is measured from upload initiation, including internal data-movement uploads, and can affect long-running uploads even without a bucket lifecycle rule. This differs from [Amazon S3](https://docs.aws.amazon.com/AmazonS3/latest/userguide/mpuoverview.html), which retains parts until the upload is completed or stopped and offers an explicit lifecycle rule to abort incomplete uploads. The interval is a schedule, not a guaranteed reclamation deadline.
## Replacement Recovery Completion
+3 -3
View File
@@ -158,7 +158,7 @@ When a session ends mid-upload the server aborts the in-flight multipart upload.
- a burst of simultaneous disconnects exhausts the global abort task pool (sized at twice the CPU parallelism, between `ABORT_PERMITS_FLOOR` 8 and `ABORT_PERMITS_CEILING` 128);
- the server process is killed outright.
Orphaned uploads are not permanent. The background stale-upload cleanup aborts incomplete multipart uploads older than `DEFAULT_STALE_UPLOADS_EXPIRY` (24 hours), checked every `DEFAULT_STALE_UPLOADS_CLEANUP_INTERVAL` (6 hours). An `AbortIncompleteMultipartUpload` bucket lifecycle rule reclaims them as well and gives per-bucket control of the window. The log lines when a session-drop abort is skipped or fails:
Background stale-upload cleanup considers incomplete multipart uploads after `DEFAULT_STALE_UPLOADS_EXPIRY` (24 hours from initiation), on a `DEFAULT_STALE_UPLOADS_CLEANUP_INTERVAL` schedule (6 hours). A bucket `AbortIncompleteMultipartUpload` rule can make an upload eligible earlier. Both paths retain their lock and cleanup-authority checks: low-quorum or otherwise uncertain residue can remain, so neither the schedule nor a lifecycle rule guarantees physical reclamation. See [stale-upload recovery](scanner-runtime-controls.md#migrating-from-minio-scanner-settings). The log lines when a session-drop abort is skipped or fails:
```text
skipped abort of orphaned multipart upload on session drop, principal lacks s3:AbortMultipartUpload, bucket lifecycle rules must reclaim parts
@@ -187,6 +187,6 @@ Every line below logs at warn level except `SFTP server listening` and `SFTP aut
| `SFTP write would exceed the S3 multipart parts limit` | Client hit the per-upload size cap; raise the part size if legitimate. |
| `SFTP close rejected: trailing part would exceed S3 multipart parts limit` | Same cap hit at file close. |
| `root READDIR truncated` | A principal can see more than 10000 buckets; listing was cut off. |
| `skipped abort of orphaned multipart upload` | Orphaned parts on storage until background cleanup or a lifecycle rule reclaims them. |
| `abort permit pool exhausted on session drop` | Mass-disconnect burst; orphaned parts on storage until reclaimed. |
| `skipped abort of orphaned multipart upload` | Parts may remain until cleanup has sufficient authority to reclaim them. |
| `abort permit pool exhausted on session drop` | Mass-disconnect burst; parts may remain when cleanup authority is unavailable. |
| `RUSTFS_SFTP_` prefix in a warning | A tuning variable was rejected and its default applied. |
+7 -1
View File
@@ -108,11 +108,17 @@ Membership is pinned by `.config/e2e-distributed-selection.txt`. Update the Linu
## Multipart fence fault acceptance
The EC:2 metadata-delay case also holds the three scoped `ListDir` RPCs only after the upload metadata has been read, while the coordinator's part directory is unavailable. It requires the ListParts response to be HTTP 503 `SlowDownRead`, then restores the exact directory, verifies the part ETag and completes the original bytes through every reader. The same case signs a ten-byte multipart limit and races two six-byte parts through different coordinators; exactly one must succeed, and the final object must match that part.
The four-node rolling and direct upgrade cases seed a multipart upload on the verified previous release. Rolling checks compare part number, size and ETag through old and new nodes while multipart mutations are paused. With one old disk offline and the other three part directories temporarily absent, metadata remains readable: the previous coordinator returns an empty list while the current coordinator returns HTTP 503 `SlowDownRead`. After the directories and disk return, both coordinators must list the original ETag and all four original metadata, part and part-metadata files must remain byte-identical. Once all nodes are current, parity-zero probes prove each coordinator can write through every disk before the single completion request; both upgrade cases then check every reader. This read-only fault case does not permit multipart mutations during a mixed rollout.
The nightly stale-multipart cleanup case also retains one and two EC:2 source copies on online disks. It waits for a path-specific background recheck diagnostic, verifies every surviving directory and file byte, and checks that the diagnostic is not attributed to a client UploadPart. This is evidence for conservative preservation and diagnosis, not automatic reclamation without a closure certificate. Node logs are retained when `RUSTFS_E2E_LOG_DIR` is set.
The `distributed::multipart_fence_acceptance_test` cases separate remote storage RPC delays from lock RPC delays through a method-selective proxy. They require the selected RPC to reach its pause before checking quorum, recovery, or cancellation. The proxy forwards opaque bodies and gRPC trailers and is confined to the test process. It does not change server timeouts, locking, retry policy, or the commit path.
After a quorum-minus-one Complete is rejected and faults are retired, ordinary Abort must succeed. `NoSuchUpload` is not accepted as recovery of that failed completion: it can hide staging metadata consumed by partial publication. Already completed or aborted sessions retain their existing missing-upload cleanup semantics. The ECStore multipart regressions assert unchanged per-disk authority bytes across primary/recovery locations, no duplicated disk votes, rejection of completed-upload resends after cleanup failure, orphan-cleanup refusal while authority remains, preservation of the previous object, and write-quorum completion replay. Scoped Abort faults decode the exact multipart volume, root path and recursive option without changing signed bytes. Post-fault read probes can wait within the existing recovery deadline for classified 503 responses, honoring Retry-After; injected writes and successful streaming bodies remain single-attempt.
`native_certified_partial_abort_retry_reclaims_survivors_below_metadata_quorum` first closes all sixteen native participants, then holds five exact-root `RecoverMultipartSource(DeleteAborted)` RPCs in EC12+4. It observes eleven physical deletions and a typed write-quorum failure. The five survivors must retain their complete original shards, Aborting state and identical full-N certificate bound by each actual owner; no ACTIVE head or old capability remains. After peer recovery the same upload retries successfully, authenticating the stored closure and deleting only those five roots, with no remaining source/control record and unchanged committed-object bytes through every reader. Historical unmarked primary/recovery authority still has its separate four-case ECStore regression: without native closure, five standalone originals remain insufficient and must be retained rather than treated as absence. The explicit late-delivery proxy mode retains bounded control packets beyond caller cancellation; dropping that mode discards them, and proxy teardown aborts its retained tasks. Late Complete acceptance delivers the original signed RenameMultipartData packets after successful Abort, staging GC and same-name bucket recreation, and requires an authenticated mutation refusal plus unchanged new-object bytes. This covers empty and non-inline parts; expired-packet discard remains a distinct fault model.
`native_certified_partial_abort_retry_reclaims_survivors_below_metadata_quorum` first closes all sixteen native participants, then holds five exact-root `RecoverMultipartSource(DeleteAborted)` RPCs in EC12+4. It observes eleven physical deletions and a typed write-quorum failure. The five survivors must retain their complete original shards, Aborting state and identical full-N certificate bound by each actual owner; no ACTIVE head or old capability remains. After peer recovery the test first stops an original participant and requires a `storage_write_quorum` refusal with byte-identical retained sources. It restores that node, then kills and restarts the coordinator and a node owning a surviving root, verifies actual storage routes and byte-identical surviving sources, and retries the same upload once. The retry authenticates the stored closure and deletes only those five roots, with no remaining source/control record and unchanged committed-object bytes through every reader. The ECStore `stale_certified_residual_survives_lock_loss_then_converges` regression separately checks one surviving EC:2 root: an unavailable original participant and a lost cleanup-lock heartbeat both refuse deletion, preserve the source and shard, and permit cleanup after recovery. Historical unmarked primary/recovery authority still has its separate four-case ECStore regression: without native closure, five standalone originals remain insufficient and must be retained rather than treated as absence. The explicit late-delivery proxy mode retains bounded control packets beyond caller cancellation; dropping that mode discards them, and proxy teardown aborts its retained tasks. Late Complete acceptance delivers the original signed RenameMultipartData packets after successful Abort, staging GC and same-name bucket recreation, and requires an authenticated mutation refusal plus unchanged new-object bytes. This covers empty and non-inline parts; expired-packet discard remains a distinct fault model.
The pinned previous-release MPU recovery case bootstraps and seeds all sixteen disks through the release's normal direct transport. It then installs the method observers on the same advertised localhost ports immediately before restarting the current coordinator, whose fresh RPC channels must traverse them. The observers must receive exact, read-only native Inspect rejections from the independently constructed set of four configured disk endpoint URLs on each old peer; different or duplicate routes cannot substitute for those participants, and old direct connections cannot satisfy that evidence. While three peers remain old, the current coordinator must return a single typed write-quorum refusal before dispatching any old or native publication RPC. Every original source directory, metadata file and shard remains byte-for-byte intact, the part ETag stays readable and no target appears. Multipart mutations are not sent through an old coordinator during rollout. After every peer is current and all configured storage routes respond, the same upload must Abort successfully, reclaim all sixteen roots and preserve the unrelated historical object through every reader. This tests the supported mixed-version pause and recovery boundary. The same case runs for every verified source in the [upgrade compatibility workflow](../../.github/workflows/e2e-upgrade.yml), with an exact one-test listing check and per-version node/test logs. This supplements the distributed suite's historical source pin.
+2 -1
View File
@@ -297,7 +297,7 @@ tokio-rustls = { workspace = true, default-features = false, features = ["loggin
aws-smithy-runtime-api = { workspace = true, features = ["http-1x"] }
aws-smithy-types = { workspace = true }
google-cloud-auth = { workspace = true, optional = true }
moka = { workspace = true, features = ["sync"] }
moka = { workspace = true, features = ["future", "sync"] }
xxhash-rust = { workspace = true, features = ["xxh3"] }
aws-sdk-s3 = { workspace = true, default-features = false, features = ["sigv4a", "default-https-client", "rt-tokio"] }
tokio-stream.workspace = true
@@ -392,6 +392,7 @@ object = { workspace = true, optional = true, features = ["read", "std", "elf"]
symbolic-demangle = { version = "=13.9.0", default-features = false, features = ["rust"], optional = true }
[dev-dependencies]
rustfs-targets = { workspace = true, features = ["test-support"] }
uuid = { workspace = true, features = ["v4", "v5", "fast-rng", "macro-diagnostics"] }
rustfs-iam = { workspace = true, features = ["test-util"] }
serial_test = { workspace = true }
+218 -15
View File
@@ -50,7 +50,7 @@ use tracing::{error, info, warn};
const LOG_COMPONENT_ADMIN_API: &str = "admin_api";
async fn converge_target_mutation_on_cluster(
pub(super) async fn converge_target_mutation_on_cluster(
context: Option<&AppContext>,
target_type: &str,
local_result: Result<(), rustfs_notify::NotificationError>,
@@ -60,18 +60,26 @@ async fn converge_target_mutation_on_cluster(
match (local_result, peer_result) {
(Ok(()), Ok(())) => Ok(()),
(Err(local), Ok(())) => {
warn!(target_type, error = %local, "Local notification runtime failed to converge");
Err(s3_error!(InternalError, "local notification runtime failed to converge"))
}
(Ok(()), Err(peer)) => Err(peer),
(Err(local), Err(peer)) => {
warn!(target_type, error = %local, "Local notification runtime failed while peer convergence also failed");
Err(s3_error!(
InternalError,
"local notification runtime and peer convergence failed: {}",
peer
))
(Ok(()), Err(_)) => Err(s3_error!(
InternalError,
"notification target update completed locally but peer convergence failed"
)),
(Err(local), peer) => {
let peer_failed = peer.is_err();
let message = match local {
rustfs_notify::NotificationError::ConfigurationConvergence { persisted: true, .. } => {
"notification target configuration was persisted but local runtime convergence failed"
}
rustfs_notify::NotificationError::ConfigurationConvergence { persisted: false, .. } => {
"notification target configuration was unchanged but local runtime convergence failed"
}
_ => "notification target update failed; persistence status is unknown",
};
if peer_failed {
Err(s3_error!(InternalError, "{}; peer convergence also failed", message))
} else {
Err(s3_error!(InternalError, "{}", message))
}
}
}
}
@@ -84,7 +92,7 @@ fn notification_target_subsystem(target_type: &str) -> S3Result<&'static str> {
.ok_or_else(|| s3_error!(InvalidArgument, "unsupported notification target type: {}", target_type))
}
async fn preflight_target_mutation_on_cluster(context: Option<&AppContext>, target_type: &str) -> S3Result<()> {
pub(super) async fn preflight_target_mutation_on_cluster(context: Option<&AppContext>, target_type: &str) -> S3Result<()> {
preflight_dynamic_config_reload_for_context(context, notification_target_subsystem(target_type)?).await
}
@@ -682,7 +690,7 @@ mod tests {
use crate::admin::handlers::{
audit::ListAuditTargets,
extensions::ListExtensionInstancesHandler,
plugins_instances::{GetPluginInstanceHandler, ListPluginInstancesHandler},
plugins_instances::{GetPluginInstanceHandler, ListPluginInstancesHandler, PutPluginInstanceHandler},
target_read::load_target_read_snapshot,
};
use crate::admin::storage_api::config::{
@@ -758,6 +766,154 @@ mod tests {
.await
.expect("persist target without loading it into the runtime");
// A concurrent credential rotation must be observed under the commit transaction.
let initial = crate::admin::storage_api::config::read_admin_config_without_migrate(env.ecstore.clone())
.await
.expect("read initial notify config");
let generation_before = {
let snapshot = crate::admin::storage_api::config::read_admin_server_config_snapshot(env.ecstore.clone())
.await
.expect("read initial generation");
snapshot.generation().expect("test config has an object generation")
};
let expected = initial.0[NOTIFY_WEBHOOK_SUB_SYS]["unavailable"].clone();
let expected_default = initial.0[NOTIFY_WEBHOOK_SUB_SYS]
.get(rustfs_config::DEFAULT_DELIMITER)
.cloned();
let mut rotated = initial.clone();
rotated
.0
.get_mut(NOTIFY_WEBHOOK_SUB_SYS)
.expect("subsystem")
.get_mut("unavailable")
.expect("target")
.insert(rustfs_config::WEBHOOK_AUTH_TOKEN.to_owned(), "rotated-secret".to_owned());
save_admin_server_config(env.ecstore.clone(), &rotated)
.await
.expect("persist concurrent rotation");
let generation_after = {
let snapshot = crate::admin::storage_api::config::read_admin_server_config_snapshot(env.ecstore.clone())
.await
.expect("read rotated generation");
snapshot.generation().expect("rotated config has an object generation")
};
assert_ne!(generation_before, generation_after, "content change must publish a new generation");
let system = rustfs_notify::NotificationSystem::new(Config::default());
let result = system
.set_target_config_if_unchanged(
NOTIFY_WEBHOOK_SUB_SYS,
"unavailable",
Some(expected.clone()),
expected_default,
None,
expected,
)
.await;
assert!(matches!(result, Err(rustfs_notify::NotificationError::ConfigurationConflict)));
let actual = crate::admin::storage_api::config::read_admin_config_without_migrate(env.ecstore.clone())
.await
.expect("read rotated config");
assert_eq!(
actual.0[NOTIFY_WEBHOOK_SUB_SYS]["unavailable"]
.lookup(rustfs_config::WEBHOOK_AUTH_TOKEN)
.as_deref(),
Some("rotated-secret")
);
save_admin_server_config(env.ecstore.clone(), &initial)
.await
.expect("restore inventory config");
let generation = {
let snapshot = crate::admin::storage_api::config::read_admin_server_config_snapshot(env.ecstore.clone())
.await
.expect("read revision baseline");
snapshot.generation().expect("baseline generation").to_string()
};
let mut unrelated = initial.clone();
let mut other = KVS::new();
other.insert("enable".into(), "off".into());
other.insert("endpoint".into(), "http://127.0.0.1:9/other".into());
unrelated
.0
.entry(NOTIFY_WEBHOOK_SUB_SYS.to_owned())
.or_default()
.insert("other".to_owned(), other);
save_admin_server_config(env.ecstore.clone(), &unrelated)
.await
.expect("persist unrelated update");
let target = initial.0[NOTIFY_WEBHOOK_SUB_SYS]["unavailable"].clone();
let default = initial.0[NOTIFY_WEBHOOK_SUB_SYS]
.get(rustfs_config::DEFAULT_DELIMITER)
.cloned();
let result = system
.set_target_config_if_unchanged(
NOTIFY_WEBHOOK_SUB_SYS,
"unavailable",
Some(target.clone()),
default,
Some(generation),
target,
)
.await;
assert!(
matches!(result, Err(rustfs_notify::NotificationError::ConfigurationConflict)),
"revision changes must reject even when target remains unchanged"
);
save_admin_server_config(env.ecstore.clone(), &initial)
.await
.expect("restore after unrelated update");
let mut with_default = initial.clone();
let mut default = KVS::new();
default.insert("queue_limit".into(), "15".into());
with_default
.0
.get_mut(NOTIFY_WEBHOOK_SUB_SYS)
.expect("subsystem")
.insert("_".into(), default.clone());
save_admin_server_config(env.ecstore.clone(), &with_default)
.await
.expect("persist inherited default");
let saved = crate::admin::storage_api::config::read_admin_config_without_migrate(env.ecstore.clone())
.await
.expect("read inherited config");
let original = saved.0[NOTIFY_WEBHOOK_SUB_SYS]["unavailable"].clone();
let old_default = saved.0[NOTIFY_WEBHOOK_SUB_SYS].get("_").cloned();
let expected = std::collections::HashMap::from([
("endpoint".to_owned(), "http://127.0.0.1:9/hook".to_owned()),
("queue_limit".to_owned(), "15".to_owned()),
]);
let mut validated = original.clone();
validated.insert("endpoint".into(), "http://127.0.0.1:9/changed".into());
validated.insert("queue_limit".into(), "15".into());
let update = crate::admin::handlers::plugins_instances::merge_conditional_target_update(
&original,
&expected,
&["endpoint".into()],
&validated,
)
.expect("merge only edited field");
let result = system
.set_target_config_if_unchanged(NOTIFY_WEBHOOK_SUB_SYS, "unavailable", Some(original), old_default, None, update)
.await;
assert!(
result.is_ok()
|| matches!(
result,
Err(rustfs_notify::NotificationError::ConfigurationConvergence { persisted: true, .. })
)
);
let saved = crate::admin::storage_api::config::read_admin_config_without_migrate(env.ecstore.clone())
.await
.expect("read conditional update");
assert_eq!(
saved.0[NOTIFY_WEBHOOK_SUB_SYS]["unavailable"].lookup("endpoint").as_deref(),
Some("http://127.0.0.1:9/changed")
);
assert_eq!(saved.0[NOTIFY_WEBHOOK_SUB_SYS]["unavailable"].lookup("queue_limit"), None);
assert_eq!(saved.0[NOTIFY_WEBHOOK_SUB_SYS]["_"].lookup("queue_limit").as_deref(), Some("15"));
save_admin_server_config(env.ecstore.clone(), &initial)
.await
.expect("restore inventory config");
let (locked_tx, locked_rx) = tokio::sync::oneshot::channel();
let (release_tx, release_rx) = tokio::sync::oneshot::channel();
let store = env.ecstore.clone();
@@ -785,6 +941,53 @@ mod tests {
assert!(observed_config.0[NOTIFY_WEBHOOK_SUB_SYS].contains_key("unavailable"));
mark_event_notifier_unreconciled();
// Conditional edits reject stale revisions before target validation or persistence.
let mut router = Router::new();
router.insert("/{id}", ()).expect("plugin detail route");
let matched = router.at("/builtin:webhook:notify:unavailable").expect("plugin params");
let transaction_guard = crate::admin::storage_api::config::read_admin_server_config_snapshot(env.ecstore.clone())
.await
.expect("hold config transaction lock");
let generation = transaction_guard.generation().expect("config generation").to_string();
let detail = response_json(GetPluginInstanceHandler {}.call(
request("/rustfs/admin/v4/plugins/instances/builtin:webhook:notify:unavailable"),
matched.params,
))
.await
.expect("read conditional edit baseline");
let revision = detail["config_revision"].as_str().expect("GET includes opaque revision");
assert_eq!(
revision,
crate::admin::handlers::plugins_instances::config_revision("builtin:webhook:notify:unavailable", &generation)
);
assert_eq!(detail["config"]["auth_token"], "***redacted***");
drop(transaction_guard);
let mut put = request("/rustfs/admin/v4/plugins/instances/builtin:webhook:notify:unavailable");
put.method = Method::PUT;
put.input = Body::from(
serde_json::to_vec(&serde_json::json!({
"expected_config": detail["config"],
"expected_revision": "stale",
"key_values": [{"key":"auth_token", "value":"replacement-secret"}]
}))
.expect("encode conditional edit"),
);
let matched = router.at("/builtin:webhook:notify:unavailable").expect("plugin params");
let err = PutPluginInstanceHandler {}
.call(put, matched.params)
.await
.expect_err("stale revision must be rejected");
assert_eq!(err.code(), &s3s::S3ErrorCode::InvalidRequest);
let after = crate::admin::storage_api::config::read_admin_config_without_migrate(env.ecstore.clone())
.await
.expect("read unchanged config");
assert_eq!(
after.0[NOTIFY_WEBHOOK_SUB_SYS]["unavailable"]
.lookup(rustfs_config::WEBHOOK_AUTH_TOKEN)
.as_deref(),
Some("inventory-secret-token")
);
with_notify_runtime_reconcile_lock(async {
for initialize_runtime in [false, true] {
if initialize_runtime {
+448 -26
View File
@@ -15,23 +15,29 @@
use crate::admin::{
auth::authorize_admin_request,
handlers::audit_runtime_config::{remove_audit_target_config, set_audit_target_config},
handlers::event::{converge_target_mutation_on_cluster, preflight_target_mutation_on_cluster},
handlers::notify_runtime_access::{
load_notification_mutation_snapshot, remove_notification_target_config, set_notification_target_config,
get_notification_system, load_notification_mutation_snapshot, remove_notification_target_config,
set_notification_target_config,
},
handlers::supervise_admin_mutation,
handlers::target_descriptor::{
AdminTargetSpec, RuntimeHealthStatus, TargetEndpointSource, TargetInstanceReadModel, admin_target_spec_from_builtin,
build_enabled_target_kvs, build_json_response, collect_target_instances, find_target_instance,
target_module_disabled_reason, target_mutation_block_reason as shared_target_mutation_block_reason,
allowed_target_keys, build_enabled_target_kvs, build_json_response, build_validated_target_kvs, collect_target_instances,
collect_validated_key_values, find_target_instance, target_module_disabled_reason,
target_mutation_block_reason as shared_target_mutation_block_reason,
},
handlers::target_read::{
collect_audit_runtime_statuses, collect_notification_runtime_statuses, load_server_config_from_store,
load_target_read_snapshot,
collect_audit_runtime_statuses, collect_notification_instance_runtime_statuses, collect_notification_runtime_statuses,
load_server_config_from_store, load_target_read_snapshot,
},
plugin_contract::{
PluginContractDomain, PluginInstanceDetail, PluginInstanceDiagnostic, PluginInstanceDiagnosticCode,
PluginInstanceDiagnosticCount, PluginInstanceEntry, PluginInstanceSource, PluginInstancesResponse,
},
router::{AdminOperation, Operation, S3Router},
runtime_sources::{app_context_from_req, current_object_store_handle_for_context},
storage_api::config::{read_admin_server_config_snapshot, read_admin_server_config_with_generation},
storage_api::s3::{self, S3ErrorCode},
};
use crate::server::{
@@ -48,6 +54,7 @@ use rustfs_policy::policy::action::{Action, AdminAction};
use rustfs_targets::catalog::builtin::{builtin_audit_target_admin_descriptors, builtin_notify_target_admin_descriptors};
use rustfs_targets::{builtin_target_plugin_operational_state, runtime_state_from_status_label};
use s3s::{Body, S3Request, S3Response, S3Result, s3_error};
use sha2::{Digest, Sha256};
use std::collections::{BTreeMap, HashMap};
use std::sync::LazyLock;
use tracing::warn;
@@ -281,6 +288,7 @@ fn plugin_instance_detail(instance: TargetInstanceReadModel) -> PluginInstanceDe
mapped.diagnostic_codes = diagnostics.iter().map(|item| item.code.clone()).collect();
PluginInstanceDetail {
config_revision: None,
instance: mapped,
diagnostics,
}
@@ -320,6 +328,10 @@ struct KeyValue {
#[serde(deny_unknown_fields)]
struct PluginInstanceBody {
key_values: Vec<KeyValue>,
#[serde(default)]
expected_config: Option<HashMap<String, String>>,
#[serde(default)]
expected_revision: Option<String>,
}
#[derive(Debug, Default, Clone, PartialEq, Eq)]
@@ -683,6 +695,106 @@ async fn find_plugin_instance(instance_id: &str) -> S3Result<Option<TargetInstan
find_target_instance(context.specs, context.route_prefix, &config, runtime_statuses, instance_id)
}
fn effective_validation_fields(
context: PluginInstanceDomainContext,
subsystem: &str,
effective: &KVS,
requested: &[KeyValue],
) -> S3Result<HashMap<String, String>> {
let allowed = allowed_target_keys(context.specs, subsystem);
let requested = collect_validated_key_values(
requested.iter().map(|kv| (kv.key.as_str(), kv.value.as_str())),
&allowed,
subsystem,
"plugin instance",
)?;
let mut candidate: HashMap<String, String> = effective
.0
.iter()
.filter(|kv| allowed.contains(kv.key.as_str()))
.map(|kv| (kv.key.clone(), kv.value.clone()))
.collect();
candidate.extend(requested);
Ok(candidate)
}
pub(super) fn merge_conditional_target_update(
original: &KVS,
expected: &HashMap<String, String>,
changed_keys: &[String],
validated: &KVS,
) -> S3Result<KVS> {
let mut result = original.clone();
for key in changed_keys {
if key == rustfs_config::ENABLE_KEY {
return Err(s3::error(
S3ErrorCode::InvalidRequest,
"target enable state cannot be changed using expected_config",
));
}
let value = validated
.lookup(key)
.ok_or_else(|| s3::error(S3ErrorCode::InvalidArgument, "missing validated target field"))?;
if expected.get(key) != Some(&value) {
result.insert(key.clone(), value);
}
}
Ok(result)
}
pub(super) fn config_revision(instance_id: &str, generation: &str) -> String {
let mut hasher = Sha256::new();
hasher.update(b"rustfs-plugin-instance-config-revision-v1\0");
hasher.update(generation.as_bytes());
hasher.update(b"\0");
hasher.update(instance_id.as_bytes());
faster_hex::hex_string(&hasher.finalize())
}
fn validate_expected_config(
context: PluginInstanceDomainContext,
config: &Config,
instance_id: &str,
expected: &HashMap<String, String>,
) -> S3Result<TargetInstanceReadModel> {
if context.domain != PluginContractDomain::Notify {
return Err(s3::error(
S3ErrorCode::InvalidArgument,
"expected_config is only supported for notification targets",
));
}
let resolved = resolve_plugin_instance_target(instance_id)?;
let instance = find_target_instance(
std::slice::from_ref(&resolved.target_spec),
context.route_prefix,
config,
HashMap::new(),
instance_id,
)?
.filter(|instance| instance.source == TargetEndpointSource::Config)
.ok_or_else(|| s3::error(S3ErrorCode::InvalidRequest, "expected_config requires an existing config-managed target"))?;
if instance.has_env_default {
return Err(s3::error(
S3ErrorCode::InvalidRequest,
"notification target inherits environment configuration and cannot be edited",
));
}
if !instance.enabled {
return Err(s3::error(
S3ErrorCode::InvalidRequest,
"disabled targets cannot be edited with expected_config",
));
}
if &map_instance_config(instance.config.clone(), secret_fields_for_instance(context.domain, &instance.subsystem)) != expected
{
return Err(s3::error(
S3ErrorCode::InvalidRequest,
"target configuration changed since it was read; refresh and retry",
));
}
Ok(instance)
}
async fn set_plugin_instance_config(
context: PluginInstanceDomainContext,
resolved: &ResolvedPluginInstanceTarget,
@@ -744,12 +856,31 @@ impl Operation for GetPluginInstanceHandler {
.ok_or_else(|| s3_error!(InvalidArgument, "missing required parameter: 'id'"))?;
refresh_persisted_module_switches().await?;
let instance = find_plugin_instance(instance_id)
.await?
.ok_or_else(|| s3_error!(NoSuchKey, "plugin instance not found"))?;
let data = serde_json::to_vec(&plugin_instance_detail(instance))
.map_err(|e| s3_error!(InternalError, "failed to serialize response: {}", e))?;
let domain = parse_plugin_instance_id(instance_id)?.1;
let (instance, generation) = if domain == PluginContractDomain::Notify {
let app_context = app_context_from_req(&req);
let context = plugin_instance_domain_context(domain);
let read_config = async {
match current_object_store_handle_for_context(app_context.as_deref()) {
Some(store) => read_admin_server_config_with_generation(store)
.await
.map_err(|_| s3::error(S3ErrorCode::InternalError, "failed to read server config")),
None => Ok((Config::new(), None)),
}
};
let ((config, generation), runtime) = tokio::try_join!(read_config, async {
Ok::<_, s3s::S3Error>(collect_notification_instance_runtime_statuses(context.specs, instance_id).await)
})?;
let instance = find_target_instance(context.specs, context.route_prefix, &config, runtime, instance_id)?;
(instance, generation.map(|value| value.to_string()))
} else {
(find_plugin_instance(instance_id).await?, None)
};
let instance = instance.ok_or_else(|| s3::error(S3ErrorCode::NoSuchKey, "plugin instance not found"))?;
let mut detail = plugin_instance_detail(instance);
detail.config_revision = generation.map(|value| config_revision(instance_id, &value));
let data = serde_json::to_vec(&detail)
.map_err(|e| s3::error(S3ErrorCode::InternalError, format!("failed to serialize response: {}", e)))?;
Ok(build_json_response(StatusCode::OK, Body::from(data), req.headers.get("x-request-id")))
}
}
@@ -760,6 +891,7 @@ pub struct PutPluginInstanceHandler {}
impl Operation for PutPluginInstanceHandler {
async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
authorize_plugin_instance_write_request(&req).await?;
let app_context = app_context_from_req(&req);
let instance_id = params
.get("id")
.ok_or_else(|| s3_error!(InvalidArgument, "missing required parameter: 'id'"))?;
@@ -780,7 +912,16 @@ impl Operation for PutPluginInstanceHandler {
let mut body: PluginInstanceBody = serde_json::from_slice(&body_bytes)
.map_err(|e| s3_error!(InvalidArgument, "invalid json body for plugin instance config: {}", e))?;
let config_snapshot = plugin_instance_mutation_config_snapshot(context).await?;
let (config_snapshot, generation) = if context.domain == PluginContractDomain::Notify {
let store = current_object_store_handle_for_context(app_context.as_deref())
.ok_or_else(|| s3::error(S3ErrorCode::InternalError, "server storage not initialized"))?;
let snapshot = read_admin_server_config_snapshot(store)
.await
.map_err(|_| s3::error(S3ErrorCode::InternalError, "failed to read server config"))?;
(snapshot.config.clone(), snapshot.generation().map(|value| value.to_string()))
} else {
(plugin_instance_mutation_config_snapshot(context).await?, None)
};
if let Some(reason) = plugin_instance_mutation_block_reason(
context,
&config_snapshot,
@@ -791,25 +932,104 @@ impl Operation for PutPluginInstanceHandler {
return Err(s3_error!(InvalidRequest, "{reason}"));
}
if body.expected_revision.is_some() && body.expected_config.is_none() {
return Err(s3::error(S3ErrorCode::InvalidArgument, "expected_revision requires expected_config"));
}
let effective_instance = if let Some(expected) = body.expected_config.as_ref() {
let revision = generation
.as_ref()
.ok_or_else(|| s3::error(S3ErrorCode::InvalidRequest, "target configuration revision unavailable"))?;
if body.expected_revision.as_deref() != Some(config_revision(instance_id, revision).as_str()) {
return Err(s3::error(
S3ErrorCode::InvalidRequest,
"target configuration changed since it was read; refresh and retry",
));
}
Some(validate_expected_config(context, &config_snapshot, instance_id, expected)?.config)
} else {
None
};
let changed_keys: Vec<String> = body
.expected_config
.as_ref()
.map(|expected| {
body.key_values
.iter()
.filter(|kv| expected.get(&kv.key) != Some(&kv.value))
.map(|kv| kv.key.clone())
.collect()
})
.unwrap_or_default();
// GET responses redact secret values; a read-modify-write round trip
// sends the placeholder back, which must restore the stored secret
// instead of persisting the placeholder literal.
restore_redacted_secret_values(
&mut body.key_values,
resolved.target_spec.secret_fields,
stored_instance_config(&config_snapshot, resolved.target_spec.subsystem, &resolved.target_name),
)?;
let stored_secrets = if body.expected_config.is_some() && context.domain == PluginContractDomain::Notify {
effective_instance.as_ref()
} else {
stored_instance_config(&config_snapshot, resolved.target_spec.subsystem, &resolved.target_name)
};
restore_redacted_secret_values(&mut body.key_values, resolved.target_spec.secret_fields, stored_secrets)?;
let kvs = build_enabled_target_kvs(
context.specs,
body.key_values.iter().map(|kv| (kv.key.as_str(), kv.value.as_str())),
resolved.target_spec.subsystem,
context.default_queue_dir,
"plugin instance",
)
.await?;
let validated_kvs = if let Some(effective) = effective_instance.as_ref() {
let candidate = effective_validation_fields(context, resolved.target_spec.subsystem, effective, &body.key_values)?;
build_validated_target_kvs(&resolved.target_spec, candidate, context.default_queue_dir).await?
} else {
build_enabled_target_kvs(
context.specs,
body.key_values.iter().map(|kv| (kv.key.as_str(), kv.value.as_str())),
resolved.target_spec.subsystem,
context.default_queue_dir,
"plugin instance",
)
.await?
};
set_plugin_instance_config(context, &resolved, kvs).await?;
let kvs = if let Some(expected_config) = body.expected_config.as_ref() {
let original = stored_instance_config(&config_snapshot, resolved.target_spec.subsystem, &resolved.target_name)
.ok_or_else(|| {
s3::error(S3ErrorCode::InvalidRequest, "expected_config requires an existing config-managed target")
})?;
merge_conditional_target_update(original, expected_config, &changed_keys, &validated_kvs)?
} else {
validated_kvs
};
if context.domain == PluginContractDomain::Notify {
let expected =
stored_instance_config(&config_snapshot, resolved.target_spec.subsystem, &resolved.target_name).cloned();
let expected_default =
stored_instance_config(&config_snapshot, resolved.target_spec.subsystem, rustfs_config::DEFAULT_DELIMITER)
.cloned();
let expected_generation = if body.expected_config.is_some() { generation } else { None };
let system = get_notification_system().await?;
let subsystem = resolved.target_spec.subsystem.to_owned();
let target_name = resolved.target_name.clone();
supervise_admin_mutation("plugin notification target mutation", async move {
preflight_target_mutation_on_cluster(app_context.as_deref(), &subsystem).await?;
let local = system
.set_target_config_if_unchanged(
&subsystem,
&target_name,
expected,
expected_default,
expected_generation,
kvs,
)
.await;
if matches!(local, Err(rustfs_notify::NotificationError::ConfigurationConflict)) {
return Err(s3::error(
S3ErrorCode::InvalidRequest,
"target configuration changed during validation; retry the request",
));
}
converge_target_mutation_on_cluster(app_context.as_deref(), &subsystem, local).await
})
.await?;
} else {
set_plugin_instance_config(context, &resolved, kvs).await?;
}
Ok(build_json_response(StatusCode::OK, Body::empty(), req.headers.get("x-request-id")))
}
@@ -1102,6 +1322,7 @@ mod tests {
health_state: "offline".to_string(),
health_reason: "not_loaded_in_runtime".to_string(),
runtime_present: false,
has_env_default: false,
source: TargetEndpointSource::Config,
enabled: true,
config: enabled_kvs("on"),
@@ -1126,6 +1347,7 @@ mod tests {
health_state: "online".to_string(),
health_reason: "reachable".to_string(),
runtime_present: true,
has_env_default: false,
source: TargetEndpointSource::Config,
enabled: true,
config: KVS(vec![
@@ -1218,6 +1440,202 @@ mod tests {
assert_eq!(found.lookup(WEBHOOK_AUTH_TOKEN).as_deref(), Some("stored-secret-token"));
}
#[test]
fn expected_config_round_trips_redacted_read_and_rejects_stale_values() {
let mut target = KVS::new();
target.insert(ENABLE_KEY.into(), "on".into());
target.insert(WEBHOOK_ENDPOINT.into(), "https://example.com/a".into());
target.insert(WEBHOOK_AUTH_TOKEN.into(), "secret-one".into());
let mut config = Config(HashMap::from([(
NOTIFY_WEBHOOK_SUB_SYS.into(),
HashMap::from([("primary".into(), target)]),
)]));
let context = super::plugin_instance_domain_context(PluginContractDomain::Notify);
let id = canonical_target_instance_id("builtin:webhook", TargetDomain::Notify, "primary");
let instance = super::find_target_instance(context.specs, context.route_prefix, &config, HashMap::new(), &id)
.expect("find instance")
.expect("configured target");
let expected = super::map_instance(instance).config;
assert_eq!(expected.get(WEBHOOK_AUTH_TOKEN).map(String::as_str), Some(super::REDACTED_SECRET_VALUE));
super::validate_expected_config(context, &config, &id, &expected).expect("GET config should be accepted");
config
.0
.get_mut(NOTIFY_WEBHOOK_SUB_SYS)
.expect("subsystem")
.get_mut("primary")
.expect("target")
.insert(WEBHOOK_AUTH_TOKEN.into(), "secret-two".into());
super::validate_expected_config(context, &config, &id, &expected).expect("rotation remains redacted");
config
.0
.get_mut(NOTIFY_WEBHOOK_SUB_SYS)
.expect("subsystem")
.get_mut("primary")
.expect("target")
.insert(WEBHOOK_ENDPOINT.into(), "https://example.com/b".into());
assert!(super::validate_expected_config(context, &config, &id, &expected).is_err());
config
.0
.get_mut(NOTIFY_WEBHOOK_SUB_SYS)
.expect("subsystem")
.get_mut("primary")
.expect("target")
.insert(ENABLE_KEY.into(), "off".into());
let disabled = super::find_target_instance(context.specs, context.route_prefix, &config, HashMap::new(), &id)
.expect("find disabled instance")
.expect("disabled target exists");
let disabled_expected = super::map_instance(disabled).config;
let error = super::validate_expected_config(context, &config, &id, &disabled_expected)
.expect_err("disabled target must not be enabled by editing");
assert!(error.to_string().contains("disabled targets"));
config.0.get_mut(NOTIFY_WEBHOOK_SUB_SYS).expect("subsystem").remove("primary");
assert!(super::validate_expected_config(context, &config, &id, &expected).is_err());
let legacy: PluginInstanceBody = serde_json::from_str(r#"{"key_values":[]}"#).expect("legacy input");
assert!(legacy.expected_config.is_none());
}
#[test]
#[serial_test::serial]
fn conditional_edit_rejects_inherited_environment_defaults() {
temp_env::with_vars(
[(rustfs_config::notify::ENV_NOTIFY_WEBHOOK_ENDPOINT, Some("https://example.com/hook"))],
|| {
let context = super::plugin_instance_domain_context(PluginContractDomain::Notify);
let mut config = Config::new();
let mut target = KVS::new();
target.insert(ENABLE_KEY.into(), "on".into());
config
.0
.entry(NOTIFY_WEBHOOK_SUB_SYS.into())
.or_default()
.insert("Primary".into(), target);
let id = canonical_target_instance_id("builtin:webhook", TargetDomain::Notify, "primary");
let error = super::validate_expected_config(context, &config, &id, &HashMap::new())
.expect_err("inherited environment defaults must prevent edits");
assert!(error.to_string().contains("inherits environment configuration"));
},
);
}
#[test]
fn conditional_edit_keeps_inherited_fields_in_default_layer() {
let mut original = KVS::new();
original.insert("endpoint".into(), "https://example.com/old".into());
let expected = HashMap::from([
("endpoint".into(), "https://example.com/old".into()),
("auth_token".into(), super::REDACTED_SECRET_VALUE.into()),
("queue_limit".into(), "10".into()),
]);
let mut validated = KVS::new();
validated.insert("endpoint".into(), "https://example.com/new".into());
validated.insert("auth_token".into(), "default-secret".into());
validated.insert("queue_limit".into(), "10".into());
let result = super::merge_conditional_target_update(&original, &expected, &["endpoint".into()], &validated)
.expect("merge edited endpoint");
assert_eq!(result.lookup("endpoint").as_deref(), Some("https://example.com/new"));
assert_eq!(result.lookup("auth_token"), None);
assert_eq!(result.lookup("queue_limit"), None);
for value in ["replacement-secret", ""] {
validated.insert("auth_token".into(), value.into());
let changed = super::merge_conditional_target_update(&original, &expected, &["auth_token".into()], &validated)
.expect("merge explicit secret change");
assert_eq!(changed.lookup("auth_token").as_deref(), Some(value));
assert_eq!(changed.lookup("queue_limit"), None);
}
}
#[tokio::test]
async fn partial_nats_update_validates_merged_authentication() {
use rustfs_config::{NATS_ADDRESS, NATS_PASSWORD, NATS_SUBJECT, NATS_TOKEN, NATS_USERNAME};
let context = super::plugin_instance_domain_context(PluginContractDomain::Notify);
let subsystem = rustfs_config::notify::NOTIFY_NATS_SUB_SYS;
let mut effective = KVS::new();
effective.insert(NATS_ADDRESS.into(), "nats://127.0.0.1:4222".into());
effective.insert(NATS_SUBJECT.into(), "events".into());
effective.insert(NATS_USERNAME.into(), "user".into());
effective.insert(NATS_PASSWORD.into(), "password".into());
for mut requested in [
vec![super::KeyValue {
key: NATS_TOKEN.into(),
value: "new-token".into(),
}],
vec![
super::KeyValue {
key: NATS_ADDRESS.into(),
value: "nats://127.0.0.1:4222".into(),
},
super::KeyValue {
key: NATS_SUBJECT.into(),
value: "events".into(),
},
super::KeyValue {
key: NATS_USERNAME.into(),
value: "user".into(),
},
super::KeyValue {
key: NATS_PASSWORD.into(),
value: super::REDACTED_SECRET_VALUE.into(),
},
super::KeyValue {
key: NATS_TOKEN.into(),
value: "new-token".into(),
},
],
] {
super::restore_redacted_secret_values(&mut requested, &[NATS_PASSWORD, NATS_TOKEN], Some(&effective))
.expect("restore placeholders before validation");
let candidate =
super::effective_validation_fields(context, subsystem, &effective, &requested).expect("merge request");
assert_eq!(candidate.get(NATS_PASSWORD).map(String::as_str), Some("password"));
let spec = context
.specs
.iter()
.find(|spec| spec.subsystem == subsystem)
.expect("NATS descriptor");
let error = super::build_validated_target_kvs(spec, candidate, context.default_queue_dir)
.await
.expect_err("mixed NATS authentication must fail before connection");
assert!(error.to_string().contains("only one auth method"));
}
}
#[test]
fn conditional_update_rejects_unknown_and_duplicate_keys() {
let context = super::plugin_instance_domain_context(PluginContractDomain::Notify);
let subsystem = rustfs_config::notify::NOTIFY_NATS_SUB_SYS;
let unknown = [super::KeyValue {
key: "unknown".into(),
value: "value".into(),
}];
let duplicate = [
super::KeyValue {
key: rustfs_config::NATS_SUBJECT.into(),
value: "one".into(),
},
super::KeyValue {
key: rustfs_config::NATS_SUBJECT.into(),
value: "two".into(),
},
];
for request in [&unknown[..], &duplicate[..]] {
assert!(super::effective_validation_fields(context, subsystem, &KVS::new(), request).is_err());
}
}
#[test]
fn config_revision_is_scoped_to_instance_and_generation() {
let revision = super::config_revision("builtin:webhook:notify:one", "00000000-0000-4000-8000-000000000001");
assert_eq!(revision.len(), 64);
assert_ne!(
revision,
super::config_revision("builtin:webhook:notify:two", "00000000-0000-4000-8000-000000000001")
);
assert_ne!(
revision,
super::config_revision("builtin:webhook:notify:one", "00000000-0000-4000-8000-000000000002")
);
}
#[test]
fn plugin_instance_body_rejects_unknown_fields() {
let err = serde_json::from_str::<PluginInstanceBody>(r#"{"key_values":[],"unexpected_field":true}"#)
@@ -1559,6 +1977,7 @@ mod tests {
health_state: "offline".to_string(),
health_reason: "not_loaded_in_runtime".to_string(),
runtime_present: false,
has_env_default: false,
source: TargetEndpointSource::Config,
enabled: true,
config: enabled_kvs("on"),
@@ -1585,6 +2004,7 @@ mod tests {
health_state: "offline".to_string(),
health_reason: "unreachable".to_string(),
runtime_present: true,
has_env_default: false,
source: TargetEndpointSource::Runtime,
enabled: true,
config: KVS::new(),
@@ -1611,6 +2031,7 @@ mod tests {
health_state: "disabled".to_string(),
health_reason: "disabled".to_string(),
runtime_present: false,
has_env_default: false,
source: TargetEndpointSource::Mixed,
enabled: false,
config: enabled_kvs("off"),
@@ -1654,6 +2075,7 @@ mod tests {
health_state: "offline".to_string(),
health_reason: "not_loaded_in_runtime".to_string(),
runtime_present: false,
has_env_default: false,
source: TargetEndpointSource::Config,
enabled: true,
config: enabled_kvs("on"),
+95 -3
View File
@@ -105,6 +105,7 @@ pub(crate) struct TargetInstanceReadModel {
pub health_state: String,
pub health_reason: String,
pub runtime_present: bool,
pub has_env_default: bool,
pub source: TargetEndpointSource,
pub enabled: bool,
pub config: KVS,
@@ -391,6 +392,31 @@ pub(crate) fn merge_target_endpoints(
Ok(endpoints)
}
fn runtime_target_plugin_id(specs: &[AdminTargetSpec], service: &str) -> &'static str {
target_spec_by_service(specs, service)
.map(|spec| builtin_target_manifest(spec.service).plugin_id)
.unwrap_or("custom:target")
}
pub(crate) async fn collect_instance_runtime_statuses<E>(
targets: Vec<SharedTarget<E>>,
specs: &[AdminTargetSpec],
domain: TargetDomain,
instance_id: &str,
) -> HashMap<EndpointKey, RuntimeHealthStatus>
where
E: rustfs_targets::PluginEvent,
{
let targets = targets
.into_iter()
.filter(|target| {
let id = target.id();
canonical_target_instance_id(runtime_target_plugin_id(specs, &id.name), domain, &id.id) == instance_id
})
.collect();
collect_runtime_statuses(targets).await
}
pub(crate) fn canonical_target_instance_id(plugin_id: &str, domain: TargetDomain, instance_id: &str) -> String {
format!("{plugin_id}:{}:{}", canonical_domain_label(domain), instance_id.to_lowercase())
}
@@ -444,6 +470,7 @@ pub(crate) fn collect_target_instances(
health_state: health.state,
health_reason: health.reason,
runtime_present,
has_env_default: instance.source_hints.has_env_default,
source,
enabled: instance.enabled,
config: instance.effective_config,
@@ -455,9 +482,10 @@ pub(crate) fn collect_target_instances(
continue;
}
let (plugin_id, subsystem): (String, String) = target_spec_by_service(specs, &service)
.map(|spec| (builtin_target_manifest(spec.service).plugin_id.to_string(), spec.subsystem.to_string()))
.unwrap_or_else(|| ("custom:target".to_string(), format!("{}_{}", canonical_domain_label(domain), service)));
let plugin_id = runtime_target_plugin_id(specs, &service).to_owned();
let subsystem = target_spec_by_service(specs, &service)
.map(|spec| spec.subsystem.to_string())
.unwrap_or_else(|| format!("{}_{}", canonical_domain_label(domain), service));
instances.push(TargetInstanceReadModel {
canonical_id: canonical_target_instance_id(&plugin_id, domain, &account_id),
plugin_id,
@@ -469,6 +497,7 @@ pub(crate) fn collect_target_instances(
health_state: status.state,
health_reason: status.reason,
runtime_present: true,
has_env_default: false,
source: TargetEndpointSource::Runtime,
enabled: true,
config: KVS::new(),
@@ -573,6 +602,15 @@ where
let kv_map = collect_validated_key_values(key_values, &allowed_keys, target_type, target_label)?;
let spec = target_spec(specs, target_type)
.ok_or_else(|| s3_error!(InvalidArgument, "unsupported target type: '{}'", target_type))?;
build_validated_target_kvs(spec, kv_map, default_queue_dir).await
}
/// Validate and build from keys already checked against the target descriptor.
pub(crate) async fn build_validated_target_kvs(
spec: &AdminTargetSpec,
kv_map: HashMap<String, String>,
default_queue_dir: &str,
) -> S3Result<KVS> {
timeout(Duration::from_secs(10), validate_target_request(spec, &kv_map, default_queue_dir))
.await
.map_err(|_| s3_error!(InvalidArgument, "target validation timed out after 10s"))??;
@@ -982,3 +1020,57 @@ mod webhook_request_tests {
.expect("a public HTTPS webhook should pass outbound preflight");
}
}
#[cfg(test)]
mod instance_runtime_status_tests {
use super::{admin_target_spec_from_builtin, collect_instance_runtime_statuses};
use rustfs_targets::{
SharedTarget, TargetDomain, catalog::builtin::builtin_notify_target_admin_descriptors, testkit::MockTarget,
};
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Notify;
#[tokio::test]
async fn instance_health_only_probes_matching_canonical_targets() {
let specs: Vec<_> = builtin_notify_target_admin_descriptors()
.iter()
.map(admin_target_spec_from_builtin)
.collect();
let wanted = MockTarget::new("Primary", "webhook");
let unrelated = MockTarget::new("Primary", "nats").with_health_gate(Arc::new(Notify::new()));
let unrelated_started = unrelated.health_started();
let targets: Vec<SharedTarget<rustfs_notify::Event>> = vec![Arc::new(wanted), Arc::new(unrelated)];
let statuses = tokio::time::timeout(
Duration::from_secs(1),
collect_instance_runtime_statuses(targets, &specs, TargetDomain::Notify, "builtin:webhook:notify:primary"),
)
.await
.expect("unrelated blocked health must not delay instance detail");
assert_eq!(statuses.len(), 1);
assert!(statuses.contains_key(&("Primary".into(), "webhook".into())));
assert!(
tokio::time::timeout(Duration::from_millis(20), unrelated_started.notified())
.await
.is_err()
);
}
#[tokio::test]
async fn instance_health_preserves_unknown_runtime_identity_collisions() {
let specs: Vec<_> = builtin_notify_target_admin_descriptors()
.iter()
.map(admin_target_spec_from_builtin)
.collect();
let targets: Vec<SharedTarget<rustfs_notify::Event>> = vec![
Arc::new(MockTarget::new("Primary", "custom-one")),
Arc::new(MockTarget::new("PRIMARY", "custom-two")),
Arc::new(MockTarget::new("Primary", "webhook")),
];
let statuses =
collect_instance_runtime_statuses(targets, &specs, TargetDomain::Notify, "custom:target:notify:primary").await;
assert_eq!(statuses.len(), 2);
assert!(statuses.contains_key(&("Primary".into(), "custom-one".into())));
assert!(statuses.contains_key(&("PRIMARY".into(), "custom-two".into())));
}
}
+20 -1
View File
@@ -13,7 +13,9 @@
// limitations under the License.
use crate::admin::{
handlers::target_descriptor::{EndpointKey, RuntimeHealthStatus, collect_runtime_statuses},
handlers::target_descriptor::{
AdminTargetSpec, EndpointKey, RuntimeHealthStatus, collect_instance_runtime_statuses, collect_runtime_statuses,
},
runtime_sources::{AppContext, current_app_context, current_object_store_handle_for_context},
storage_api::{
config::read_admin_config_without_migrate,
@@ -52,6 +54,23 @@ pub(crate) async fn collect_notification_runtime_statuses() -> HashMap<EndpointK
}
}
pub(crate) async fn collect_notification_instance_runtime_statuses(
specs: &[AdminTargetSpec],
instance_id: &str,
) -> HashMap<EndpointKey, RuntimeHealthStatus> {
if let Some(system) = rustfs_notify::notification_system() {
collect_instance_runtime_statuses(
system.get_target_values().await,
specs,
rustfs_targets::TargetDomain::Notify,
instance_id,
)
.await
} else {
HashMap::new()
}
}
pub(crate) async fn collect_audit_runtime_statuses() -> HashMap<EndpointKey, RuntimeHealthStatus> {
if let Some(system) = rustfs_audit::audit_system() {
collect_runtime_statuses(system.get_target_values().await).await
+3
View File
@@ -361,6 +361,8 @@ pub(crate) struct PluginInstanceDiagnosticCount {
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub(crate) struct PluginInstanceDetail {
#[serde(skip_serializing_if = "Option::is_none")]
pub config_revision: Option<String>,
#[serde(flatten)]
pub instance: PluginInstanceEntry,
#[serde(skip_serializing_if = "Vec::is_empty")]
@@ -539,6 +541,7 @@ mod tests {
#[test]
fn plugin_instance_detail_contract_serializes_diagnostics_when_present() {
let detail = PluginInstanceDetail {
config_revision: None,
instance: PluginInstanceEntry {
id: "builtin:webhook:notify:primary".to_string(),
plugin_id: "builtin:webhook".to_string(),
+8 -2
View File
@@ -826,6 +826,12 @@ pub(crate) async fn read_admin_config_without_migrate(api: Arc<ECStore>) -> Resu
ecstore_config::com::read_config_without_migrate(api).await
}
pub(crate) async fn read_admin_server_config_with_generation(
api: Arc<ECStore>,
) -> Result<(rustfs_config::server_config::Config, Option<uuid::Uuid>)> {
crate::storage::storage_api::read_server_config_with_generation(api).await
}
pub(crate) async fn read_existing_admin_server_config_no_lock(api: Arc<ECStore>) -> Result<rustfs_config::server_config::Config> {
ecstore_config::com::read_existing_server_config_no_lock(api).await
}
@@ -1001,8 +1007,8 @@ pub(crate) mod config {
pub(crate) use super::{
AdminServerConfigSaveResult, AdminServerConfigSnapshot, RUSTFS_META_BUCKET, STORAGE_CLASS_SUB_SYS, delete_admin_config,
init_admin_config_defaults, read_admin_config, read_admin_config_without_migrate, read_admin_server_config_snapshot,
read_existing_admin_server_config_no_lock, save_admin_config, save_admin_server_config_snapshot,
with_admin_server_config_read_lock,
read_admin_server_config_with_generation, read_existing_admin_server_config_no_lock, save_admin_config,
save_admin_server_config_snapshot, with_admin_server_config_read_lock,
};
#[cfg(test)]
pub(crate) use super::{
+24 -19
View File
@@ -73,14 +73,15 @@ use super::trailer_adapter::trailer_source;
use crate::app::object::{
ConcurrencyManager, ForegroundWriteAdmission, get_concurrency_manager, guard_put_object_body_read_timeout,
put_object_body_read_timeout, reject_non_wildcard_write_if_none_match, reject_oversize_single_upload,
reject_veeam_sosapi_mutation,
};
use crate::app::object_data_cache::{
ObjectDataCacheAdapter, invalidate_object_data_cache_after_complete_multipart_success,
invalidate_object_data_cache_before_mutation,
};
use crate::app::object_usecase::{
acquire_copy_bucket_lifecycle_locks, apply_quota_admission, build_put_like_object_lock_metadata, map_quota_check_outcome,
s3s_body_error_to_io, validate_existing_object_lock_for_write,
apply_quota_admission, build_put_like_object_lock_metadata, map_quota_check_outcome, s3s_body_error_to_io,
validate_existing_object_lock_for_write,
};
use crate::app::runtime_sources::{
AppContext, current_app_context, current_object_data_cache_for_context, current_object_store_handle_for_context,
@@ -542,6 +543,8 @@ impl DefaultMultipartUsecase {
bucket, key, upload_id, ..
} = req.input;
reject_veeam_sosapi_mutation(|| self.object_store(), &bucket, &key).await?;
let Some(store) = self.object_store() else {
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
};
@@ -610,6 +613,7 @@ impl DefaultMultipartUsecase {
..
} = input;
reject_veeam_sosapi_mutation(|| self.object_store(), &bucket, &key).await?;
reject_non_wildcard_write_if_none_match(if_none_match.as_ref(), &req.headers)?;
validate_table_catalog_object_mutation(&bucket, &key).await?;
@@ -1020,6 +1024,7 @@ impl DefaultMultipartUsecase {
return Err(s3_error!(InvalidStorageClass));
}
reject_veeam_sosapi_mutation(|| self.object_store(), &bucket, &key).await?;
validate_table_catalog_object_mutation(&bucket, &key).await?;
let Some(store) = self.object_store() else {
@@ -1243,6 +1248,7 @@ impl DefaultMultipartUsecase {
let part_id = parse_upload_part_number(part_number)?;
reject_veeam_sosapi_mutation(|| self.object_store(), &bucket, &key).await?;
validate_table_catalog_object_mutation(&bucket, &key).await?;
let size = resolve_upload_part_size(content_length, body.as_ref())?;
@@ -1641,6 +1647,7 @@ impl DefaultMultipartUsecase {
..
} = req.input;
reject_veeam_sosapi_mutation(|| self.object_store(), &bucket, &key).await?;
let params = parse_list_parts_params(part_number_marker, max_parts)?;
let Some(store) = self.object_store() else {
@@ -1702,6 +1709,7 @@ impl DefaultMultipartUsecase {
..
} = req.input;
reject_veeam_sosapi_mutation(|| self.object_store(), &bucket, &key).await?;
let (src_bucket, src_key, raw_src_version_id) = match copy_source {
CopySource::AccessPoint { .. } => return Err(s3_error!(NotImplemented)),
CopySource::Outpost { .. } => return Err(s3_error!(NotImplemented)),
@@ -1727,23 +1735,20 @@ impl DefaultMultipartUsecase {
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
};
let (source_bucket_lifecycle_guard, destination_bucket_lifecycle_guard_storage) =
acquire_copy_bucket_lifecycle_locks(store.as_ref(), &src_bucket, &bucket).await?;
let current_source_incarnation_id = store
.bucket_incarnation_id_from_disk(&src_bucket)
let (
source_bucket_lifecycle_guard,
current_source_incarnation_id,
destination_bucket_lifecycle_guard_storage,
current_destination_incarnation_id,
) = store
.acquire_copy_bucket_incarnation_fences(
&src_bucket,
expected_source_incarnation_id,
&bucket,
expected_destination_incarnation_id,
)
.await
.map_err(ApiError::from)?;
if expected_source_incarnation_id.is_some_and(|expected| expected != current_source_incarnation_id) {
return Err(ApiError::from(StorageError::BucketNotFound(src_bucket.clone())).into());
}
let current_destination_incarnation_id = if src_bucket == bucket {
current_source_incarnation_id
} else {
store.bucket_incarnation_id_from_disk(&bucket).await.map_err(ApiError::from)?
};
if expected_destination_incarnation_id.is_some_and(|expected| expected != current_destination_incarnation_id) {
return Err(ApiError::from(StorageError::BucketNotFound(bucket.clone())).into());
}
let destination_bucket_lifecycle_guard = destination_bucket_lifecycle_guard_storage
.as_ref()
.unwrap_or(&source_bucket_lifecycle_guard);
@@ -1762,9 +1767,9 @@ impl DefaultMultipartUsecase {
..Default::default()
};
if src_bucket != bucket {
dst_opts.add_bucket_lifecycle_lock_guard(&source_bucket_lifecycle_guard);
source_bucket_lifecycle_guard.attach_to_object_options(&mut dst_opts);
}
dst_opts.add_bucket_lifecycle_lock_guard(destination_bucket_lifecycle_guard);
destination_bucket_lifecycle_guard.attach_to_object_options(&mut dst_opts);
let mp_info = store
.get_multipart_info(&bucket, &key, &upload_id, &dst_opts)
+35 -65
View File
@@ -124,48 +124,6 @@ where
.map_err(|err| ApiError::from(copy_namespace_lock_error(bucket, &object, "write", err)).into())
}
pub(crate) async fn acquire_copy_bucket_lifecycle_lock<S>(store: &S, bucket: &str) -> S3Result<NamespaceLockGuard>
where
S: NamespaceLocking<Error = EcstoreError, NamespaceLock = rustfs_lock::NamespaceLockWrapper> + ?Sized,
{
let lock = store
.new_ns_lock(bucket, BUCKET_LIFECYCLE_LOCK_OBJECT)
.await
.map_err(ApiError::from)?;
lock.get_read_lock(get_lock_acquire_timeout()).await.map_err(|err| {
ApiError::from(copy_namespace_lock_error(
bucket,
BUCKET_LIFECYCLE_LOCK_OBJECT,
"bucket_lifecycle_read",
err,
))
.into()
})
}
pub(crate) async fn acquire_copy_bucket_lifecycle_locks<S>(
store: &S,
source_bucket: &str,
destination_bucket: &str,
) -> S3Result<(NamespaceLockGuard, Option<NamespaceLockGuard>)>
where
S: NamespaceLocking<Error = EcstoreError, NamespaceLock = rustfs_lock::NamespaceLockWrapper> + ?Sized,
{
if source_bucket == destination_bucket {
return Ok((acquire_copy_bucket_lifecycle_lock(store, source_bucket).await?, None));
}
if source_bucket < destination_bucket {
let source_guard = acquire_copy_bucket_lifecycle_lock(store, source_bucket).await?;
let destination_guard = acquire_copy_bucket_lifecycle_lock(store, destination_bucket).await?;
Ok((source_guard, Some(destination_guard)))
} else {
let destination_guard = acquire_copy_bucket_lifecycle_lock(store, destination_bucket).await?;
let source_guard = acquire_copy_bucket_lifecycle_lock(store, source_bucket).await?;
Ok((source_guard, Some(destination_guard)))
}
}
impl DefaultObjectUsecase {
pub fn execute_copy_object(
&self,
@@ -281,6 +239,7 @@ impl DefaultObjectUsecase {
// Validate both source and destination keys
validate_object_key(&src_key, "COPY (source)")?;
validate_object_key(&key, "COPY (dest)")?;
reject_veeam_sosapi_mutation(|| self.object_store(), &bucket, &key).await?;
validate_table_catalog_object_mutation(&bucket, &key).await?;
let replaces_metadata = match metadata_directive.as_ref().map(|directive| directive.as_str()) {
None | Some(MetadataDirective::COPY) => false,
@@ -383,29 +342,20 @@ impl DefaultObjectUsecase {
let Some(store) = self.object_store() else {
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
};
let (source_bucket_lifecycle_guard, destination_bucket_lifecycle_guard_storage) =
acquire_copy_bucket_lifecycle_locks(store.as_ref(), &src_bucket, &bucket).await?;
let current_source_incarnation_id = store
.bucket_incarnation_id_from_disk(&src_bucket)
let (
source_bucket_lifecycle_guard,
current_source_incarnation_id,
destination_bucket_lifecycle_guard_storage,
current_destination_incarnation_id,
) = store
.acquire_copy_bucket_incarnation_fences(
&src_bucket,
src_get_opts.expected_bucket_incarnation_id,
&bucket,
dst_opts.expected_bucket_incarnation_id,
)
.await
.map_err(ApiError::from)?;
if src_get_opts
.expected_bucket_incarnation_id
.is_some_and(|expected| expected != current_source_incarnation_id)
{
return Err(ApiError::from(StorageError::BucketNotFound(src_bucket.clone())).into());
}
let current_destination_incarnation_id = if src_bucket == bucket {
current_source_incarnation_id
} else {
store.bucket_incarnation_id_from_disk(&bucket).await.map_err(ApiError::from)?
};
if dst_opts
.expected_bucket_incarnation_id
.is_some_and(|expected| expected != current_destination_incarnation_id)
{
return Err(ApiError::from(StorageError::BucketNotFound(bucket.clone())).into());
}
let destination_bucket_lifecycle_guard = destination_bucket_lifecycle_guard_storage
.as_ref()
.unwrap_or(&source_bucket_lifecycle_guard);
@@ -422,9 +372,9 @@ impl DefaultObjectUsecase {
src_get_opts.expected_bucket_incarnation_id = Some(current_source_incarnation_id);
dst_opts.expected_bucket_incarnation_id = Some(current_destination_incarnation_id);
if src_bucket != bucket {
dst_opts.add_bucket_lifecycle_lock_guard(&source_bucket_lifecycle_guard);
source_bucket_lifecycle_guard.attach_to_object_options(&mut dst_opts);
}
dst_opts.add_bucket_lifecycle_lock_guard(destination_bucket_lifecycle_guard);
destination_bucket_lifecycle_guard.attach_to_object_options(&mut dst_opts);
// Bucket metadata uses the bucket name as its namespace-lock key. Load
// every copy-time bucket snapshot before a same-object key can collide
@@ -1081,6 +1031,26 @@ mod tests {
assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
}
#[tokio::test]
async fn copy_object_rejects_reserved_catalog_destination_before_store_lookup() {
let input = CopyObjectInput::builder()
.copy_source(CopySource::Bucket {
bucket: "source-bucket".into(),
key: "source-key".into(),
version_id: None,
})
.bucket(format!("missing-{}", Uuid::new_v4()))
.key(".rustfs-table/metadata/current.json".to_string())
.build()
.unwrap();
let error = DefaultObjectUsecase::without_context()
.execute_copy_object(build_request(input, Method::PUT))
.await
.expect_err("reserved catalog destination must fail closed without bucket metadata");
assert_eq!(error.code(), &S3ErrorCode::InvalidRequest);
assert_eq!(error.message(), Some(crate::table_catalog::RESERVED_CATALOG_OBJECT_MESSAGE));
}
#[tokio::test]
async fn execute_copy_object_rejects_invalid_storage_class() {
let input = CopyObjectInput::builder()
+1
View File
@@ -1038,6 +1038,7 @@ impl DefaultObjectUsecase {
// Validate object key
validate_object_key(&key, "DELETE")?;
reject_veeam_sosapi_mutation(|| self.object_store(), &bucket, &key).await?;
let replica = req
.headers
+1
View File
@@ -2461,6 +2461,7 @@ impl DefaultObjectUsecase {
}
let table_data_plane_publication_guards = retained_table_data_plane_publication_guards(&auth_req);
drop(auth_req);
extract_try!(reject_veeam_sosapi_mutation(|| Some(Arc::clone(&commit_context.store)), &bucket, &fpath).await);
if archive_entry_mod_time.is_some() {
opts.mod_time = archive_entry_mod_time;
}
+52
View File
@@ -14,6 +14,7 @@
//! GetObject / GetObjectAttributes read path: cold fill, resume, stream tuning.
use super::veeam_sosapi;
use super::*;
use crate::app::storage_api::object_usecase::object_cache::{
GetObjectReadCondition, GetObjectReadMetadata, evaluate_read_preconditions, has_read_preconditions,
@@ -4145,6 +4146,37 @@ impl DefaultObjectUsecase {
result
}
async fn serve_veeam_sosapi_get(
&self,
req: &S3Request<GetObjectInput>,
store: &Arc<ECStore>,
kind: veeam_sosapi::SosapiKind,
part_number: Option<usize>,
range: Option<&HTTPRangeSpec>,
) -> S3Result<S3Response<GetObjectOutput>> {
let representation =
veeam_sosapi::read_representation(store, &req.input.bucket, kind, &req.headers, part_number, range).await?;
let content_type = ContentType::from_str(veeam_sosapi::CONTENT_TYPE)
.map_err(|_| S3Error::with_message(S3ErrorCode::InternalError, "invalid SOSAPI content type"))?;
let ranged = representation.content_range.is_some();
let output = GetObjectOutput {
body: Some(StreamingBlob::from(s3s::Body::from(representation.body))),
content_length: Some(representation.content_length),
content_type: Some(content_type),
content_range: representation.content_range,
e_tag: Some(to_s3s_etag(&representation.etag_hex)),
last_modified: Some(Timestamp::from(representation.last_modified)),
accept_ranges: Some(ACCEPT_RANGES_BYTES.to_string()),
..Default::default()
};
let mut response = wrap_response_with_cors(&req.input.bucket, &req.method, &req.headers, output).await;
inject_accept_ranges_header(&mut response.headers);
if ranged {
response.status = Some(StatusCode::PARTIAL_CONTENT);
}
Ok(response)
}
async fn execute_get_object_inner(&self, mut req: S3Request<GetObjectInput>) -> S3Result<S3Response<GetObjectOutput>> {
let helper = OperationHelper::new(&req, EventName::ObjectAccessedGet, S3Operation::GetObject).suppress_event();
@@ -4223,6 +4255,26 @@ impl DefaultObjectUsecase {
}
record_get_object_s3_handler_stage_duration(GET_OBJECT_STAGE_BUCKET_VALIDATION, bucket_validation_start);
if validated.version_id.is_none()
&& let Some(kind) = veeam_sosapi::handled_kind(&validated.key)
{
let served = self
.serve_veeam_sosapi_get(&req, &store, kind, validated.part_number, validated.rs.as_ref())
.await;
match served {
Ok(response) => {
lifecycle.finish_ok();
let result = Ok(response);
let _ = helper.complete(&result);
return result;
}
Err(err) => {
lifecycle.finish_err();
return Self::complete_get_object_error(helper, err);
}
}
}
let request_context_start = stage_metrics_enabled.then(std::time::Instant::now);
let request_context = match Self::prepare_get_object_request_context(validated, &req.headers).await {
Ok(request_context) => request_context,
+46
View File
@@ -14,6 +14,7 @@
//! HeadObject path.
use super::veeam_sosapi;
use super::*;
use crate::on_demand_migration::{
BucketOdmState, HeadPolicy, OdmLookup, OdmOp, OdmOutcome, OnDemandMigrationSys, SourceClient, SourceError, SourceHead,
@@ -298,6 +299,40 @@ impl DefaultObjectUsecase {
}
}
async fn serve_veeam_sosapi_head(
&self,
req: &S3Request<HeadObjectInput>,
store: &Arc<ECStore>,
kind: veeam_sosapi::SosapiKind,
part_number: Option<usize>,
range: Option<&HTTPRangeSpec>,
) -> S3Result<S3Response<HeadObjectOutput>> {
let representation =
veeam_sosapi::read_representation(store, &req.input.bucket, kind, &req.headers, part_number, range).await?;
let content_type = ContentType::from_str(veeam_sosapi::CONTENT_TYPE)
.map_err(|_| S3Error::with_message(S3ErrorCode::InternalError, "invalid SOSAPI content type"))?;
let content_range = representation.content_range;
let output = HeadObjectOutput {
content_length: Some(representation.content_length),
content_type: Some(content_type),
accept_ranges: Some(ACCEPT_RANGES_BYTES.to_string()),
last_modified: Some(Timestamp::from(representation.last_modified)),
e_tag: Some(to_s3s_etag(&representation.etag_hex)),
..Default::default()
};
let mut response = wrap_response_with_cors(&req.input.bucket, &req.method, &req.headers, output).await;
inject_accept_ranges_header(&mut response.headers);
if let Some(content_range) = content_range {
response.headers.insert(
http::header::CONTENT_RANGE,
HeaderValue::from_str(&content_range)
.map_err(|_| S3Error::with_message(S3ErrorCode::InternalError, "invalid content range"))?,
);
response.status = Some(StatusCode::PARTIAL_CONTENT);
}
Ok(response)
}
#[instrument(level = "debug", skip(self, req))]
pub async fn execute_head_object(&self, mut req: S3Request<HeadObjectInput>) -> S3Result<S3Response<HeadObjectOutput>> {
if let Some(context) = &self.context {
@@ -336,6 +371,17 @@ impl DefaultObjectUsecase {
};
validate_bucket_exists(&store, &bucket).await?;
if version_id.is_none()
&& let Some(kind) = veeam_sosapi::handled_kind(&key)
{
let response = self
.serve_veeam_sosapi_head(&req, &store, kind, part_number, rs.as_ref())
.await?;
let result = Ok(response);
let _ = helper.complete(&result);
return result;
}
let opts: ObjectOptions = get_opts(&bucket, &key, version_id, part_number, &req.headers)
.await
.map_err(ApiError::from)?;
+3 -5
View File
@@ -134,7 +134,6 @@ use rustfs_config::MI_B;
use rustfs_filemeta::{NULL_VERSION_ID, RestoreStatusOps, parse_restore_obj_status};
use rustfs_io_core::{BytesPool, PooledBuffer};
use rustfs_io_metrics;
use rustfs_lock::NamespaceLockGuard;
use rustfs_notify::EventArgsBuilder;
use rustfs_object_capacity::capacity_manager::get_capacity_manager;
use rustfs_policy::policy::action::{Action, S3Action};
@@ -220,8 +219,8 @@ mod restore;
pub(crate) mod shared;
#[cfg(test)]
mod test_support;
mod veeam_sosapi;
pub(crate) use self::copy::*;
#[cfg(test)]
pub(crate) use self::delete::*;
pub(crate) use self::extract::*;
@@ -264,9 +263,8 @@ use tracing::{debug, error, instrument, warn};
use uuid::Uuid;
use super::storage_api::object_usecase::{
BUCKET_LIFECYCLE_LOCK_OBJECT, GetObjectReader, StorageDeletedObject, StorageObjectInfo as ObjectInfo,
StorageObjectLockDeleteOptions, StorageObjectOptions as ObjectOptions, StorageObjectToDelete as ObjectToDelete,
StoragePutObjReader as PutObjReader,
GetObjectReader, StorageDeletedObject, StorageObjectInfo as ObjectInfo, StorageObjectLockDeleteOptions,
StorageObjectOptions as ObjectOptions, StorageObjectToDelete as ObjectToDelete, StoragePutObjReader as PutObjReader,
};
use crate::app::object_data_cache::{
ColdFillCoordinateOutcome, ColdFillDiskPermitOwner, ColdFillError, ColdFillProducer, GetObjectBodyCacheLookup,
+3
View File
@@ -1272,6 +1272,7 @@ impl DefaultObjectUsecase {
format!("{RUSTFS_MAX_CONTENT_LENGTH_QUERY} is not supported for archive extraction"),
));
}
reject_veeam_sosapi_mutation(|| self.object_store(), req.input.bucket.as_str(), req.input.key.as_str()).await?;
if is_put_object_extract_requested(&req.headers) && !inbound_replication_put {
return Box::pin(self.execute_put_object_extract(req)).await;
}
@@ -1505,6 +1506,8 @@ impl DefaultObjectUsecase {
sse_customer_key_md5,
} = sse;
reject_veeam_sosapi_mutation(|| self.object_store(), bucket.as_str(), key.as_str()).await?;
// The app check preserves the existing S3 error contract; the storage
// commit path reserves the exact net logical growth under its locks.
let quota_stage_start = put_stage_metrics_enabled.then(Instant::now);
+16 -13
View File
@@ -164,6 +164,7 @@ impl DefaultObjectUsecase {
..
} = req.input.clone();
reject_veeam_sosapi_mutation(|| self.object_store(), &bucket, &object).await?;
validate_table_catalog_object_mutation(&bucket, &object).await?;
// Typed S3 errors on every RestoreObject failure (backlog#2205): a
@@ -213,8 +214,8 @@ impl DefaultObjectUsecase {
// `apply_bucket_generation_guard` deliberately tolerates a missing guard
// (only the S3 access layer installs one), so this must not hard-require
// it. Resolve the current generation instead, exactly as the copy path
// does. The fence is unaffected: the value is re-read from disk and
// compared below, before the restore is admitted.
// does. The fence is unaffected: the value is validated under the
// lifecycle fence below, before the restore is admitted.
let restore_bucket_incarnation_id = match opts.expected_bucket_incarnation_id {
Some(incarnation_id) => incarnation_id,
None => {
@@ -246,10 +247,12 @@ impl DefaultObjectUsecase {
// in-flight commit on the same object) is transient — answer 503
// SlowDown so SDK clients back off and retry instead of treating it
// as a hard failure.
let mut restore_bucket_lifecycle_guard = Some(acquire_copy_bucket_lifecycle_lock(store.as_ref(), &bucket).await?);
if store.bucket_incarnation_id_from_disk(&bucket).await.map_err(ApiError::from)? != restore_bucket_incarnation_id {
return Err(ApiError::from(StorageError::BucketNotFound(bucket.clone())).into());
}
let mut restore_bucket_lifecycle_guard = Some(
store
.acquire_bucket_incarnation_fence(&bucket, restore_bucket_incarnation_id)
.await
.map_err(ApiError::from)?,
);
let mut accept_guard = {
let guard = store
.acquire_restore_accept_guard(&bucket, &object)
@@ -299,12 +302,12 @@ impl DefaultObjectUsecase {
S3Error::with_message(S3ErrorCode::RestoreAlreadyInProgress, "Object restore is already in progress.")
})?;
restore_bucket_lifecycle_guard = Some(acquire_copy_bucket_lifecycle_lock(store.as_ref(), &bucket).await?);
if store.bucket_incarnation_id_from_disk(&bucket).await.map_err(ApiError::from)?
!= restore_bucket_incarnation_id
{
return Err(ApiError::from(StorageError::BucketNotFound(bucket.clone())).into());
}
restore_bucket_lifecycle_guard = Some(
store
.acquire_bucket_incarnation_fence(&bucket, restore_bucket_incarnation_id)
.await
.map_err(ApiError::from)?,
);
accept_guard = Some(
store
.acquire_restore_accept_guard(&bucket, &object)
@@ -398,7 +401,7 @@ impl DefaultObjectUsecase {
..Default::default()
};
if let Some(guard) = restore_bucket_lifecycle_guard.as_ref() {
restore_dst_opts.add_bucket_lifecycle_lock_guard(guard);
guard.attach_to_object_options(&mut restore_dst_opts);
}
if let Some(guard) = accept_guard.as_ref() {
guard.add_namespace_lock_fence(&mut restore_dst_opts);
+23
View File
@@ -562,6 +562,29 @@ pub(super) fn inject_accept_ranges_header(headers: &mut HeaderMap) {
headers.insert(http::header::ACCEPT_RANGES, HeaderValue::from_static(ACCEPT_RANGES_BYTES));
}
/// Reject a mutation of a Veeam SOSAPI key while the feature is enabled.
///
/// The store closure runs only after the key matches, so ordinary object
/// writes pay two string compares. Bucket existence is checked first so a
/// missing bucket stays `NoSuchBucket`.
pub(crate) async fn reject_veeam_sosapi_mutation(
store: impl FnOnce() -> Option<Arc<ECStore>>,
bucket: &str,
key: &str,
) -> S3Result<()> {
if !super::veeam_sosapi::mutation_blocked(key) {
return Ok(());
}
let Some(store) = store() else {
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
};
validate_bucket_exists(&store, bucket).await?;
Err(S3Error::with_message(
S3ErrorCode::MethodNotAllowed,
"Veeam SOSAPI system objects are read-only",
))
}
/// Derive the response-header echo pairs for an additional-checksum algorithm
/// (XXHash3/64/128, SHA-512) from the server-computed content checksum, for
/// PutObject to echo back (#1256). Returns empty for the five s3s-typed algorithms
+565
View File
@@ -0,0 +1,565 @@
// 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.
//! Veeam SOSAPI virtual objects.
//!
//! Authenticated GET and HEAD of two exact bucket-root keys return generated
//! XML. The objects are not stored: no `xl.meta`, version, or Object Lock
//! interaction. Mutations of those keys are rejected while the feature is on.
use std::collections::{HashMap, HashSet};
use std::future::Future;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, OnceLock};
use std::time::Duration;
use super::super::storage_api::s3::{S3Error, S3ErrorCode, S3Result};
use bytes::Bytes;
use md5::Digest as _;
use md5::Md5;
use moka::{future::Cache as AsyncCache, sync::Cache as SyncCache};
use time::OffsetDateTime;
use tracing::debug;
use super::super::storage_api::object_usecase::data_usage::load_bucket_usage_from_backend_cached;
use super::super::storage_api::object_usecase::{ECStore, StorageObjectInfo as ObjectInfo, check_preconditions};
use super::HTTPRangeSpec;
use super::shared::{LOG_COMPONENT_APP, LOG_SUBSYSTEM_OBJECT};
use crate::app::storage_api::admin_usecase::contract::StorageAdminApi;
use crate::app::storage_api::object_usecase::bucket::quota::{BucketQuota, QuotaType};
use http::HeaderMap;
use uuid::Uuid;
pub(super) const SYSTEM_KEY: &str = ".system-d26a9498-cb7c-4a87-a44a-8ae204f5ba6c/system.xml";
pub(super) const CAPACITY_KEY: &str = ".system-d26a9498-cb7c-4a87-a44a-8ae204f5ba6c/capacity.xml";
pub(super) const CONTENT_TYPE: &str = "application/xml";
const CAPACITY_XML_CACHE_TTL: Duration = Duration::from_secs(12);
const CAPACITY_HISTORY_IDLE: Duration = Duration::from_secs(3600);
const CAPACITY_CACHE_ENTRIES: u64 = 1024;
const CLUSTER_CACHE_ENTRIES: u64 = 16;
const MAX_BUCKET_REFRESHES: usize = 64;
const MAX_CLUSTER_REFRESHES: usize = 16;
const EVENT_USAGE_MISSING: &str = "veeam_sosapi_usage_missing";
const EVENT_QUOTA_UNAVAILABLE: &str = "veeam_sosapi_quota_unavailable";
static LOGGED_USAGE_MISSING: AtomicBool = AtomicBool::new(false);
static LOGGED_QUOTA_UNAVAILABLE: AtomicBool = AtomicBool::new(false);
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum SosapiKind {
System,
Capacity,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum SosapiReadError {
InvalidRange,
PartNumber,
}
pub(super) struct SosapiRepresentation {
pub body: Bytes,
pub content_length: i64,
pub content_range: Option<String>,
pub etag_hex: String,
pub last_modified: OffsetDateTime,
}
pub(super) struct CapacityInputs {
pub used_bucket: u64,
pub hard_quota: Option<u64>,
pub capacity_cluster: u64,
pub free_cluster: u64,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) struct CapacityFigures {
pub capacity: u64,
pub used: u64,
pub available: u64,
}
#[derive(Clone, Debug, Hash, PartialEq, Eq)]
struct CapacityCacheKey {
store_id: Uuid,
bucket: String,
incarnation: Uuid,
}
#[derive(Clone)]
struct CachedXml {
bytes: Bytes,
etag_hex: String,
modified_at: OffsetDateTime,
}
// Bucket refreshes may wait for a cluster permit; cluster probes never acquire a bucket permit.
struct CapacityCaches {
current: AsyncCache<CapacityCacheKey, CachedXml>,
history: SyncCache<CapacityCacheKey, CachedXml>,
refresh_permits: tokio::sync::Semaphore,
}
impl CapacityCaches {
fn new() -> Self {
Self {
current: AsyncCache::builder()
.max_capacity(CAPACITY_CACHE_ENTRIES)
.time_to_live(CAPACITY_XML_CACHE_TTL)
.build(),
history: SyncCache::builder()
.max_capacity(CAPACITY_CACHE_ENTRIES)
.time_to_idle(CAPACITY_HISTORY_IDLE)
.build(),
refresh_permits: tokio::sync::Semaphore::new(MAX_BUCKET_REFRESHES),
}
}
}
struct StableXml {
bytes: Bytes,
etag_hex: String,
modified_at: OffsetDateTime,
}
pub(super) fn match_sosapi_key(key: &str) -> Option<SosapiKind> {
match key {
SYSTEM_KEY => Some(SosapiKind::System),
CAPACITY_KEY => Some(SosapiKind::Capacity),
_ => None,
}
}
pub(super) fn veeam_sosapi_enabled() -> bool {
rustfs_utils::get_env_bool(rustfs_config::ENV_VEEAM_SOSAPI_ENABLE, rustfs_config::DEFAULT_VEEAM_SOSAPI_ENABLE)
}
/// SOSAPI key whose virtual read should run. `None` when the key is ordinary
/// or the operator turned the feature off.
pub(super) fn handled_kind(key: &str) -> Option<SosapiKind> {
let kind = match_sosapi_key(key)?;
veeam_sosapi_enabled().then_some(kind)
}
/// True only for an exact SOSAPI key while the feature is enabled.
///
/// Callers use this on the mutation hot path: a miss is two string compares
/// and does not read the object store.
pub(super) fn mutation_blocked(key: &str) -> bool {
match_sosapi_key(key).is_some() && veeam_sosapi_enabled()
}
pub(super) fn capacity_figures(inputs: CapacityInputs) -> CapacityFigures {
let used_bucket = inputs.used_bucket;
let capacity = inputs
.hard_quota
.filter(|quota| *quota > 0)
.unwrap_or(inputs.capacity_cluster);
let used = used_bucket.min(capacity);
let available = capacity.saturating_sub(used).min(inputs.free_cluster);
CapacityFigures {
capacity,
used,
available,
}
}
/// Full generated document. Range and part-number checks happen after read
/// preconditions so a failing `If-Match` stays 412.
async fn generated(store: &Arc<ECStore>, bucket: &str, kind: SosapiKind) -> S3Result<(Bytes, String, OffsetDateTime)> {
match kind {
SosapiKind::System => {
let stable = system_xml_stable();
Ok((stable.bytes.clone(), stable.etag_hex.clone(), stable.modified_at))
}
SosapiKind::Capacity => cached_or_build_capacity(store, bucket).await,
}
}
pub(super) async fn read_representation(
store: &Arc<ECStore>,
bucket: &str,
kind: SosapiKind,
headers: &HeaderMap,
part_number: Option<usize>,
range: Option<&HTTPRangeSpec>,
) -> S3Result<SosapiRepresentation> {
let (full, etag_hex, last_modified) = generated(store, bucket, kind).await?;
check_preconditions(
headers,
&ObjectInfo {
etag: Some(etag_hex.clone()),
mod_time: Some(last_modified),
..Default::default()
},
)?;
representation_from(full, etag_hex, last_modified, part_number, range).map_err(|err| match err {
SosapiReadError::InvalidRange => {
S3Error::with_message(S3ErrorCode::InvalidRange, "The requested range is not satisfiable")
}
SosapiReadError::PartNumber => {
S3Error::with_message(S3ErrorCode::InvalidArgument, "partNumber is not supported for Veeam SOSAPI objects")
}
})
}
fn representation_from(
full: Bytes,
etag_hex: String,
last_modified: OffsetDateTime,
part_number: Option<usize>,
range: Option<&HTTPRangeSpec>,
) -> Result<SosapiRepresentation, SosapiReadError> {
if part_number.is_some() {
return Err(SosapiReadError::PartNumber);
}
apply_range(full, etag_hex, last_modified, range)
}
fn apply_range(
full: Bytes,
etag_hex: String,
last_modified: OffsetDateTime,
range: Option<&HTTPRangeSpec>,
) -> Result<SosapiRepresentation, SosapiReadError> {
let total = i64::try_from(full.len()).unwrap_or(i64::MAX);
let Some(range) = range else {
return Ok(SosapiRepresentation {
content_length: total,
content_range: None,
body: full,
etag_hex,
last_modified,
});
};
let (start, length) = range.get_offset_length(total).map_err(|_| SosapiReadError::InvalidRange)?;
let length_usize = usize::try_from(length).map_err(|_| SosapiReadError::InvalidRange)?;
let end = start.checked_add(length_usize).ok_or(SosapiReadError::InvalidRange)?;
if end > full.len() {
return Err(SosapiReadError::InvalidRange);
}
let start_i64 = i64::try_from(start).map_err(|_| SosapiReadError::InvalidRange)?;
let end_inclusive = start_i64 + length - 1;
Ok(SosapiRepresentation {
body: full.slice(start..end),
content_length: length,
content_range: Some(format!("bytes {start_i64}-{end_inclusive}/{total}")),
etag_hex,
last_modified,
})
}
fn system_xml_stable() -> &'static StableXml {
static BODY: OnceLock<StableXml> = OnceLock::new();
BODY.get_or_init(|| {
let bytes = Bytes::from(system_xml(env!("CARGO_PKG_VERSION")));
let etag_hex = md5_hex(&bytes);
StableXml {
bytes,
etag_hex,
modified_at: OffsetDateTime::from_unix_timestamp(crate::version::build::BUILD_TIMESTAMP)
.unwrap_or(OffsetDateTime::UNIX_EPOCH),
}
})
}
fn system_xml(version: &str) -> String {
let version = xml_escape_text(version);
format!(
"<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n<SystemInfo>\n <ProtocolVersion>\"1.0\"</ProtocolVersion>\n <ModelName>\"RustFS {version}\"</ModelName>\n <ProtocolCapabilities>\n <CapacityInfo>true</CapacityInfo>\n <UploadSessions>false</UploadSessions>\n <IAMSTS>false</IAMSTS>\n </ProtocolCapabilities>\n</SystemInfo>\n"
)
}
fn capacity_xml(figures: &CapacityFigures) -> String {
format!(
"<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n<CapacityInfo>\n <Capacity>{}</Capacity>\n <Available>{}</Available>\n <Used>{}</Used>\n</CapacityInfo>\n",
figures.capacity, figures.available, figures.used,
)
}
/// Escape XML text. Quotes stay as ASCII quotes: Veeam's parser requires that
/// spelling inside `ProtocolVersion` and `ModelName`.
fn xml_escape_text(value: &str) -> String {
let mut escaped = String::with_capacity(value.len());
for ch in value.chars() {
match ch {
'&' => escaped.push_str("&amp;"),
'<' => escaped.push_str("&lt;"),
'>' => escaped.push_str("&gt;"),
_ => escaped.push(ch),
}
}
escaped
}
fn md5_hex(bytes: &[u8]) -> String {
const HEX: &[u8; 16] = b"0123456789abcdef";
let digest = Md5::digest(bytes);
let mut out = String::with_capacity(32);
for byte in digest {
out.push(char::from(HEX[usize::from(byte >> 4)]));
out.push(char::from(HEX[usize::from(byte & 0x0f)]));
}
out
}
fn capacity_caches() -> &'static CapacityCaches {
static CACHES: OnceLock<CapacityCaches> = OnceLock::new();
CACHES.get_or_init(CapacityCaches::new)
}
async fn cached_or_build_capacity(store: &Arc<ECStore>, bucket: &str) -> S3Result<(Bytes, String, OffsetDateTime)> {
let incarnation = store
.bucket_incarnation_id_from_disk(bucket)
.await
.map_err(|_| capacity_unavailable())?;
let key = CapacityCacheKey {
store_id: store.id,
bucket: bucket.to_owned(),
incarnation,
};
cached_or_load_capacity(capacity_caches(), &key, || async {
let inputs = load_capacity_inputs(store, bucket, incarnation).await?;
if store
.bucket_incarnation_id_from_disk(bucket)
.await
.map_err(|_| capacity_unavailable())?
!= incarnation
{
return Err(capacity_unavailable());
}
Ok(capacity_figures(inputs))
})
.await
}
async fn cached_or_load_capacity<F, Fut>(
caches: &CapacityCaches,
key: &CapacityCacheKey,
load: F,
) -> S3Result<(Bytes, String, OffsetDateTime)>
where
F: FnOnce() -> Fut,
Fut: Future<Output = S3Result<CapacityFigures>>,
{
let value = caches
.current
.try_get_with_by_ref(key, async {
let _permit = caches.refresh_permits.acquire().await.map_err(|_| capacity_unavailable())?;
let bytes = Bytes::from(capacity_xml(&load().await?));
let etag_hex = md5_hex(&bytes);
let modified_at = caches
.history
.get(key)
.filter(|entry| entry.bytes == bytes)
.map_or_else(OffsetDateTime::now_utc, |entry| entry.modified_at);
let value = CachedXml {
bytes,
etag_hex,
modified_at,
};
caches.history.insert(key.clone(), value.clone());
Ok::<_, S3Error>(value)
})
.await
.map_err(|_| capacity_unavailable())?;
Ok((value.bytes, value.etag_hex, value.modified_at))
}
fn capacity_unavailable() -> S3Error {
S3Error::with_message(S3ErrorCode::ServiceUnavailable, "Veeam SOSAPI capacity is temporarily unavailable")
}
async fn load_capacity_inputs(store: &Arc<ECStore>, bucket: &str, incarnation: Uuid) -> S3Result<CapacityInputs> {
let (quota, quota_incarnation, _) = store.bucket_quota_and_incarnation_from_disk(bucket).await.map_err(|_| {
log_quota_unavailable_once(bucket);
capacity_unavailable()
})?;
if quota_incarnation != incarnation {
return Err(capacity_unavailable());
}
let used_bucket = load_bucket_usage_from_backend_cached(Arc::clone(store), bucket, incarnation)
.await
.ok()
.flatten()
.ok_or_else(|| {
log_usage_missing_once(bucket);
capacity_unavailable()
})?;
let (capacity_cluster, free_cluster) = cached_cluster_capacity(store).await?;
Ok(CapacityInputs {
used_bucket,
hard_quota: quota.as_ref().and_then(hard_quota_bytes),
capacity_cluster,
free_cluster,
})
}
fn hard_quota_bytes(quota: &BucketQuota) -> Option<u64> {
match quota.quota_type {
QuotaType::Hard => quota.quota.filter(|size| *size > 0),
}
}
struct ClusterCapacityCache {
current: AsyncCache<Uuid, (u64, u64)>,
refresh_permits: tokio::sync::Semaphore,
}
impl ClusterCapacityCache {
fn new() -> Self {
Self {
current: AsyncCache::builder()
.max_capacity(CLUSTER_CACHE_ENTRIES)
.time_to_live(CAPACITY_XML_CACHE_TTL)
.build(),
refresh_permits: tokio::sync::Semaphore::new(MAX_CLUSTER_REFRESHES),
}
}
}
fn cluster_capacity_cache() -> &'static ClusterCapacityCache {
static CACHE: OnceLock<ClusterCapacityCache> = OnceLock::new();
CACHE.get_or_init(ClusterCapacityCache::new)
}
async fn cached_cluster_capacity(store: &Arc<ECStore>) -> S3Result<(u64, u64)> {
cached_cluster_capacity_with(cluster_capacity_cache(), store.id, || async {
let info = StorageAdminApi::storage_info(store.as_ref()).await;
if !cluster_capacity_observations_are_live(&info) {
return Err(capacity_unavailable());
}
conservative_cluster_capacity(&info).ok_or_else(capacity_unavailable)
})
.await
}
fn cluster_capacity_observations_are_live(info: &rustfs_madmin::StorageInfo) -> bool {
!info.disks.is_empty()
&& !info.observations.is_empty()
&& info
.observations
.iter()
.all(|observation| !observation.cached && observation.status == rustfs_madmin::StorageInfoProbeStatus::Succeeded)
&& info.disks.iter().all(|disk| {
(disk.state.eq_ignore_ascii_case("ok") || disk.state.eq_ignore_ascii_case("online"))
&& disk.capacity_observation_source.as_deref() == Some("live_probe")
})
}
async fn cached_cluster_capacity_with<F, Fut>(cache: &ClusterCapacityCache, store_id: Uuid, load: F) -> S3Result<(u64, u64)>
where
F: FnOnce() -> Fut,
Fut: Future<Output = S3Result<(u64, u64)>>,
{
cache
.current
.try_get_with(store_id, async {
let _permit = cache.refresh_permits.acquire().await.map_err(|_| capacity_unavailable())?;
load().await
})
.await
.map_err(|_| capacity_unavailable())
}
/// A set can consume only the space available on each of its shard disks.
/// This remains an estimate when multiple shard paths share a backing device.
fn conservative_cluster_capacity(info: &rustfs_madmin::StorageInfo) -> Option<(u64, u64)> {
let backend = &info.backend;
let pools = backend.drives_per_set.len();
if pools == 0
|| backend.total_sets.len() != pools
|| backend.standard_sc_data.len() != pools
|| backend.standard_sc_parities.len() != pools
{
return None;
}
let expected = backend
.drives_per_set
.iter()
.zip(&backend.total_sets)
.try_fold(0usize, |sum, (&width, &sets)| sum.checked_add(width.checked_mul(sets)?))?;
if expected != info.disks.len() {
return None;
}
let mut seen = HashSet::with_capacity(expected);
let mut sets: HashMap<(usize, usize), (u64, u64, usize)> = HashMap::new();
for disk in &info.disks {
let pool = usize::try_from(disk.pool_index).ok()?;
let set = usize::try_from(disk.set_index).ok()?;
let index = usize::try_from(disk.disk_index).ok()?;
let width = *backend.drives_per_set.get(pool)?;
if width == 0 || set >= *backend.total_sets.get(pool)? || index >= width || !seen.insert((pool, set, index)) {
return None;
}
let entry = sets.entry((pool, set)).or_insert((u64::MAX, u64::MAX, 0));
entry.0 = entry.0.min(disk.total_space);
entry.1 = entry.1.min(disk.available_space);
entry.2 = entry.2.checked_add(1)?;
}
let mut total = 0u64;
let mut free = 0u64;
for ((pool, _), (min_total, min_free, count)) in sets {
let width = backend.drives_per_set[pool];
let data = backend.standard_sc_data[pool];
if count != width
|| data == 0
|| data.checked_add(backend.standard_sc_parities[pool]) != Some(width)
|| min_total == 0
|| min_free > min_total
{
return None;
}
let data = u64::try_from(data).ok()?;
total = total.checked_add(min_total.checked_mul(data)?)?;
free = free.checked_add(min_free.checked_mul(data)?)?;
}
Some((total, free))
}
fn log_usage_missing_once(bucket: &str) {
if LOGGED_USAGE_MISSING
.compare_exchange(false, true, Ordering::Relaxed, Ordering::Relaxed)
.is_ok()
{
debug!(
event = EVENT_USAGE_MISSING,
component = LOG_COMPONENT_APP,
subsystem = LOG_SUBSYSTEM_OBJECT,
state = "snapshot_missing",
bucket = %bucket,
"Veeam SOSAPI capacity is unavailable because the data-usage snapshot is missing"
);
}
}
fn log_quota_unavailable_once(bucket: &str) {
if LOGGED_QUOTA_UNAVAILABLE
.compare_exchange(false, true, Ordering::Relaxed, Ordering::Relaxed)
.is_ok()
{
debug!(
event = EVENT_QUOTA_UNAVAILABLE,
component = LOG_COMPONENT_APP,
subsystem = LOG_SUBSYSTEM_OBJECT,
state = "quota_unavailable",
bucket = %bucket,
"Veeam SOSAPI capacity is unavailable because the bucket quota could not be read"
);
}
}
#[cfg(test)]
mod tests;
File diff suppressed because it is too large Load Diff
+9 -2
View File
@@ -107,6 +107,14 @@ pub(crate) mod data_usage {
crate::storage::storage_api::ecstore_data_usage::load_admin_data_usage_from_backend_cached(store).await
}
pub(crate) async fn load_bucket_usage_from_backend_cached(
store: Arc<crate::storage::storage_api::ECStore>,
bucket: &str,
incarnation: uuid::Uuid,
) -> Result<Option<u64>, crate::storage::storage_api::StorageError> {
crate::storage::storage_api::ecstore_data_usage::load_bucket_usage_from_backend_cached(store, bucket, incarnation).await
}
#[cfg(test)]
pub(crate) async fn load_data_usage_from_backend_cached(
store: Arc<crate::storage::storage_api::ECStore>,
@@ -679,8 +687,8 @@ pub(crate) mod bucket {
pub(crate) type QuotaChecker = crate::storage::storage_api::ecstore_bucket::quota::checker::QuotaChecker;
}
#[cfg(test)]
pub(crate) type BucketQuota = crate::storage::storage_api::ecstore_bucket::quota::BucketQuota;
pub(crate) type QuotaType = crate::storage::storage_api::ecstore_bucket::quota::QuotaType;
pub(crate) type QuotaOperation = crate::storage::storage_api::ecstore_bucket::quota::QuotaOperation;
pub(crate) type QuotaCheckResult = crate::storage::storage_api::ecstore_bucket::quota::QuotaCheckResult;
pub(crate) type QuotaError = crate::storage::storage_api::ecstore_bucket::quota::QuotaError;
@@ -1203,7 +1211,6 @@ pub(crate) mod bucket_usecase {
}
pub(crate) mod object_usecase {
pub(crate) use super::storage_contracts::BUCKET_LIFECYCLE_LOCK_OBJECT;
pub(crate) use super::storage_contracts::object::DeleteObjectCondition;
pub(crate) use crate::storage::storage_api::ecstore_object::{WriteCommitGuard, WriteCompletion};
+14 -2
View File
@@ -528,6 +528,7 @@ struct ConfigInfoJson {
kms_backend: String,
buffer_profile: String,
workload_profile: Option<WorkloadProfileJson>,
veeam_sosapi_enable: bool,
}
/// Workload profile information for JSON output
@@ -572,6 +573,10 @@ fn collect_config_info_json() -> ConfigInfoJson {
kms_backend: snapshot.kms_backend.clone(),
buffer_profile: snapshot.buffer_profile.clone(),
workload_profile,
veeam_sosapi_enable: rustfs_utils::get_env_bool(
rustfs_config::ENV_VEEAM_SOSAPI_ENABLE,
rustfs_config::DEFAULT_VEEAM_SOSAPI_ENABLE,
),
}
}
@@ -889,6 +894,8 @@ fn format_protocol_config_info() -> String {
rustfs_utils::get_env_u64(rustfs_config::ENV_WEBDAV_REQUEST_TIMEOUT, DEFAULT_WEBDAV_REQUEST_TIMEOUT_SECS);
let webdav_max_connections =
rustfs_utils::get_env_usize(rustfs_config::ENV_WEBDAV_MAX_CONNECTIONS, DEFAULT_WEBDAV_MAX_CONNECTIONS);
let veeam_sosapi_enable =
rustfs_utils::get_env_bool(rustfs_config::ENV_VEEAM_SOSAPI_ENABLE, rustfs_config::DEFAULT_VEEAM_SOSAPI_ENABLE);
format!(
"| FTPS | --- |\n\
@@ -909,7 +916,9 @@ fn format_protocol_config_info() -> String {
| WebDAV > CA File (`{}`) | {} |\n\
| WebDAV > Max Body Size (`{}`) | {} bytes |\n\
| WebDAV > Request Timeout (`{}`) | {} seconds |\n\
| WebDAV > Max Connections (`{}`) | {} |",
| WebDAV > Max Connections (`{}`) | {} |\n\
| Veeam SOSAPI | --- |\n\
| Veeam SOSAPI > Enabled (`{}`) | {} |",
if cfg!(feature = "ftps") { "enabled" } else { "disabled" },
rustfs_config::ENV_FTPS_ENABLE,
ftps_enable,
@@ -941,7 +950,9 @@ fn format_protocol_config_info() -> String {
rustfs_config::ENV_WEBDAV_REQUEST_TIMEOUT,
webdav_request_timeout,
rustfs_config::ENV_WEBDAV_MAX_CONNECTIONS,
webdav_max_connections
webdav_max_connections,
rustfs_config::ENV_VEEAM_SOSAPI_ENABLE,
veeam_sosapi_enable
)
}
@@ -1137,5 +1148,6 @@ mod tests {
assert!(output.contains("| WebDAV > Enabled (`RUSTFS_WEBDAV_ENABLE`) |"));
assert!(output.contains("| WebDAV > Address (`RUSTFS_WEBDAV_ADDRESS`) |"));
assert!(output.contains("| WebDAV > Max Body Size (`RUSTFS_WEBDAV_MAX_BODY_SIZE`) |"));
assert!(output.contains("| Veeam SOSAPI > Enabled (`RUSTFS_VEEAM_SOSAPI_ENABLE`) |"));
}
}
+16
View File
@@ -393,6 +393,10 @@ pub(crate) fn parse_object_version_id(version_id: Option<String>) -> S3Result<Op
}
}
async fn reject_veeam_sosapi_mutation(fs: &FS, bucket: &str, key: &str) -> S3Result<()> {
s3_api::reject_veeam_sosapi_mutation(|| fs.server_ctx.object_store(), bucket, key).await
}
async fn validate_table_catalog_object_mutation(bucket: &str, key: &str) -> S3Result<()> {
table_catalog::validate_bucket_object_mutation(bucket, key)
.await
@@ -663,6 +667,7 @@ impl S3 for FS {
..
} = req.input.clone();
reject_veeam_sosapi_mutation(self, &bucket, &object).await?;
validate_table_catalog_object_mutation(&bucket, &object).await?;
let Some(store) = self.server_ctx.object_store() else {
@@ -1004,6 +1009,8 @@ impl S3 for FS {
bucket, key, version_id, ..
} = req.input.clone();
reject_veeam_sosapi_mutation(self, &bucket, &key).await?;
let Some(store) = self.server_ctx.object_store() else {
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
};
@@ -1113,6 +1120,8 @@ impl S3 for FS {
bucket, key, version_id, ..
} = req.input.clone();
reject_veeam_sosapi_mutation(self, &bucket, &key).await?;
let Some(store) = self.server_ctx.object_store() else {
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
};
@@ -1156,6 +1165,8 @@ impl S3 for FS {
let bucket = req.input.bucket.as_str();
let object = req.input.key.as_str();
reject_veeam_sosapi_mutation(self, bucket, object).await?;
let Some(store) = self.server_ctx.object_store() else {
error!(
component = LOG_COMPONENT_STORAGE,
@@ -1564,6 +1575,8 @@ impl S3 for FS {
let key = &req.input.key;
let version_id = req.input.version_id.clone();
reject_veeam_sosapi_mutation(self, bucket, key).await?;
let Some(store) = self.server_ctx.object_store() else {
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
};
@@ -1604,6 +1617,7 @@ impl S3 for FS {
..
} = req.input.clone();
reject_veeam_sosapi_mutation(self, &bucket, &key).await?;
validate_table_catalog_object_mutation(&bucket, &key).await?;
let Some(store) = self.server_ctx.object_store() else {
@@ -1806,6 +1820,7 @@ impl S3 for FS {
..
} = req.input.clone();
reject_veeam_sosapi_mutation(self, &bucket, &key).await?;
validate_table_catalog_object_mutation(&bucket, &key).await?;
let Some(store) = self.server_ctx.object_store() else {
@@ -1923,6 +1938,7 @@ impl S3 for FS {
..
} = req.input.clone();
reject_veeam_sosapi_mutation(self, &bucket, &object).await?;
validate_table_catalog_object_mutation(&bucket, &object).await?;
crate::storage::s3_api::tagging::validate_object_tag_set(&tagging.tag_set)?;
+1 -1
View File
@@ -17,7 +17,7 @@
//! This file intentionally starts as skeleton-only. Behavior remains in place
//! until each helper is moved with dedicated small refactor steps.
pub(crate) use crate::app::object::s3_response_version_id;
pub(crate) use crate::app::object::{reject_veeam_sosapi_mutation, s3_response_version_id};
use crate::app::{
bucket_usecase::DefaultBucketUsecase, multipart_usecase::DefaultMultipartUsecase, object_usecase::DefaultObjectUsecase,
};
+9 -3
View File
@@ -477,9 +477,9 @@ pub(crate) mod ecstore_config {
pub(crate) mod ecstore_data_usage {
pub(crate) use rustfs_ecstore::api::data_usage::{
apply_bucket_usage_memory_overlay, get_bucket_usage_memory, init_compression_total_memory_from_backend,
load_admin_data_usage_from_backend_cached, load_data_usage_from_backend, lookup_degraded_bucket_usage_baseline,
quota_object_size, record_bucket_delete_marker_memory, record_bucket_object_delete_memory,
record_bucket_object_version_write_memory, record_bucket_object_write_memory,
load_admin_data_usage_from_backend_cached, load_bucket_usage_from_backend_cached, load_data_usage_from_backend,
lookup_degraded_bucket_usage_baseline, quota_object_size, record_bucket_delete_marker_memory,
record_bucket_object_delete_memory, record_bucket_object_version_write_memory, record_bucket_object_write_memory,
record_bucket_object_write_unknown_previous_memory, store_compression_total_in_backend,
};
// Test-only observables for the rustfs/backlog#1306 revert detector.
@@ -1349,6 +1349,12 @@ pub(crate) async fn read_config_no_lock(api: Arc<ECStore>, file: &str) -> Result
ecstore_config::com::read_config_no_lock(api, file).await
}
pub(crate) async fn read_server_config_with_generation(
api: Arc<ECStore>,
) -> Result<(rustfs_config::server_config::Config, Option<uuid::Uuid>)> {
ecstore_config::com::read_server_config_with_generation(api).await
}
pub(crate) async fn read_existing_server_config_no_lock(api: Arc<ECStore>) -> Result<rustfs_config::server_config::Config> {
ecstore_config::com::read_existing_server_config_no_lock(api).await
}
+18 -1
View File
@@ -21682,7 +21682,6 @@ fn object_mutation_entrypoints_call_reserved_prefix_guard() {
for expected in [
"validate_object_key(&key, request_method_name)?;\n validate_table_catalog_object_mutation(&bucket, &key).await?;",
"validate_object_key(&key, \"COPY (dest)\")?;\n validate_table_catalog_object_mutation(&bucket, &key).await?;",
"if let Err(err) = validate_table_catalog_object_mutation(&bucket, &obj_id.key).await",
"validate_table_catalog_object_mutation(&bucket, &object).await?;",
"validate_object_key(&key, \"PUT\")?;\n validate_table_catalog_object_mutation(&bucket, &key).await?;",
@@ -21690,6 +21689,24 @@ fn object_mutation_entrypoints_call_reserved_prefix_guard() {
] {
assert!(source.contains(expected), "missing object mutation guard: {expected}");
}
let copy_object = include_str!("../app/object/copy.rs")
.split_once("async fn execute_copy_object_body(")
.map(|(_, body)| body.split_once("\n#[cfg(test)]").map_or(body, |(entrypoint, _)| entrypoint))
.expect("copy object entrypoint should remain in app/object/copy.rs");
let copy_key = copy_object
.find("validate_object_key(&key, \"COPY (dest)\")?;")
.expect("copy destination key validation");
let copy_guard = copy_object
.find("validate_table_catalog_object_mutation(&bucket, &key).await?;")
.expect("copy destination catalog guard");
let copy_store = copy_object
.find("let Some(store) = self.object_store()")
.expect("copy object store resolution");
assert!(
copy_key < copy_guard && copy_guard < copy_store,
"copy must guard the destination before store access"
);
assert!(
delete_object.contains("validate_object_key(&key, \"DELETE\")?;")
&& delete_object.contains("validate_table_catalog_object_mutation(&bucket, &key).await?;"),
+1 -1
View File
@@ -65,7 +65,7 @@ cd "$(dirname "$0")/.."
# other reductions. The merged tree measures 209 import files and 1586
# invocation lines, tightening main's reviewed 1592-line baseline by six.
S3S_IMPORT_FILES_BASELINE=209
S3_ERROR_LINES_BASELINE=1571
S3_ERROR_LINES_BASELINE=1570
# ecstore-scoped ratchet (rustfs/backlog#1842): the storage engine must not
# know S3 wire/DTO types (ARCHITECTURE.md invariant 4). The S3-*consuming*
# client was extracted to crates/s3-client, where s3s usage is legitimate;
+3 -1
View File
@@ -85,7 +85,9 @@ def histogram_point(point):
require(all(a < b for a, b in zip(bounds, bounds[1:])), "unordered histogram boundaries")
require(len(buckets) == len(bounds) + 1, "missing or inconsistent histogram buckets")
buckets = [uint(value, "bucket count") for value in buckets]
count = uint(point.get("count", 0), "histogram count")
# Proto3 JSON conversion can omit a default-zero count. Without presence
# information, reject it instead of treating a missing observation as zero.
count = uint(point.get("count"), "histogram count")
require(sum(buckets) == count, "histogram count does not match buckets")
return {**times, "kind": "histogram", "count": count, "bounds": bounds, "buckets": buckets}
+40 -9
View File
@@ -224,6 +224,22 @@ def require_complete_abba_matrix(manifest: dict[str, Any], report: dict[str, Any
require(not missing, f"missing ABBA matrix comparison: {missing[0] if missing else ''}")
def has_complete_abba_matrix(manifest: dict[str, Any], report: dict[str, Any], comparisons: list[dict[str, Any]]) -> bool:
rounds = manifest.get("rounds")
if type(rounds) is not int or not 3 <= rounds <= 10:
return False
expected = {(scenario, comparison, round_id) for scenario in SCENARIOS
for comparison in ("build", "background") for round_id in range(1, rounds + 1)}
observed = []
for row in comparisons:
if not isinstance(row, dict) or not isinstance(row.get("scenario"), str) or not isinstance(row.get("comparison"), str) \
or type(row.get("round")) is not int:
return False
observed.append((row["scenario"], row["comparison"], row["round"]))
return (len(observed) == len(expected) and set(observed) == expected
and report.get("cells") == len(expected) * len(LEGS))
def summarize_abba(abba_dir: Path) -> dict[str, Any]:
manifest_path = abba_dir / "manifest.json"
report_path = abba_dir / "report.json"
@@ -238,6 +254,7 @@ def summarize_abba(abba_dir: Path) -> dict[str, Any]:
require(report_state not in PASS_STATES, "passing report requires comparisons")
comparisons = []
require(isinstance(comparisons, list), "report.comparisons must be a list")
matrix_complete = has_complete_abba_matrix(manifest, report, comparisons)
counts = Counter()
p99_regressions: list[Decimal] = []
@@ -250,6 +267,8 @@ def summarize_abba(abba_dir: Path) -> dict[str, Any]:
error_rate_values: list[Decimal | None] = []
pressure_samples = 0
pressure_high_samples = 0
pressure_samples_complete = matrix_complete
pressure_high_samples_complete = matrix_complete
attempt_cost_samples = 0
start_p95_values: list[Decimal | None] = []
duplicate_task_values: list[Decimal | None] = []
@@ -296,13 +315,22 @@ def summarize_abba(abba_dir: Path) -> dict[str, Any]:
lock_hold_values.append(maybe_number(value, "heal_lock_hold_p95_ms"))
w10_w11 = comparison.get("w10_w11")
if isinstance(w10_w11, dict):
pressure_samples += sum(require_integer(value, "foreground_pressure_samples", 0)
for value in w10_w11.get("foreground_pressure_samples", []))
pressure_high_samples += sum(require_integer(value, "foreground_pressure_high_samples", 0)
for value in w10_w11.get("foreground_pressure_high_samples", []))
samples = w10_w11.get("foreground_pressure_samples")
if isinstance(samples, list) and len(samples) == len(LEGS) and all(value is not None for value in samples):
pressure_samples += sum(require_integer(value, "foreground_pressure_samples", 0) for value in samples)
else:
pressure_samples_complete = False
high_samples = w10_w11.get("foreground_pressure_high_samples")
if isinstance(high_samples, list) and len(high_samples) == len(LEGS) and all(value is not None for value in high_samples):
pressure_high_samples += sum(require_integer(value, "foreground_pressure_high_samples", 0) for value in high_samples)
else:
pressure_high_samples_complete = False
for value in w10_w11.get("attempt_cost_per_healed_object", []):
if value is not None:
attempt_cost_samples += 1
else:
pressure_samples_complete = False
pressure_high_samples_complete = False
w10 = comparison.get("w10")
if isinstance(w10, dict) and comparison.get("scenario") == "running-heal" and comparison.get("comparison") == "build":
w10_rows.append({
@@ -316,8 +344,7 @@ def summarize_abba(abba_dir: Path) -> dict[str, Any]:
})
if "throughput_ops" in comparison:
throughput_values.append(maybe_number(comparison.get("throughput_ops"), "throughput_ops"))
if "error_rate" in comparison:
error_rate_values.append(maybe_number(comparison.get("error_rate"), "error_rate"))
error_rate_values.append(maybe_number(comparison.get("error_rate"), "error_rate"))
w11 = comparison.get("w11")
if isinstance(w11, dict) and comparison.get("scenario") == "running-heal" and comparison.get("comparison") == "build":
w11_rows.append({
@@ -338,6 +365,9 @@ def summarize_abba(abba_dir: Path) -> dict[str, Any]:
validate_release_evidence_manifest({**manifest, "evidence": "measured"})
for index, comparison in enumerate(comparisons):
require_measured_comparison_evidence(comparison, index)
require(all(value is not None for value in error_rate_values), "error_rate must be measured")
require(pressure_samples_complete, "foreground_pressure_samples must be measured")
require(pressure_high_samples_complete, "foreground_pressure_high_samples must be measured")
gate_state = "pass" if passed else "fail"
if report_state == "synthetic_validated":
reason = "synthetic evidence validates the harness only; measured performance remains pending"
@@ -370,9 +400,10 @@ def summarize_abba(abba_dir: Path) -> dict[str, Any]:
"foreground_p95_ms": None if max_decimal(foreground_p95_values) is None else float(max_decimal(foreground_p95_values)),
"foreground_p99_ms": None if max_decimal(foreground_p99_values) is None else float(max_decimal(foreground_p99_values)),
"throughput_ops": None if max_decimal(throughput_values) is None else float(max_decimal(throughput_values)),
"error_rate": 0.0 if not error_rate_values else float(max(error_rate_values)),
"foreground_pressure_samples": pressure_samples,
"foreground_pressure_high_samples": pressure_high_samples,
"error_rate": (None if not matrix_complete or any(value is None for value in error_rate_values)
else float(max_decimal(error_rate_values))),
"foreground_pressure_samples": pressure_samples if pressure_samples_complete else None,
"foreground_pressure_high_samples": pressure_high_samples if pressure_high_samples_complete else None,
"attempt_cost_samples": attempt_cost_samples,
"p2_worst_post_stop_work_multiple": None if max_decimal(p2_values) is None else float(max_decimal(p2_values)),
"w09_worst_heal_start_p95_ms": None if max_decimal(start_p95_values) is None else float(max_decimal(start_p95_values)),
+58
View File
@@ -11,6 +11,11 @@ import unittest
import scanner_otlp_window as collector
try:
from opentelemetry.proto.collector.metrics.v1.metrics_service_pb2 import ExportMetricsServiceRequest
except ImportError:
ExportMetricsServiceRequest = None
METRIC = "rustfs_heal_admission_start_duration_seconds"
START = 1_800_000_000_000_000_001
@@ -38,6 +43,21 @@ def point(document):
return metric(document)["histogram"]["dataPoints"][0]
def protobuf_export(end, buckets):
request = ExportMetricsServiceRequest()
scope = request.resource_metrics.add().scope_metrics.add()
metric = scope.metrics.add()
metric.name = METRIC
metric.histogram.aggregation_temporality = 2
sample = metric.histogram.data_points.add()
sample.start_time_unix_nano = START
sample.time_unix_nano = end
sample.count = sum(buckets)
sample.bucket_counts.extend(buckets)
sample.explicit_bounds.extend([0.001, 0.01])
return request.SerializeToString()
class WindowTests(unittest.TestCase):
def setUp(self):
self.before = export(START + 10, [200, 100, 5])
@@ -66,6 +86,44 @@ class WindowTests(unittest.TestCase):
self.assertIsNone(row["p95"])
self.assertIsNone(row["p99"])
def test_missing_histogram_count_is_not_an_empty_sample(self):
self.before = export(START + 10, [0, 0, 0])
self.after = export(START + 20, [0, 0, 0])
del point(self.before)["count"]
with self.assertRaisesRegex(ValueError, "histogram count"):
self.result()
def test_explicit_zero_histogram_count_remains_a_valid_empty_sample(self):
self.before = export(START + 10, [0, 0, 0])
self.after = export(START + 20, [0, 0, 0])
row = self.result()["series"][0]
self.assertEqual(row["count"], 0)
self.assertIsNone(row["p95"])
self.assertIsNone(row["p99"])
@unittest.skipUnless(ExportMetricsServiceRequest is not None, "requires opentelemetry-proto")
def test_protobuf_default_zero_count_without_presence_is_rejected(self):
with tempfile.TemporaryDirectory() as tmp:
path = Path(tmp) / "zero.pb"
path.write_bytes(protobuf_export(START + 10, [0, 0, 0]))
before, _ = collector.load(path, "protobuf")
self.after = export(START + 20, [0, 0, 0])
with self.assertRaisesRegex(ValueError, "histogram count"):
collector.window(before, self.after, [METRIC])
@unittest.skipUnless(ExportMetricsServiceRequest is not None, "requires opentelemetry-proto")
def test_protobuf_nonzero_counts_produce_window_bounds(self):
with tempfile.TemporaryDirectory() as tmp:
before_path = Path(tmp) / "before.pb"
after_path = Path(tmp) / "after.pb"
before_path.write_bytes(protobuf_export(START + 10, [0, 1, 0]))
after_path.write_bytes(protobuf_export(START + 20, [0, 2, 0]))
before, _ = collector.load(before_path, "protobuf")
after, _ = collector.load(after_path, "protobuf")
row = collector.window(before, after, [METRIC], START + 10, START + 20)["series"][0]
self.assertEqual(row["count"], 1)
self.assertEqual(row["p99"], {"lower_exclusive": 0.001, "upper_inclusive": 0.01})
def test_open_tail_is_not_replaced_with_historical_max(self):
self.after = export(START + 20, [200, 100, 105])
self.assertEqual(self.result()["series"][0]["p95"],
@@ -226,6 +226,7 @@ class ScannerHealPerfSummaryTest(unittest.TestCase):
})
result = summary.build_summary(args)
self.assertEqual(result["verdict"], "PASS")
self.assertEqual(result["abba"]["error_rate"], 0.0)
self.assertEqual(result["abba"]["provenance"]["manifest_sha256"], sha(self.abba / "manifest.json"))
self.assertEqual(result["abba"]["w09_duplicate_task_count"], 0.0)
self.assertEqual(result["abba"]["w09_worst_heal_start_p95_ms"], 43.0)
@@ -325,10 +326,64 @@ class ScannerHealPerfSummaryTest(unittest.TestCase):
self.assertEqual(result["verdict"], "FAIL")
self.assertEqual(result["abba"]["completed_cells"], 7)
self.assertEqual(result["abba"]["comparisons_total"], 0)
self.assertIsNone(result["abba"]["error_rate"])
self.assertIsNone(result["abba"]["foreground_pressure_samples"])
self.assertIsNone(result["abba"]["foreground_pressure_high_samples"])
self.assertIn("collector failed", result["reason"])
self.assertIn("- completed_cells: 7", summary.markdown(result))
self.assertIn("- error: collector failed", summary.markdown(result))
def test_partial_error_rate_is_unknown(self):
self.report.update(status="inconclusive", performance="pending")
for value in (None, "missing"):
with self.subTest(value=value):
if value == "missing":
self.report["comparisons"][0].pop("error_rate")
else:
self.report["comparisons"][0]["error_rate"] = value
self.write_inputs()
self.assertIsNone(summary.summarize_abba(self.abba)["error_rate"])
def test_partial_pressure_samples_are_unknown(self):
self.report.update(status="inconclusive", performance="pending")
self.report["comparisons"][0]["w10_w11"].pop("foreground_pressure_samples")
for high_samples in ([0, 0, 0], [0, 0, 0, None]):
with self.subTest(high_samples=high_samples):
self.report["comparisons"][1]["w10_w11"]["foreground_pressure_high_samples"] = high_samples
self.write_inputs()
result = summary.summarize_abba(self.abba)
self.assertIsNone(result["foreground_pressure_samples"])
self.assertIsNone(result["foreground_pressure_high_samples"])
def test_complete_zero_pressure_samples_remain_zero(self):
self.report.update(status="inconclusive", performance="pending")
for row in self.report["comparisons"]:
row["w10_w11"]["foreground_pressure_high_samples"] = [0, 0, 0, 0]
self.write_inputs()
self.assertEqual(summary.summarize_abba(self.abba)["foreground_pressure_high_samples"], 0)
def test_missing_comparison_keeps_matrix_totals_unknown(self):
self.report.update(status="inconclusive", performance="pending")
self.report["comparisons"].pop()
self.write_inputs()
result = summary.summarize_abba(self.abba)
self.assertIsNone(result["error_rate"])
self.assertIsNone(result["foreground_pressure_samples"])
self.assertIsNone(result["foreground_pressure_high_samples"])
def test_passing_report_requires_error_and_pressure_samples(self):
for field in ("error_rate", "foreground_pressure_samples", "foreground_pressure_high_samples"):
with self.subTest(field=field):
self.setUp()
row = self.report["comparisons"][0]
if field == "error_rate":
row.pop(field)
else:
row["w10_w11"].pop(field)
self.write_inputs()
with self.assertRaisesRegex(ValueError, "must be measured"):
summary.summarize_abba(self.abba)
def test_passing_abba_report_requires_comparisons(self):
del self.report["comparisons"]
self.write_inputs()