From d22cb5d07a3d9e2de2bc23e0ebe780e888b8ec79 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Fri, 21 Aug 2026 06:39:29 +0800 Subject: [PATCH] fix: resolve release-blocking integration failures (#6320) * fix: resolve release-blocking integration failures * fix: satisfy stable clippy lints * fix: satisfy Rust 1.98 CI lints --- crates/common/src/metrics.rs | 2 +- .../src/inline_fast_path_cluster_test.rs | 21 +- .../kms_authorization_negative_matrix_test.rs | 15 +- crates/e2e_test/src/quota_test.rs | 63 +++-- crates/e2e_test/src/reliant/tiering.rs | 51 +++- .../ecstore/src/bucket/bucket_target_sys.rs | 25 +- .../bucket/lifecycle/bucket_lifecycle_ops.rs | 4 +- .../lifecycle/transition_transaction.rs | 2 +- crates/ecstore/src/bucket/quota/checker.rs | 7 +- .../replication/replication_resyncer.rs | 12 +- .../src/services/tier/tier_mutation_intent.rs | 2 +- crates/ecstore/src/set_disk/ops/object.rs | 256 +++++++++++++++++- crates/ecstore/src/set_disk/replication.rs | 87 +++++- crates/heal/src/heal/task/tests.rs | 2 +- crates/obs/src/metrics/stats_collector.rs | 2 +- crates/policy/src/policy/function/date.rs | 15 +- crates/scanner/src/remote_scanner/stream.rs | 2 +- rustfs/src/admin/handlers/inspect_archive.rs | 2 +- rustfs/src/admin/router.rs | 2 +- rustfs/src/app/object_usecase.rs | 10 + rustfs/src/storage/rpc/http_service.rs | 35 ++- rustfs/tests/inspect_cli.rs | 4 +- 22 files changed, 509 insertions(+), 112 deletions(-) diff --git a/crates/common/src/metrics.rs b/crates/common/src/metrics.rs index 9a6198d35..51eef967a 100644 --- a/crates/common/src/metrics.rs +++ b/crates/common/src/metrics.rs @@ -729,7 +729,7 @@ fn timestamp_elapsed_seconds_since(now: Timestamp, earlier: Timestamp) -> u64 { return 0; } - u64::try_from(duration.as_secs()).map_or(u64::MAX, |seconds| seconds) + u64::try_from(duration.as_secs()).unwrap_or(u64::MAX) } #[derive(Clone, Copy, Debug, Default)] diff --git a/crates/e2e_test/src/inline_fast_path_cluster_test.rs b/crates/e2e_test/src/inline_fast_path_cluster_test.rs index f1cdc1acf..04a0a9f05 100644 --- a/crates/e2e_test/src/inline_fast_path_cluster_test.rs +++ b/crates/e2e_test/src/inline_fast_path_cluster_test.rs @@ -1028,20 +1028,6 @@ impl<'a> ReaderPathExpectation<'a> { } } - fn with_size_bucket( - object: ReaderObject<'a>, - expected_path: &'a str, - object_class: &'a str, - expected_size_bucket: &'a str, - ) -> Self { - Self { - object, - expected_path, - object_class, - expected_size_bucket: Some(expected_size_bucket), - } - } - fn with_any_size_bucket(object: ReaderObject<'a>, expected_path: &'a str, object_class: &'a str) -> Self { Self { object, @@ -1909,12 +1895,7 @@ async fn four_node_compressed_inline_fallback() -> TestResult { assert_reader_path( &collector, &client, - ReaderPathExpectation::with_size_bucket( - ReaderObject::new(bucket, key, &body, put.e_tag(), None), - LEGACY_DUPLEX, - COMPRESSED, - size_bucket(4 * KIB), - ), + ReaderPathExpectation::for_class(ReaderObject::new(bucket, key, &body, put.e_tag(), None), LEGACY_DUPLEX, COMPRESSED), ) .await?; diff --git a/crates/e2e_test/src/kms/kms_authorization_negative_matrix_test.rs b/crates/e2e_test/src/kms/kms_authorization_negative_matrix_test.rs index a6fd6cb43..ac212536d 100644 --- a/crates/e2e_test/src/kms/kms_authorization_negative_matrix_test.rs +++ b/crates/e2e_test/src/kms/kms_authorization_negative_matrix_test.rs @@ -130,7 +130,7 @@ fn policy_document(statements: Vec) -> String { serde_json::json!({ "Version": "2012-10-17", "Statement": statements }).to_string() } -async fn put_sse_kms(client: &Client, key: &str, kms_key_id: &str) -> Result<(), aws_sdk_s3::Error> { +async fn put_sse_kms(client: &Client, key: &str, kms_key_id: &str) -> Result<(), Box> { client .put_object() .bucket(BUCKET) @@ -141,16 +141,23 @@ async fn put_sse_kms(client: &Client, key: &str, kms_key_id: &str) -> Result<(), .send() .await .map(|_| ()) - .map_err(aws_sdk_s3::Error::from) + .map_err(|error| Box::new(aws_sdk_s3::Error::from(error))) } /// Assert the operation failed with `AccessDenied` rather than any other error. /// /// A bare `is_err` would also accept `KMSKeyDisabled` or an internal error, which /// would hide both a leak of key state and an outage masquerading as a denial. -fn assert_access_denied(result: Result, what: &str) { +fn assert_access_denied>( + result: Result, + what: &str, +) { let error = result.expect_err(&format!("{what} must be denied")); - assert_eq!(error.code(), Some("AccessDenied"), "{what} must fail with AccessDenied: {error:?}"); + assert_eq!( + error.borrow().code(), + Some("AccessDenied"), + "{what} must fail with AccessDenied: {error:?}" + ); } /// Retry an SSE-KMS write until the identity's policy has reached the request path. diff --git a/crates/e2e_test/src/quota_test.rs b/crates/e2e_test/src/quota_test.rs index e7c095674..0e73f2212 100644 --- a/crates/e2e_test/src/quota_test.rs +++ b/crates/e2e_test/src/quota_test.rs @@ -169,6 +169,42 @@ impl QuotaTestEnv { bucket: &str, quota_bytes: u64, ) -> Result<(), Box> { + self.wait_for_quota_usage_for(bucket).await?; + + let quota_path = format!("/rustfs/admin/v3/quota/{bucket}"); + let quota_config = serde_json::json!({ + "quota": quota_bytes, + "quota_type": "HARD" + }) + .to_string(); + let readiness = async { + loop { + let (status, response) = admin_request( + &self.env.url, + Method::PUT, + "a_path, + Some(quota_config.clone()), + &self.env.access_key, + &self.env.secret_key, + ) + .await?; + if status.is_success() { + return Ok::<(), Box>(()); + } + if status != StatusCode::SERVICE_UNAVAILABLE { + return Err(format!("failed to set quota for {bucket}: {status} {response}").into()); + } + + sleep(Duration::from_secs(1)).await; + } + }; + match timeout(Duration::from_secs(30), readiness).await { + Ok(result) => result, + Err(_) => Err(format!("quota readiness did not converge for {bucket} within 30 seconds").into()), + } + } + + pub async fn wait_for_quota_usage_for(&self, bucket: &str) -> Result<(), Box> { let stats_path = format!("/rustfs/admin/v3/quota-stats/{bucket}"); let readiness = async { loop { @@ -181,28 +217,12 @@ impl QuotaTestEnv { if status != StatusCode::SERVICE_UNAVAILABLE { return Err(format!("quota usage readiness failed for {bucket}: {status} {response}").into()); } - sleep(Duration::from_secs(1)).await; } }; match timeout(Duration::from_secs(30), readiness).await { - Ok(result) => result?, - Err(_) => { - return Err(format!("quota usage did not become authoritative for {bucket} within 30 seconds").into()); - } - } - - let url = format!("{}/rustfs/admin/v3/quota/{}", self.env.url, bucket); - let quota_config = serde_json::json!({ - "quota": quota_bytes, - "quota_type": "HARD" - }); - - let response = awscurl_put(&url, "a_config.to_string(), &self.env.access_key, &self.env.secret_key).await?; - if response.contains("error") { - Err(format!("Failed to set quota: {}", response).into()) - } else { - Ok(()) + Ok(result) => result, + Err(_) => Err(format!("quota usage did not become authoritative for {bucket} within 30 seconds").into()), } } @@ -621,12 +641,7 @@ mod integration_tests { assert!(response.contains("quota") && response.contains("null")); // Test 2: PUT quota - valid config - let quota_config = serde_json::json!({ - "quota": 1048576, - "quota_type": "HARD" - }); - let response = awscurl_put(&url, "a_config.to_string(), &env.env.access_key, &env.env.secret_key).await?; - assert!(response.contains("success") || !response.contains("error")); + env.set_bucket_quota(1048576).await?; // Test 3: GET quota after setting let response = awscurl_get(&url, &env.env.access_key, &env.env.secret_key).await?; diff --git a/crates/e2e_test/src/reliant/tiering.rs b/crates/e2e_test/src/reliant/tiering.rs index 7f5a1c913..cb478cbf9 100644 --- a/crates/e2e_test/src/reliant/tiering.rs +++ b/crates/e2e_test/src/reliant/tiering.rs @@ -110,6 +110,7 @@ const USER_META_KEY: &str = "ilm7-origin"; const USER_META_VAL: &str = "hermetic-transition"; const HDR_SOURCE_REPLICATION_REQUEST: &str = "x-rustfs-source-replication-request"; const HDR_SOURCE_MTIME: &str = "x-rustfs-source-mtime"; +const TIER_MUTATION_RECOVERY_CHANGED: &str = "Remote tier mutation recovery changed before publish"; /// 5 MiB — the S3 minimum size for a non-final multipart part; the object's only /// internal part boundary sits at this offset. @@ -183,19 +184,39 @@ async fn add_rustfs_tier(hot: &RustFSTestEnvironment, cold: &RustFSTestEnvironme }) .to_string(); - let (status, resp) = signed_admin_request( - &hot.url, - Method::PUT, - "/rustfs/admin/v3/tier", - Some(&body), - &hot.access_key, - &hot.secret_key, - ) - .await?; - if !status.is_success() { - return Err(format!("AddTier(RustFS) failed: status={status}, body={resp}").into()); + let verify_path = format!("/rustfs/admin/v3/tier/{TIER_NAME}"); + let deadline = Instant::now() + StdDuration::from_secs(30); + let mut recovery_changed = false; + loop { + if recovery_changed { + let (status, _) = + signed_admin_request(&hot.url, Method::GET, &verify_path, None, &hot.access_key, &hot.secret_key).await?; + if status.is_success() { + return Ok(()); + } + } + let (status, resp) = signed_admin_request( + &hot.url, + Method::PUT, + "/rustfs/admin/v3/tier", + Some(&body), + &hot.access_key, + &hot.secret_key, + ) + .await?; + if status.is_success() { + return Ok(()); + } + if resp.contains(TIER_MUTATION_RECOVERY_CHANGED) { + recovery_changed = true; + } else if !recovery_changed || !resp.contains("TierNameAlreadyExist") { + return Err(format!("AddTier(RustFS) failed: status={status}, body={resp}").into()); + } + if Instant::now() >= deadline { + return Err(format!("AddTier(RustFS) failed: status={status}, body={resp}").into()); + } + tokio::time::sleep(StdDuration::from_millis(100)).await; } - Ok(()) } async fn remove_rustfs_tier_force(hot: &RustFSTestEnvironment) -> TestResult { @@ -207,10 +228,12 @@ async fn remove_rustfs_tier_force(hot: &RustFSTestEnvironment) -> TestResult { if status.is_success() { return Ok(()); } - if !resp.contains("TierNameBackendInUse") || Instant::now() >= deadline { + if (!resp.contains("TierNameBackendInUse") && !resp.contains(TIER_MUTATION_RECOVERY_CHANGED)) + || Instant::now() >= deadline + { return Err(format!("RemoveTier(RustFS) failed: status={status}, body={resp}").into()); } - // AddTier cleanup is asynchronous; wait until its committed mutation fence clears. + // Tier mutation cleanup and startup recovery are asynchronous. tokio::time::sleep(StdDuration::from_millis(100)).await; } } diff --git a/crates/ecstore/src/bucket/bucket_target_sys.rs b/crates/ecstore/src/bucket/bucket_target_sys.rs index ef1918648..45cbb87c6 100644 --- a/crates/ecstore/src/bucket/bucket_target_sys.rs +++ b/crates/ecstore/src/bucket/bucket_target_sys.rs @@ -1968,7 +1968,7 @@ impl TargetClient { bucket: &str, object: &str, version_id: Option, - ) -> Result> { + ) -> Result>> { // Announce the replication check so a RustFS target returns SSE-C // object metadata (etag/size) without the customer key the replication // worker cannot hold; otherwise SSE-C replicas never converge on HEAD. @@ -1981,8 +1981,7 @@ impl TargetClient { // object with an identical ETag, and the worker concludes the object // already converged — so it never actually replicates it. insert_header(&mut headers, SUFFIX_SOURCE_PROXY_REQUEST, "false"); - match self - .client + self.client .head_object() .bucket(bucket) .key(object) @@ -1999,10 +1998,7 @@ impl TargetClient { }) .send() .await - { - Ok(res) => Ok(res), - Err(e) => Err(e), - } + .map_err(Box::new) } /// HEAD used by the read-proxy path (GET/HEAD of an object not yet @@ -2023,7 +2019,7 @@ impl TargetClient { range: Option, part_number: Option, extra_headers: HeaderMap, - ) -> Result> { + ) -> Result>> { let headers = proxy_outbound_headers(extra_headers); self.client .head_object() @@ -2036,6 +2032,7 @@ impl TargetClient { .map_request(move |req| apply_extra_headers(req, &headers)) .send() .await + .map_err(Box::new) } /// GET used by the read-proxy path (MinIO `proxyGetToReplicationTarget`). @@ -2051,7 +2048,7 @@ impl TargetClient { range: Option, part_number: Option, extra_headers: HeaderMap, - ) -> Result> { + ) -> Result>> { let headers = proxy_outbound_headers(extra_headers); self.client .get_object() @@ -2064,6 +2061,7 @@ impl TargetClient { .map_request(move |req| apply_extra_headers(req, &headers)) .send() .await + .map_err(Box::new) } /// GetObjectTagging for the tagging read-proxy path @@ -2073,7 +2071,7 @@ impl TargetClient { bucket: &str, object: &str, version_id: Option, - ) -> Result> { + ) -> Result>> { let headers = proxy_outbound_headers(HeaderMap::new()); self.client .get_object_tagging() @@ -2084,6 +2082,7 @@ impl TargetClient { .map_request(move |req| apply_extra_headers(req, &headers)) .send() .await + .map_err(Box::new) } /// PutObjectTagging for the tagging proxy path @@ -2094,7 +2093,7 @@ impl TargetClient { object: &str, version_id: Option, tagging: SdkTagging, - ) -> Result> { + ) -> Result>> { let headers = proxy_outbound_headers(HeaderMap::new()); self.client .put_object_tagging() @@ -2106,6 +2105,7 @@ impl TargetClient { .map_request(move |req| apply_extra_headers(req, &headers)) .send() .await + .map_err(Box::new) } /// DeleteObjectTagging for the tagging proxy path @@ -2115,7 +2115,7 @@ impl TargetClient { bucket: &str, object: &str, version_id: Option, - ) -> Result> { + ) -> Result>> { let headers = proxy_outbound_headers(HeaderMap::new()); self.client .delete_object_tagging() @@ -2126,6 +2126,7 @@ impl TargetClient { .map_request(move |req| apply_extra_headers(req, &headers)) .send() .await + .map_err(Box::new) } /// On success returns the version id the target assigned (from diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index abdca65c1..1162787f7 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -2180,7 +2180,7 @@ pub async fn recover_manual_transition_jobs_once( if limit == 0 { return Err(Error::other("manual transition job recovery limit must be greater than zero")); } - let list_limit = i32::try_from(limit).map_or(i32::MAX, |value| value); + let list_limit = i32::try_from(limit).unwrap_or(i32::MAX); let page = api .clone() .list_objects_v2( @@ -2386,7 +2386,7 @@ async fn replay_manual_transition_pending_tasks( version_id: task.version_id, etag: task.etag, mod_time, - size: task.size.map_or(0, |size| size), + size: task.size.unwrap_or(0), is_latest: task.is_latest.unwrap_or(false), ..Default::default() }; diff --git a/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs b/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs index 7fc231281..f9831d394 100644 --- a/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs +++ b/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs @@ -1016,7 +1016,7 @@ pub async fn recover_transition_transaction_records( return Err(Error::other("transition transaction recovery limit must be greater than zero")); } - let list_limit = i32::try_from(limit).map_or(i32::MAX, |value| value); + let list_limit = i32::try_from(limit).unwrap_or(i32::MAX); let list = api .clone() .list_objects_v2( diff --git a/crates/ecstore/src/bucket/quota/checker.rs b/crates/ecstore/src/bucket/quota/checker.rs index ebb1b4dbb..0d71095e3 100644 --- a/crates/ecstore/src/bucket/quota/checker.rs +++ b/crates/ecstore/src/bucket/quota/checker.rs @@ -76,7 +76,12 @@ impl QuotaChecker { let current_usage = self.get_real_time_usage(bucket).await?; - let admission_size = if uses_durable_reservations { 0 } else { operation_size }; + // The reporting path projects this operation; storage mutations reserve it at commit. + let admission_size = if uses_durable_reservations && !force_usage_calculation { + 0 + } else { + operation_size + }; let expected_usage = match operation { QuotaOperation::PutObject | QuotaOperation::PostObject | QuotaOperation::CopyObject => { current_usage.saturating_add(admission_size) diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 882ece91c..5ec5bb03e 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -214,7 +214,7 @@ async fn head_object_for_worker( target_bucket: &str, object: &str, version_id: Option, -) -> std::result::Result> { +) -> std::result::Result>> { target_client.head_object(target_bucket, object, version_id).await } @@ -233,7 +233,7 @@ async fn mark_replication_target_offline_if_needed(target_client: &Arc std::result::Result, SdkError> { +) -> std::result::Result, Box>> { match head_object_for_worker(tgt_client, &tgt_client.bucket, object, None).await { Ok(oi) => Ok(Some(oi)), Err(e) if e.as_service_error().is_some_and(|se| se.is_not_found()) || has_raw_status(&e, 404) => Ok(None), @@ -1152,11 +1152,11 @@ fn spawn_resync_walk_task( /// updating the per-object status counters and returning the accounted size /// together with any verification error. async fn verify_resync_head_result( - head_result: std::result::Result>, + head_result: std::result::Result>>, roi: &ReplicateObjectInfo, st: &mut TargetReplicationResyncStatus, target_client: &Arc, -) -> (i64, Option>) { +) -> (i64, Option>>) { match head_result { Ok(_) => { st.replicated_count += 1; @@ -1275,7 +1275,7 @@ async fn resync_worker_process_object( "Processed resync object" ); } - st.error = err.as_ref().and_then(resync_target_error_detail); + st.error = err.as_ref().and_then(|err| resync_target_error_detail(err)); st } @@ -2467,7 +2467,7 @@ async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_cli Ok(_) => {} Err(e) => { let non_retryable = matches!( - &e, + &*e, SdkError::ServiceError(service_err) if is_retryable_delete_replication_head_error( service_err.err().is_not_found(), diff --git a/crates/ecstore/src/services/tier/tier_mutation_intent.rs b/crates/ecstore/src/services/tier/tier_mutation_intent.rs index 94c940653..4b9ea7e61 100644 --- a/crates/ecstore/src/services/tier/tier_mutation_intent.rs +++ b/crates/ecstore/src/services/tier/tier_mutation_intent.rs @@ -657,7 +657,7 @@ where prefix, marker, None, - i32::try_from(limit).map_or(i32::MAX, |value| value), + i32::try_from(limit).unwrap_or(i32::MAX), false, None, false, diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 20ff815a7..9ba7f8da7 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -1322,7 +1322,12 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { let object_info = prepared_object_info .unwrap_or_else(|| build_get_object_info(fi, bucket, object, opts.versioned || opts.version_suspended)); let object_class = classify_get_codec_streaming_object_class(&range, &object_info, fi); - let size_bucket = rustfs_io_metrics::get_object_size_bucket(object_info.size); + let metrics_size = if stage_metrics_enabled { + object_info.get_actual_size().unwrap_or(object_info.size) + } else { + object_info.size + }; + let size_bucket = rustfs_io_metrics::get_object_size_bucket(metrics_size); record_get_stage_duration_if_enabled(GET_OBJECT_PATH_SET_DISK, GET_STAGE_OBJECT_INFO, object_info_stage_start); let metadata_elapsed = metadata_stage_start.elapsed().as_secs_f64(); rustfs_io_metrics::record_get_object_metadata_phase_duration(metadata_elapsed); @@ -3766,7 +3771,7 @@ pub(crate) async fn complete_transition_upload( producer: Producer, expected_size: u64, consumed: Arc, -) -> std::result::Result +) -> std::result::Result> where Remote: Future>, Producer: Future>, @@ -3784,23 +3789,23 @@ where Err(_) => StorageError::Unexpected, Ok(Ok(_)) => StorageError::Io(remote_error), }; - return Err(TransitionUploadFailure { error, candidate: None }); + return Err(Box::new(TransitionUploadFailure { error, candidate: None })); } }; let candidate = TransitionUploadCandidate::from_put_response(remote_version); let produced = match producer_result { Ok(Ok(produced)) => produced, Ok(Err(error)) => { - return Err(TransitionUploadFailure { + return Err(Box::new(TransitionUploadFailure { error, candidate: Some(candidate), - }); + })); } Err(_) => { - return Err(TransitionUploadFailure { + return Err(Box::new(TransitionUploadFailure { error: StorageError::Unexpected, candidate: Some(candidate), - }); + })); } }; let consumed = consumed.load(Ordering::Acquire); @@ -3810,10 +3815,10 @@ where } else { StorageError::MoreData }; - return Err(TransitionUploadFailure { + return Err(Box::new(TransitionUploadFailure { error, candidate: Some(candidate), - }); + })); } Ok(TransitionUploadCompletion { candidate, @@ -7284,7 +7289,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { } let gr = gr?; let reader = BufReader::new(gr.stream); - let hash_reader = HashReader::from_stream(reader, gr.object_info.size, gr.object_info.size, None, None, false)?; + let hash_reader = HashReader::from_stream(reader, gr.object_info.size, oi.get_actual_size()?, None, None, false)?; let mut p_reader = PutObjReader::new(hash_reader); return match self_.clone().put_object(bucket, object, &mut p_reader, &ropts).await { Ok(restored_info) => { @@ -8826,7 +8831,7 @@ mod transition_commit_failure_tests { use s3s::dto::RestoreRequest; use tokio::io::{AsyncReadExt, AsyncWriteExt}; - fn restore_operation_id_metadata(operation_id: Uuid) -> HashMap { + pub(super) fn restore_operation_id_metadata(operation_id: Uuid) -> HashMap { let mut metadata = HashMap::new(); rustfs_utils::http::metadata_compat::insert_str( &mut metadata, @@ -8836,7 +8841,7 @@ mod transition_commit_failure_tests { metadata } - fn restore_metadata(operation_id: Uuid, ongoing: bool) -> HashMap { + pub(super) fn restore_metadata(operation_id: Uuid, ongoing: bool) -> HashMap { let mut metadata = restore_operation_id_metadata(operation_id); metadata.insert(s3s::header::X_AMZ_RESTORE.as_str().to_string(), format!("ongoing-request=\"{ongoing}\"")); metadata @@ -10097,6 +10102,51 @@ mod transition_commit_failure_tests { .await .expect("operation B should replace operation A before final commit"); + let mismatch = set_disks + .finalize_restore_metadata( + bucket, + object, + &set_disks + .get_object_info(bucket, object, &ObjectOptions::default()) + .await + .expect("operation B metadata should be readable"), + &ObjectOptions { + user_defined: restore_operation_id_metadata(operation_a), + ..Default::default() + }, + ) + .await + .expect_err("operation A must not finalize operation B metadata"); + assert!(matches!( + mismatch, + Error::Io(ref error) + if error.kind() == std::io::ErrorKind::Other + && error.to_string() == "restore operation id changed before metadata finalization" + )); + let current = set_disks + .get_object_info(bucket, object, &ObjectOptions::default()) + .await + .expect("operation B metadata should remain after mismatched finalization"); + assert_eq!( + rustfs_utils::http::metadata_compat::get_consistent_str( + current.user_defined.as_ref(), + rustfs_utils::http::metadata_compat::SUFFIX_RESTORE_OPERATION_ID, + ), + Some(operation_b.to_string().as_str()), + "mismatched finalization must not remove operation B" + ); + assert!( + parse_restore_obj_status( + current + .user_defined + .get(s3s::header::X_AMZ_RESTORE.as_str()) + .expect("operation B restore header should remain pending"), + ) + .expect("operation B restore header should parse") + .on_going(), + "mismatched finalization must not publish restore completion" + ); + let mut stale_restore_reader = PutObjReader::from_vec(b"stale A restored body".repeat(1024)); let result = set_disks .put_object( @@ -10126,18 +10176,37 @@ mod transition_commit_failure_tests { let mut matching_restore_reader = PutObjReader::from_vec(b"matching B restored body".repeat(1024)); let operation_b_restore_metadata = restore_metadata(operation_b, false); - set_disks + let restored = set_disks .put_object( bucket, object, &mut matching_restore_reader, &ObjectOptions { - user_defined: operation_b_restore_metadata, + user_defined: operation_b_restore_metadata.clone(), ..Default::default() }, ) .await .expect("matching operation B should be allowed to commit"); + set_disks + .finalize_restore_metadata( + bucket, + object, + &restored, + &ObjectOptions { + user_defined: restore_operation_id_metadata(operation_b), + transition: TransitionOptions { + restore_request: RestoreRequest { + days: Some(1), + ..Default::default() + }, + ..Default::default() + }, + ..Default::default() + }, + ) + .await + .expect("matching operation B should finalize after its commit consumes the operation id"); let restored = set_disks .get_object_info(bucket, object, &ObjectOptions::default()) .await @@ -10503,13 +10572,16 @@ mod transition_commit_failure_tests { #[cfg(all(test, feature = "test-util"))] mod transition_upload_integrity_tests { use super::hermetic_set_disks_support::{hermetic_set_disks, hermetic_set_disks_with_lockers}; + use super::transition_commit_failure_tests::{restore_metadata, restore_operation_id_metadata}; use super::*; use crate::bucket::lifecycle::lifecycle::{TRANSITION_PENDING, TransitionOptions}; use crate::disk::DiskAPI as _; use crate::layout::endpoints::SetupType; use crate::services::tier::test_util::register_mock_tier; + use crate::set_disk::replication::RestoreFinalizeBarrier; use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _}; use http::HeaderMap; + use rustfs_filemeta::RestoreStatusOps as _; use rustfs_lock::client::local::LocalClient; use rustfs_lock::{LockClient, LockError, LockId, LockInfo, LockRequest, LockResponse, LockStats}; use std::collections::HashSet; @@ -10655,6 +10727,162 @@ mod transition_upload_integrity_tests { } } + async fn write_committed_restore( + set_disks: &Arc, + disk_stores: &[DiskStore], + bucket: &str, + object: &str, + operation_id: Uuid, + ) -> ObjectInfo { + for disk in disk_stores { + disk.make_volume(bucket).await.expect("bucket volume should be created"); + } + let mut source = PutObjReader::from_vec(b"restore source body".repeat(1024)); + set_disks + .put_object(bucket, object, &mut source, &ObjectOptions::default()) + .await + .expect("source object should be written"); + set_disks + .put_object_metadata( + bucket, + object, + &ObjectOptions { + eval_metadata: Some(restore_metadata(operation_id, true)), + ..Default::default() + }, + ) + .await + .expect("pending restore metadata should be installed"); + + let mut restored_reader = PutObjReader::from_vec(b"restored body".repeat(1024)); + set_disks + .put_object( + bucket, + object, + &mut restored_reader, + &ObjectOptions { + user_defined: restore_metadata(operation_id, true), + ..Default::default() + }, + ) + .await + .expect("matching restore commit should consume its operation id") + } + + fn restore_finalize_options(operation_id: Uuid) -> ObjectOptions { + ObjectOptions { + user_defined: restore_operation_id_metadata(operation_id), + transition: TransitionOptions { + restore_request: s3s::dto::RestoreRequest { + days: Some(1), + ..Default::default() + }, + ..Default::default() + }, + ..Default::default() + } + } + + async fn assert_committed_restore_remains_pending(set_disks: &Arc, bucket: &str, object: &str) { + let current = set_disks + .get_object_info(bucket, object, &ObjectOptions::default()) + .await + .expect("pending restore metadata should remain readable"); + assert!( + restore_operation_id_from_metadata(current.user_defined.as_ref()) + .expect("operation id metadata should parse") + .is_none(), + "successful restore commit must have consumed the operation id" + ); + assert!( + rustfs_filemeta::parse_restore_obj_status( + current + .user_defined + .get(s3s::header::X_AMZ_RESTORE.as_str()) + .expect("pending restore header should remain"), + ) + .expect("restore header should parse") + .on_going(), + "failed finalization must not publish completion metadata" + ); + } + + #[tokio::test(flavor = "current_thread", start_paused = true)] + #[serial_test::serial] + async fn restore_finalize_rejects_acquired_lock_loss_after_commit() { + let refresh_calls = Arc::new(AtomicUsize::new(0)); + let lockers: Vec> = (0..4) + .map(|_| Arc::new(LockLostRefreshClient::new(Arc::clone(&refresh_calls))) as Arc) + .collect(); + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks_with_lockers(4, 0, 2, lockers).await; + let bucket = "restore-finalize-acquired-lock-lost-bucket"; + let object = "object.bin"; + let operation_id = Uuid::new_v4(); + let restored = write_committed_restore(&set_disks, &disk_stores, bucket, object, operation_id).await; + let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await; + let barrier = RestoreFinalizeBarrier::install(bucket, object); + let finalize_set = Arc::clone(&set_disks); + let finalize = tokio::spawn(async move { + finalize_set + .finalize_restore_metadata(bucket, object, &restored, &restore_finalize_options(operation_id)) + .await + }); + barrier.wait_until_paused().await; + tokio::time::advance(Duration::from_secs(11)).await; + tokio::task::yield_now().await; + assert!(refresh_calls.load(Ordering::SeqCst) > 0, "restore finalization lock must attempt renewal"); + barrier.release(); + + let error = finalize + .await + .expect("restore finalization task should join") + .expect_err("lost acquired lock must reject restore finalization"); + assert!(matches!( + error, + Error::Io(ref error) + if error.kind() == std::io::ErrorKind::Other + && error.to_string() == "restore finalization lock lost before metadata update" + )); + assert_committed_restore_remains_pending(&set_disks, bucket, object).await; + } + + #[tokio::test] + #[serial_test::serial] + async fn restore_finalize_rejects_outer_fence_loss_after_metadata_read() { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "restore-finalize-outer-fence-lost-bucket"; + let object = "object.bin"; + let operation_id = Uuid::new_v4(); + let restored = write_committed_restore(&set_disks, &disk_stores, bucket, object, operation_id).await; + let (fence, loss_handle) = NamespaceLockFence::loss_handle_for_test(); + let barrier = RestoreFinalizeBarrier::install(bucket, object); + let finalize_set = Arc::clone(&set_disks); + let finalize = tokio::spawn(async move { + let mut opts = restore_finalize_options(operation_id); + opts.no_lock = true; + opts.namespace_lock_fence = Some(fence); + finalize_set.finalize_restore_metadata(bucket, object, &restored, &opts).await + }); + barrier.wait_until_paused().await; + loss_handle.store(true, std::sync::atomic::Ordering::Release); + barrier.release(); + + let error = finalize + .await + .expect("restore finalization task should join") + .expect_err("lost outer fence must reject restore finalization"); + assert!(matches!( + error, + Error::NamespaceLockQuorumUnavailable { + mode: "restore_finalize_metadata", + required: 1, + achieved: 0, + .. + } + )); + assert_committed_restore_remains_pending(&set_disks, bucket, object).await; + } + async fn assert_local_source_intact(set_disks: &Arc, bucket: &str, object: &str, payload: &[u8]) { let mut restored = Vec::new(); set_disks diff --git a/crates/ecstore/src/set_disk/replication.rs b/crates/ecstore/src/set_disk/replication.rs index 86126c535..5b38c8211 100644 --- a/crates/ecstore/src/set_disk/replication.rs +++ b/crates/ecstore/src/set_disk/replication.rs @@ -18,6 +18,78 @@ use rustfs_filemeta::RestoreStatusOps; use rustfs_utils::http::headers::{AMZ_RESTORE_EXPIRY_DAYS, AMZ_RESTORE_REQUEST_DATE}; use s3s::dto::{RestoreStatus, Timestamp}; +#[cfg(all(test, feature = "test-util"))] +struct RestoreFinalizeBarrierState { + bucket: String, + object: String, + arrived: tokio::sync::Notify, + release: tokio::sync::Notify, +} + +#[cfg(all(test, feature = "test-util"))] +static RESTORE_FINALIZE_BARRIER: std::sync::OnceLock>>> = + std::sync::OnceLock::new(); + +#[cfg(all(test, feature = "test-util"))] +pub(in crate::set_disk) struct RestoreFinalizeBarrier { + state: Arc, +} + +#[cfg(all(test, feature = "test-util"))] +impl RestoreFinalizeBarrier { + pub(in crate::set_disk) fn install(bucket: &str, object: &str) -> Self { + let state = Arc::new(RestoreFinalizeBarrierState { + bucket: bucket.to_string(), + object: object.to_string(), + arrived: tokio::sync::Notify::new(), + release: tokio::sync::Notify::new(), + }); + let mut slot = RESTORE_FINALIZE_BARRIER + .get_or_init(|| std::sync::Mutex::new(None)) + .lock() + .expect("restore finalize barrier mutex should not poison"); + assert!(slot.is_none(), "restore finalize barrier must be installed by one test at a time"); + *slot = Some(Arc::clone(&state)); + Self { state } + } + + pub(in crate::set_disk) async fn wait_until_paused(&self) { + self.state.arrived.notified().await; + } + + pub(in crate::set_disk) fn release(&self) { + self.state.release.notify_one(); + } +} + +#[cfg(all(test, feature = "test-util"))] +impl Drop for RestoreFinalizeBarrier { + fn drop(&mut self) { + let mut slot = RESTORE_FINALIZE_BARRIER + .get_or_init(|| std::sync::Mutex::new(None)) + .lock() + .expect("restore finalize barrier mutex should not poison"); + if slot.as_ref().is_some_and(|state| Arc::ptr_eq(state, &self.state)) { + *slot = None; + } + } +} + +#[cfg(all(test, feature = "test-util"))] +async fn maybe_pause_restore_finalize(bucket: &str, object: &str) { + let barrier = RESTORE_FINALIZE_BARRIER + .get_or_init(|| std::sync::Mutex::new(None)) + .lock() + .expect("restore finalize barrier mutex should not poison") + .as_ref() + .filter(|barrier| barrier.bucket == bucket && barrier.object == object) + .cloned(); + if let Some(barrier) = barrier { + barrier.arrived.notify_one(); + barrier.release.notified().await; + } +} + #[derive(Clone, Copy, Debug, Eq, PartialEq)] struct RestoreCleanupIdentity { version_id: Option, @@ -80,7 +152,7 @@ impl SetDisks { .clone() .unwrap_or_else(|| get_raw_etag(obj_info.user_defined.as_ref())); let version_id = expected.version_id.map(|v| v.to_string()); - let _lock_guard = if !opts.no_lock { + let lock_guard = if !opts.no_lock { Some( self.acquire_write_lock_diag("restore_finalize_metadata", bucket, object) .await?, @@ -99,13 +171,16 @@ impl SetDisks { .get_object_fileinfo_gated(bucket, object, &read_opts, false, false) .await? .into_owned(); - if let Some(expected_operation_id) = expected_operation_id { - require_restore_operation_id(&fi.metadata, expected_operation_id)?; + if let Some(expected_operation_id) = expected_operation_id + && restore_operation_id_from_metadata(&fi.metadata)?.is_some_and(|actual| actual != expected_operation_id) + { + return Err(Error::other("restore operation id changed before metadata finalization")); } if !expected.matches_file_info(&fi, &expected_etag) { return Err(Error::other("restored object changed before restore metadata finalization")); } - ensure_restore_metadata_lock_held(bucket, object, opts, "restore_finalize_metadata")?; + #[cfg(all(test, feature = "test-util"))] + maybe_pause_restore_finalize(bucket, object).await; let restore_expiry = lifecycle::expected_expiry_time(OffsetDateTime::now_utc(), opts.transition.restore_request.days.unwrap_or(1)); fi.metadata.insert( @@ -117,6 +192,10 @@ impl SetDisks { .to_string(), ); self.invalidate_get_object_metadata_cache(bucket, object).await; + ensure_restore_metadata_lock_held(bucket, object, opts, "restore_finalize_metadata")?; + if lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()) { + return Err(Error::other("restore finalization lock lost before metadata update")); + } self.update_object_meta_with_opts( bucket, object, diff --git a/crates/heal/src/heal/task/tests.rs b/crates/heal/src/heal/task/tests.rs index 25040d341..464ff3606 100644 --- a/crates/heal/src/heal/task/tests.rs +++ b/crates/heal/src/heal/task/tests.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use super::super::{DiskOption, DiskStore, Endpoint, HealDiskExt as _, new_disk}; +use super::super::{DiskOption, DiskStore, Endpoint, new_disk}; use super::*; use crate::heal::storage::{HealListItem, HealObjectInfo}; use rustfs_common::trace_bus::{TraceEvent, TraceFunc, TraceKind, TraceSubscription, TraceVal, subscribe_trace_events}; diff --git a/crates/obs/src/metrics/stats_collector.rs b/crates/obs/src/metrics/stats_collector.rs index e80e3a199..20e461952 100644 --- a/crates/obs/src/metrics/stats_collector.rs +++ b/crates/obs/src/metrics/stats_collector.rs @@ -337,7 +337,7 @@ fn timestamp_elapsed_seconds_since(now: Timestamp, earlier: Timestamp) -> u64 { return 0; } - u64::try_from(duration.as_secs()).map_or(u64::MAX, |seconds| seconds) + u64::try_from(duration.as_secs()).unwrap_or(u64::MAX) } fn scanner_scan_mode_code(scan_mode: &str) -> u64 { diff --git a/crates/policy/src/policy/function/date.rs b/crates/policy/src/policy/function/date.rs index 9c8dd5981..b66d9e37a 100644 --- a/crates/policy/src/policy/function/date.rs +++ b/crates/policy/src/policy/function/date.rs @@ -31,7 +31,7 @@ impl DateFunc { return false; }; - if !op(&inner.values.0, &rv) { + if !op(&rv, &inner.values.0) { return false; } } @@ -95,6 +95,7 @@ mod tests { key_name::KeyName::{self, *}, key_name::S3KeyName::*, }; + use std::collections::HashMap; use test_case::test_case; use time::{OffsetDateTime, format_description::well_known::Rfc3339}; @@ -122,4 +123,16 @@ mod tests { assert_eq!(v, expect); Ok(()) } + + #[test] + fn evaluate_compares_request_date_to_policy_date() { + let function = new_func(S3(S3ObjectLockRetainUntilDate), None, "2030-01-01T00:00:00Z"); + let later = HashMap::from([("object-lock-retain-until-date".to_string(), vec!["2099-01-01T00:00:00Z".to_string()])]); + let earlier = HashMap::from([("object-lock-retain-until-date".to_string(), vec!["2029-01-01T00:00:00Z".to_string()])]); + + assert!(function.evaluate(OffsetDateTime::gt, &later)); + assert!(!function.evaluate(OffsetDateTime::gt, &earlier)); + assert!(function.evaluate(OffsetDateTime::lt, &earlier)); + assert!(!function.evaluate(OffsetDateTime::lt, &later)); + } } diff --git a/crates/scanner/src/remote_scanner/stream.rs b/crates/scanner/src/remote_scanner/stream.rs index aef1c4960..93ba33100 100644 --- a/crates/scanner/src/remote_scanner/stream.rs +++ b/crates/scanner/src/remote_scanner/stream.rs @@ -22,7 +22,7 @@ use crate::scanner_io::{ use crate::storage_api::owner::NS_SCANNER_PROTOCOL_VERSION; use crate::{ DATA_USAGE_CACHE_NAME, DataUsageCache, DataUsageCachePrepareOutcome, DataUsageCacheSource, DataUsageEntryInfo, - DataUsageScanPlanDigest, Disk, ScannerDiskExt as _, ScannerError, StorageError, resolve_scanner_object_store_handle, + DataUsageScanPlanDigest, Disk, ScannerError, StorageError, resolve_scanner_object_store_handle, }; use hmac::{Hmac, KeyInit, Mac}; use rustfs_common::heal_channel::HealScanMode; diff --git a/rustfs/src/admin/handlers/inspect_archive.rs b/rustfs/src/admin/handlers/inspect_archive.rs index 4b6cc0798..e66ba942e 100644 --- a/rustfs/src/admin/handlers/inspect_archive.rs +++ b/rustfs/src/admin/handlers/inspect_archive.rs @@ -643,7 +643,7 @@ mod tests { fn decode_hex_fixture(value: &str) -> Vec { value .split_ascii_whitespace() - .flat_map(|line| line.as_bytes().chunks_exact(2)) + .flat_map(|line| line.as_bytes().as_chunks::<2>().0.iter()) .map(|pair| { let pair = std::str::from_utf8(pair).expect("fixture contains ASCII hex"); u8::from_str_radix(pair, 16).expect("fixture contains valid hex") diff --git a/rustfs/src/admin/router.rs b/rustfs/src/admin/router.rs index 498069eb4..272a93386 100644 --- a/rustfs/src/admin/router.rs +++ b/rustfs/src/admin/router.rs @@ -2694,7 +2694,7 @@ async fn ssec_passthrough_probe_object( let head = target_client .head_object(target_bucket, probe_key, head_version) .await - .map_err(S3ClientError::from)?; + .map_err(|err| S3ClientError::from(*err))?; Ok(ReplicationSsecProbeOutcome { evidence_present: head.sse_customer_algorithm().is_some_and(|algorithm| !algorithm.is_empty()), diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 2e3ec2767..ee95b6244 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -3683,6 +3683,13 @@ where let authorization_headers = pax_headers.clone(); + if let Some(value) = pax_headers.remove("x-amz-tagging") { + let value = value + .to_str() + .map_err(|_| s3_error!(InvalidArgument, "Invalid Snowball object tagging value"))?; + metadata.insert(AMZ_OBJECT_TAGGING.to_owned(), value.to_owned()); + } + let object_lock_mode = pax_headers .remove(AMZ_OBJECT_LOCK_MODE_LOWER) .map(|value| { @@ -10740,6 +10747,7 @@ mod tests { let mut record = pax_record("minio.metadata.Content-Type", b"text/plain"); record.extend(pax_record("minio.metadata.X-Amz-Meta-Owner", b"alice")); record.extend(pax_record("minio.metadata.project", b"alpha-demo")); + record.extend(pax_record("minio.metadata.x-amz-tagging", b"classification=public")); record.extend(pax_record("minio.versionId", Uuid::nil().to_string().as_bytes())); record.extend(pax_record("minio.metadata.x-amz-replication-status", b"REPLICA")); record.extend(pax_record("minio.metadata.X-Amz-Object-Lock-Mode", b"GOVERNANCE")); @@ -10778,6 +10786,8 @@ mod tests { assert_eq!(metadata.get("content-type").map(String::as_str), Some("text/plain")); assert_eq!(metadata.get("owner").map(String::as_str), Some("alice")); assert_eq!(metadata.get("project").map(String::as_str), Some("alpha-demo")); + assert_eq!(metadata.get(AMZ_OBJECT_TAGGING).map(String::as_str), Some("classification=public")); + assert!(!metadata.contains_key("x-amz-tagging")); assert_eq!(metadata.get(AMZ_OBJECT_LOCK_MODE_LOWER).map(String::as_str), Some("GOVERNANCE")); assert_eq!( metadata.get(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER).map(String::as_str), diff --git a/rustfs/src/storage/rpc/http_service.rs b/rustfs/src/storage/rpc/http_service.rs index bdd54fe12..c2983cc43 100644 --- a/rustfs/src/storage/rpc/http_service.rs +++ b/rustfs/src/storage/rpc/http_service.rs @@ -1500,7 +1500,7 @@ where return write_body_chunks_to_writer(body, writer).await; }; - let expected_size = (!query.append && query.size >= 0) + let expected_size = (!query.append && query.size > 0) .then(|| { u64::try_from(query.size) .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "put_file auth size cannot be represented")) @@ -2325,6 +2325,39 @@ mod tests { assert_eq!(writer, b"append-data"); } + #[tokio::test] + async fn put_file_auth_zero_size_create_uses_trailing_auth_record() { + let _ = rustfs_credentials::set_global_rpc_secret("put-file-auth-body-test-secret".to_string()); + let nonce = uuid::Uuid::parse_str("43434343-4444-4555-8666-777777777777").expect("nonce"); + let url = concat!( + "/rustfs/rpc/put_file_stream?disk=disk-a&volume=bucket&path=object%2Fpart.1", + "&append=false&size=0&put_file_auth=digest-trailer-v1&put_file_nonce=43434343-4444-4555-8666-777777777777" + ); + let digest = hex_simd::encode_to_string(sha2::Sha256::digest(b"unknown-size-data"), hex_simd::AsciiCase::Lower); + let trailer = build_put_file_auth_trailer(url, &Method::PUT, nonce, &digest).expect("trailer should build"); + let query = PutFileQuery { + disk: "disk-a".to_string(), + volume: "bucket".to_string(), + path: "object/part.1".to_string(), + append: false, + size: 0, + put_file_auth: Some("digest-trailer-v1".to_string()), + put_file_nonce: Some(nonce), + put_file_server_epoch: Some(*super::PUT_FILE_CAPABILITY_SERVER_EPOCH), + }; + let mut payload = b"unknown-size-data".to_vec(); + payload.extend_from_slice(&trailer); + let body = iter(vec![Ok::(Bytes::from(payload))]); + let mut writer = Vec::new(); + + let copied = write_put_file_body_chunks_to_writer(body, &mut writer, &query, Some(nonce), url) + .await + .expect("zero-size create body should verify"); + + assert_eq!(copied, 17); + assert_eq!(writer, b"unknown-size-data"); + } + #[tokio::test] async fn put_file_auth_append_body_rejects_missing_trailer() { let nonce = uuid::Uuid::parse_str("44444444-5555-4666-8777-888888888888").expect("nonce"); diff --git a/rustfs/tests/inspect_cli.rs b/rustfs/tests/inspect_cli.rs index b96b82e3d..b830f0bbb 100644 --- a/rustfs/tests/inspect_cli.rs +++ b/rustfs/tests/inspect_cli.rs @@ -25,7 +25,9 @@ fn decode_hex(source: &str) -> Vec { .collect::(); digits .as_bytes() - .chunks_exact(2) + .as_chunks::<2>() + .0 + .iter() .map(|pair| u8::from_str_radix(std::str::from_utf8(pair).expect("hex pair"), 16).expect("fixture hex byte")) .collect() }