Compare commits

..

1 Commits

Author SHA1 Message Date
overtrue f4da36b364 test(e2e): activate conditional write regressions 2026-08-23 11:27:51 +08:00
26 changed files with 317 additions and 4428 deletions
+2 -2
View File
@@ -1,2 +1,2 @@
sha256-darwin=9f767b37ed8b1c82da62ea441462d75487785c8086e56f08fb6f6cd89c6e2e52
sha256-linux=fbdaf42b220958d4b1e8880e0f8b5a7992d38e21051bb60596dd4538424757d6
sha256-darwin=03bdfb6a9d6e25d744c385f1461e651f05e4da78e5b0ead2adb3e8b2463e3834
sha256-linux=78c46adad135231017fb8679fb91877b5ae3ec6ce1f75d3d045d2c1076c12a49
-10
View File
@@ -78,12 +78,6 @@ test-group = 'embedded-test-ports'
filter = 'package(rustfs-ecstore) & test(manual_transition_page_checkpoint_persists_durable_job_progress)'
test-group = 'ecstore-serial-flaky'
# The durable ILM decommission regressions build isolated multi-pool stores and
# deliberately take source or target disks offline while checking fencing.
[[profile.default.overrides]]
filter = 'package(rustfs-ecstore) & (test(decommission_migrates_and_verifies_registered_durable_ilm_records) | test(decommission_durable_ilm_target_read_error_is_not_masked_by_peer_success) | test(decommission_durable_ilm_terminal_receipt_recovers_failed_source_cleanup) | test(decommission_durable_ilm_receipt_pagination_fails_closed_on_second_page) | test(decommission_durable_ilm_recovery_keeps_multiple_active_sources))'
test-group = 'ecstore-serial-flaky'
# Serialize the bucket-incarnation / lifecycle-fence tests. They drive
# init_bucket_metadata_sys and bucket_metadata_sys_of, i.e. process-global
# OnceLock state that serial_test's #[serial] cannot protect across nextest's
@@ -196,10 +190,6 @@ test-group = 'embedded-test-ports'
filter = 'package(rustfs-ecstore) & test(manual_transition_page_checkpoint_persists_durable_job_progress)'
test-group = 'ecstore-serial-flaky'
[[profile.ci.overrides]]
filter = 'package(rustfs-ecstore) & (test(decommission_migrates_and_verifies_registered_durable_ilm_records) | test(decommission_durable_ilm_target_read_error_is_not_masked_by_peer_success) | test(decommission_durable_ilm_terminal_receipt_recovers_failed_source_cleanup) | test(decommission_durable_ilm_receipt_pagination_fails_closed_on_second_page) | test(decommission_durable_ilm_recovery_keeps_multiple_active_sources))'
test-group = 'ecstore-serial-flaky'
# Serialize the bucket-incarnation / lifecycle-fence tests under the ci profile
# too (see the matching default-profile override near the top). No retries.
[[profile.ci.overrides]]
+1 -2
View File
@@ -30,8 +30,7 @@ make build-docker BUILD_OS=ubuntu22.04
- Crate membership: `Cargo.toml` `[workspace].members`
- Architecture, layering, crate map: [ARCHITECTURE.md](ARCHITECTURE.md)
- Migration guardrails & readiness contracts: [docs/architecture/](docs/architecture/README.md)
- CI workflow steps: `.github/workflows/`; event, timeout, and required-status
matrix: [docs/testing/ci-gates.md](docs/testing/ci-gates.md)
- CI gates: `.github/workflows/ci.yml` (source of truth; never copy its steps into docs)
- Test-layer taxonomy, per-layer entry commands, serial/nextest rules, flake
policy: [docs/testing/README.md](docs/testing/README.md)
- Tier/ILM transition debugging (xl.meta inspection, versionId tracing):
-2
View File
@@ -70,8 +70,6 @@ make pre-pr
> For the full test-layer taxonomy (unit / ecstore black-box / e2e / s3s-e2e / S3 compatibility / chaos / fuzz / bench), each layer's entry command, the naming conventions the migration gate depends on, and the serial/nextest rules, see [docs/testing/README.md](docs/testing/README.md).
> For the event, timeout, required-status, and local reproduction matrix, see [docs/testing/ci-gates.md](docs/testing/ci-gates.md).
### 🔒 Automated Pre-commit Hooks
#### What `make pre-commit` and `make pre-pr` actually run
Generated
+27 -22
View File
@@ -1858,9 +1858,9 @@ dependencies = [
[[package]]
name = "cc"
version = "1.4.4"
version = "1.4.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0ad534f4357a5264cce5019c989cf66a4f0dc4e0d1b1d15f8aacec0ff7360273"
checksum = "509591b7bcd67f4ef775afad7662703b4935daaa6ec0e5605cfb1090b32a2b6d"
dependencies = [
"find-msvc-tools",
"jobserver",
@@ -2522,6 +2522,12 @@ dependencies = [
"subtle",
]
[[package]]
name = "cty"
version = "0.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b365fabc795046672053e29c954733ec3b05e4be654ab130fe8f1f94d7051f35"
[[package]]
name = "curve25519-dalek"
version = "4.1.3"
@@ -5982,6 +5988,15 @@ version = "0.2.16"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981"
[[package]]
name = "libmimalloc-sys"
version = "0.1.49"
source = "git+https://github.com/xonatius/mimalloc_rust.git?rev=6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11#6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11"
dependencies = [
"cc",
"cty",
]
[[package]]
name = "libredox"
version = "0.1.20"
@@ -6382,6 +6397,14 @@ dependencies = [
"synstructure 0.13.2",
]
[[package]]
name = "mimalloc"
version = "0.1.52"
source = "git+https://github.com/xonatius/mimalloc_rust.git?rev=6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11#6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11"
dependencies = [
"libmimalloc-sys",
]
[[package]]
name = "mime"
version = "0.3.17"
@@ -9139,11 +9162,13 @@ dependencies = [
"insta",
"jiff",
"libc",
"libmimalloc-sys",
"libsystemd",
"matchit 0.9.2",
"md-5 0.11.0",
"metrics",
"metrics-util",
"mimalloc",
"mime_guess",
"opentelemetry",
"opentelemetry_sdk",
@@ -9179,8 +9204,6 @@ dependencies = [
"rustfs-lock",
"rustfs-log-analyzer",
"rustfs-madmin",
"rustfs-mimalloc",
"rustfs-mimalloc-sys",
"rustfs-notify",
"rustfs-object-capacity",
"rustfs-object-data-cache",
@@ -9852,24 +9875,6 @@ dependencies = [
"tokio",
]
[[package]]
name = "rustfs-mimalloc"
version = "0.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a406f4aa07084301d485beec873af6dccc8e3f8762da244743df92038b1db1a6"
dependencies = [
"rustfs-mimalloc-sys",
]
[[package]]
name = "rustfs-mimalloc-sys"
version = "0.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c3051b819175f58445d4c369a72f0ab88149f3885ba8bea2aff3be01f53fe7cd"
dependencies = [
"cc",
]
[[package]]
name = "rustfs-notify"
version = "1.0.0-rc.3"
+2 -2
View File
@@ -350,8 +350,8 @@ russh-sftp = "2.4.0"
dav-server = "0.11.0"
# Performance Analysis and Memory Profiling
rustfs-mimalloc = { version = "0.5.0" }
rustfs-mimalloc-sys = { version = "0.5.0" }
mimalloc = { version = "0.1.52", git = "https://github.com/xonatius/mimalloc_rust.git", rev = "6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11" }
libmimalloc-sys = { version = "0.1.49", git = "https://github.com/xonatius/mimalloc_rust.git", rev = "6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11", features = ["extended"] }
hotpath = { version = "0.23.3", default-features = false }
# Snapshot testing for output format regression detection
insta = { version = "1.48" }
+123 -162
View File
@@ -1,52 +1,26 @@
#![cfg(test)]
use aws_config::meta::region::RegionProviderChain;
use crate::common::{RustFSTestEnvironment, TEST_BUCKET, init_logging};
use aws_sdk_s3::Client;
use aws_sdk_s3::config::{Credentials, Region};
use aws_sdk_s3::error::SdkError;
use aws_sdk_s3::error::{ProvideErrorMetadata, SdkError};
use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart};
use bytes::Bytes;
use std::error::Error;
use std::fmt::Debug;
const ENDPOINT: &str = "http://localhost:9000";
const ACCESS_KEY: &str = "rustfsadmin";
const SECRET_KEY: &str = "rustfsadmin";
const BUCKET: &str = "api-test";
type TestResult = Result<(), Box<dyn Error + Send + Sync>>;
async fn create_aws_s3_client() -> Result<Client, Box<dyn Error>> {
let region_provider = RegionProviderChain::default_provider().or_else(Region::new("us-east-1"));
let shared_config = aws_config::defaults(aws_config::BehaviorVersion::latest())
.region(region_provider)
.credentials_provider(Credentials::new(ACCESS_KEY, SECRET_KEY, None, None, "static"))
.endpoint_url(ENDPOINT)
.load()
.await;
let client = Client::from_conf(
aws_sdk_s3::Config::from(&shared_config)
.to_builder()
.force_path_style(true)
.build(),
fn assert_s3_error_code<T, E>(result: Result<T, SdkError<E>>, expected: &str)
where
T: Debug,
E: ProvideErrorMetadata + Debug,
{
let error = result.expect_err("conditional request must fail");
assert_eq!(
error.as_service_error().and_then(ProvideErrorMetadata::code),
Some(expected),
"unexpected conditional request error: {error:?}"
);
Ok(client)
}
/// Setup test bucket, creating it if it doesn't exist
async fn setup_test_bucket(client: &Client) -> Result<(), Box<dyn Error>> {
match client.create_bucket().bucket(BUCKET).send().await {
Ok(_) => {}
Err(SdkError::ServiceError(e)) => {
let e = e.into_err();
let error_code = e.meta().code().unwrap_or("");
if !error_code.eq("BucketAlreadyExists") {
return Err(e.into());
}
}
Err(e) => {
return Err(e.into());
}
}
Ok(())
}
/// Generate test data of specified size
@@ -60,7 +34,12 @@ fn generate_test_data(size: usize) -> Vec<u8> {
}
/// Upload an object and return its ETag
async fn upload_object_with_metadata(client: &Client, bucket: &str, key: &str, data: &[u8]) -> Result<String, Box<dyn Error>> {
async fn upload_object_with_metadata(
client: &Client,
bucket: &str,
key: &str,
data: &[u8],
) -> Result<String, Box<dyn Error + Send + Sync>> {
let response = client
.put_object()
.bucket(bucket)
@@ -69,188 +48,164 @@ async fn upload_object_with_metadata(client: &Client, bucket: &str, key: &str, d
.send()
.await?;
let etag = response.e_tag().unwrap_or("").to_string();
Ok(etag)
response
.e_tag()
.map(str::to_owned)
.ok_or_else(|| std::io::Error::other("put object response did not include an ETag").into())
}
/// Cleanup test objects from bucket
async fn cleanup_objects(client: &Client, bucket: &str, keys: &[&str]) {
for key in keys {
let _ = client.delete_object().bucket(bucket).key(*key).send().await;
}
}
/// Generate unique test object key
fn generate_test_key(prefix: &str) -> String {
use std::time::{SystemTime, UNIX_EPOCH};
let timestamp = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos();
format!("{prefix}-{timestamp}")
async fn object_body(client: &Client, key: &str) -> Result<Bytes, Box<dyn Error + Send + Sync>> {
let response = client.get_object().bucket(TEST_BUCKET).key(key).send().await?;
Ok(response.body.collect().await?.into_bytes())
}
#[tokio::test]
#[ignore = "requires running RustFS server at localhost:9000"]
async fn test_conditional_put_okay() -> Result<(), Box<dyn std::error::Error>> {
let client = create_aws_s3_client().await?;
setup_test_bucket(&client).await?;
async fn test_conditional_put_okay() -> TestResult {
init_logging();
let mut env = RustFSTestEnvironment::new().await?;
env.start_rustfs_server(vec![]).await?;
env.create_test_bucket(TEST_BUCKET).await?;
let client = env.create_s3_client();
let test_key = generate_test_key("conditional-put-ok");
let test_key = "conditional-put-ok";
let initial_data = generate_test_data(1024); // 1KB test data
let updated_data = generate_test_data(2048); // 2KB updated data
let matching_data = generate_test_data(2048); // 2KB updated data
let non_matching_data = generate_test_data(3072); // 3KB updated data
// Upload initial object and get its ETag
let initial_etag = upload_object_with_metadata(&client, BUCKET, &test_key, &initial_data).await?;
let initial_etag = upload_object_with_metadata(&client, TEST_BUCKET, test_key, &initial_data).await?;
// Test 1: PUT with matching If-Match condition (should succeed)
let response1 = client
client
.put_object()
.bucket(BUCKET)
.key(&test_key)
.body(Bytes::from(updated_data.clone()).into())
.bucket(TEST_BUCKET)
.key(test_key)
.body(Bytes::from(matching_data.clone()).into())
.if_match(&initial_etag)
.send()
.await;
assert!(response1.is_ok(), "PUT with matching If-Match should succeed");
.await?;
assert_eq!(object_body(&client, test_key).await?.as_ref(), matching_data);
// Test 2: PUT with non-matching If-None-Match condition (should succeed)
let fake_etag = "\"fake-etag-12345\"";
let response2 = client
client
.put_object()
.bucket(BUCKET)
.key(&test_key)
.body(Bytes::from(updated_data.clone()).into())
.bucket(TEST_BUCKET)
.key(test_key)
.body(Bytes::from(non_matching_data.clone()).into())
.if_none_match(fake_etag)
.send()
.await;
assert!(response2.is_ok(), "PUT with non-matching If-None-Match should succeed");
// Cleanup
cleanup_objects(&client, BUCKET, &[&test_key]).await;
.await?;
assert_eq!(object_body(&client, test_key).await?.as_ref(), non_matching_data);
Ok(())
}
#[tokio::test]
#[ignore = "requires running RustFS server at localhost:9000"]
async fn test_conditional_put_failed() -> Result<(), Box<dyn std::error::Error>> {
let client = create_aws_s3_client().await?;
setup_test_bucket(&client).await?;
async fn test_conditional_put_failed() -> TestResult {
init_logging();
let mut env = RustFSTestEnvironment::new().await?;
env.start_rustfs_server(vec![]).await?;
env.create_test_bucket(TEST_BUCKET).await?;
let client = env.create_s3_client();
let test_key = generate_test_key("conditional-put-failed");
let test_key = "conditional-put-failed";
let initial_data = generate_test_data(1024);
let updated_data = generate_test_data(2048);
// Upload initial object and get its ETag
let initial_etag = upload_object_with_metadata(&client, BUCKET, &test_key, &initial_data).await?;
let initial_etag = upload_object_with_metadata(&client, TEST_BUCKET, test_key, &initial_data).await?;
// Test 1: PUT with non-matching If-Match condition (should fail with 412)
let fake_etag = "\"fake-etag-should-not-match\"";
let response1 = client
.put_object()
.bucket(BUCKET)
.key(&test_key)
.bucket(TEST_BUCKET)
.key(test_key)
.body(Bytes::from(updated_data.clone()).into())
.if_match(fake_etag)
.send()
.await;
assert!(response1.is_err(), "PUT with non-matching If-Match should fail");
if let Err(e) = response1 {
if let SdkError::ServiceError(e) = e {
let e = e.into_err();
let error_code = e.meta().code().unwrap_or("");
assert_eq!("PreconditionFailed", error_code);
} else {
panic!("Unexpected error: {e:?}");
}
}
assert_s3_error_code(response1, "PreconditionFailed");
assert_eq!(object_body(&client, test_key).await?.as_ref(), initial_data);
// Test 2: PUT with matching If-None-Match condition (should fail with 412)
let response2 = client
.put_object()
.bucket(BUCKET)
.key(&test_key)
.bucket(TEST_BUCKET)
.key(test_key)
.body(Bytes::from(updated_data.clone()).into())
.if_none_match(&initial_etag)
.send()
.await;
assert!(response2.is_err(), "PUT with matching If-None-Match should fail");
if let Err(e) = response2 {
if let SdkError::ServiceError(e) = e {
let e = e.into_err();
let error_code = e.meta().code().unwrap_or("");
assert_eq!("PreconditionFailed", error_code);
} else {
panic!("Unexpected error: {e:?}");
}
}
// Cleanup - only need to clean up the initial object since failed PUTs shouldn't create objects
cleanup_objects(&client, BUCKET, &[&test_key]).await;
assert_s3_error_code(response2, "PreconditionFailed");
assert_eq!(object_body(&client, test_key).await?.as_ref(), initial_data);
Ok(())
}
#[tokio::test]
#[ignore = "requires running RustFS server at localhost:9000"]
async fn test_conditional_put_when_object_does_not_exist() -> Result<(), Box<dyn std::error::Error>> {
let client = create_aws_s3_client().await?;
setup_test_bucket(&client).await?;
async fn test_conditional_put_when_object_does_not_exist() -> TestResult {
init_logging();
let mut env = RustFSTestEnvironment::new().await?;
env.start_rustfs_server(vec![]).await?;
env.create_test_bucket(TEST_BUCKET).await?;
let client = env.create_s3_client();
let key = "some_key";
cleanup_objects(&client, BUCKET, &[key]).await;
let key = "conditional-put-missing";
// When the object does not exist, the If-Match condition should always fail
let response1 = client
.put_object()
.bucket(BUCKET)
.bucket(TEST_BUCKET)
.key(key)
.body(Bytes::from(generate_test_data(1024)).into())
.if_match("*")
.send()
.await;
assert!(response1.is_err());
if let Err(e) = response1 {
if let SdkError::ServiceError(e) = e {
let e = e.into_err();
let error_code = e.meta().code().unwrap_or("");
assert_eq!("NoSuchKey", error_code);
} else {
panic!("Unexpected error: {e:?}");
}
}
assert_s3_error_code(response1, "NoSuchKey");
// When the object does not exist, the If-None-Match condition should be able to succeed
let response2 = client
let created_data = generate_test_data(1024);
client
.put_object()
.bucket(BUCKET)
.bucket(TEST_BUCKET)
.key(key)
.body(Bytes::from(generate_test_data(1024)).into())
.body(Bytes::from(created_data.clone()).into())
.if_none_match("*")
.send()
.await;
assert!(response2.is_ok());
.await?;
assert_eq!(object_body(&client, key).await?.as_ref(), created_data);
cleanup_objects(&client, BUCKET, &[key]).await;
Ok(())
}
#[tokio::test]
#[ignore = "requires running RustFS server at localhost:9000"]
async fn test_conditional_multi_part_upload() -> Result<(), Box<dyn std::error::Error>> {
let client = create_aws_s3_client().await?;
setup_test_bucket(&client).await?;
async fn test_conditional_multi_part_upload() -> TestResult {
init_logging();
let mut env = RustFSTestEnvironment::new().await?;
env.start_rustfs_server(vec![]).await?;
env.create_test_bucket(TEST_BUCKET).await?;
let client = env.create_s3_client();
let test_key = generate_test_key("multipart-upload-ok");
let test_key = "conditional-multipart-upload";
let test_data = generate_test_data(1024);
let initial_etag = upload_object_with_metadata(&client, BUCKET, &test_key, &test_data).await?;
let initial_etag = upload_object_with_metadata(&client, TEST_BUCKET, test_key, &test_data).await?;
let part_size = 5 * 1024 * 1024; // 5MB per part (minimum for multipart)
let num_parts = 3;
let mut parts = Vec::new();
let mut expected_data = Vec::with_capacity(part_size * usize::try_from(num_parts)?);
// Initiate multipart upload
let initiate_response = client.create_multipart_upload().bucket(BUCKET).key(&test_key).send().await?;
let initiate_response = client
.create_multipart_upload()
.bucket(TEST_BUCKET)
.key(test_key)
.send()
.await?;
let upload_id = initiate_response
.upload_id()
@@ -258,12 +213,13 @@ async fn test_conditional_multi_part_upload() -> Result<(), Box<dyn std::error::
// Upload parts
for part_number in 1..=num_parts {
let part_data = generate_test_data(part_size);
let part_data = vec![u8::try_from(part_number)?; part_size];
expected_data.extend_from_slice(&part_data);
let upload_part_response = client
.upload_part()
.bucket(BUCKET)
.key(&test_key)
.bucket(TEST_BUCKET)
.key(test_key)
.upload_id(upload_id)
.part_number(part_number)
.body(Bytes::from(part_data).into())
@@ -286,57 +242,62 @@ async fn test_conditional_multi_part_upload() -> Result<(), Box<dyn std::error::
// Test 1: Multipart upload with wildcard If-None-Match, should fail
let complete_response = client
.complete_multipart_upload()
.bucket(BUCKET)
.key(&test_key)
.bucket(TEST_BUCKET)
.key(test_key)
.upload_id(upload_id)
.multipart_upload(completed_upload.clone())
.if_none_match("*")
.send()
.await;
assert!(complete_response.is_err());
assert_s3_error_code(complete_response, "PreconditionFailed");
// Test 2: Multipart upload with matching If-None-Match, should fail
let complete_response = client
.complete_multipart_upload()
.bucket(BUCKET)
.key(&test_key)
.bucket(TEST_BUCKET)
.key(test_key)
.upload_id(upload_id)
.multipart_upload(completed_upload.clone())
.if_none_match(initial_etag.clone())
.send()
.await;
assert!(complete_response.is_err());
assert_s3_error_code(complete_response, "PreconditionFailed");
// Test 3: Multipart upload with unmatching If-Match, should fail
let complete_response = client
.complete_multipart_upload()
.bucket(BUCKET)
.key(&test_key)
.bucket(TEST_BUCKET)
.key(test_key)
.upload_id(upload_id)
.multipart_upload(completed_upload.clone())
.if_match("\"abcdef\"")
.send()
.await;
assert!(complete_response.is_err());
assert_s3_error_code(complete_response, "PreconditionFailed");
let staged_parts = client
.list_parts()
.bucket(TEST_BUCKET)
.key(test_key)
.upload_id(upload_id)
.send()
.await?;
assert_eq!(staged_parts.parts().len(), usize::try_from(num_parts)?);
// Test 4: Multipart upload with matching If-Match, should succeed
let complete_response = client
client
.complete_multipart_upload()
.bucket(BUCKET)
.key(&test_key)
.bucket(TEST_BUCKET)
.key(test_key)
.upload_id(upload_id)
.multipart_upload(completed_upload.clone())
.multipart_upload(completed_upload)
.if_match(initial_etag)
.send()
.await;
assert!(complete_response.is_ok());
// Cleanup
cleanup_objects(&client, BUCKET, &[&test_key]).await;
.await?;
assert_eq!(object_body(&client, test_key).await?.as_ref(), expected_data);
Ok(())
}
@@ -9136,7 +9136,6 @@ mod tests {
assert_eq!(loaded.report.scanned, 37);
assert_eq!(loaded.report.eligible, 11);
assert_eq!(loaded.report.enqueued, 5);
assert_eq!(loaded.cursor_revision, Some(1));
assert!(loaded.lease_expires_at_unix_nanos > 0);
let token = loaded
.report
@@ -9155,30 +9154,6 @@ mod tests {
assert_eq!(admission.lease_id, loaded.lease_id);
assert_eq!(admission.lease_expires_at_unix_nanos, loaded.lease_expires_at_unix_nanos);
let mut same_marker_report = report.clone();
same_marker_report.scanned += 1;
persist_manual_transition_page_checkpoint(
&checkpoint_options,
&same_marker_report,
Some("logs/page-end".to_string()),
Some("opaque-next-version".to_string()),
)
.await
.expect("same-marker version checkpoint should persist through the durable progress sink");
let same_marker_checkpointed = load_manual_transition_job_record(ecstore.clone(), job_id)
.await
.expect("same-marker version checkpoint should reload");
assert_eq!(same_marker_checkpointed.cursor_revision, Some(2));
let (_, version_marker) = decode_manual_transition_continuation_token(
same_marker_checkpointed
.report
.continuation_token
.as_deref()
.expect("same-marker version checkpoint should persist a cursor"),
)
.expect("same-marker version cursor should decode");
assert_eq!(version_marker.as_deref(), Some("opaque-next-version"));
create_test_bucket(&ecstore, &bucket).await;
let lifecycle_xml = format!(
r#"<?xml version="1.0" encoding="UTF-8"?>
@@ -9235,7 +9210,6 @@ mod tests {
assert_eq!(checkpointed.report.scanned, 1000);
assert_eq!(checkpointed.report.eligible, 1000);
assert_eq!(checkpointed.report.dry_run_eligible, 1000);
assert_eq!(checkpointed.cursor_revision, Some(3));
let token = checkpointed
.report
.continuation_token
File diff suppressed because it is too large Load Diff
@@ -24,10 +24,6 @@ use crate::bucket::lifecycle::bucket_lifecycle_ops::{
ManualTransitionQueueSnapshot, ManualTransitionRunOptions, ManualTransitionRunReport,
};
use crate::bucket::lifecycle::config_boundary;
use crate::bucket::lifecycle::durable_namespace::{
MANUAL_TRANSITION_JOB_NAMESPACE, MANUAL_TRANSITION_SCOPE_NAMESPACE, MANUAL_TRANSITION_TASK_NAMESPACE,
MANUAL_TRANSITION_WORKER_RESULT_NAMESPACE,
};
use crate::disk::RUSTFS_META_BUCKET;
use crate::error::{Error, Result as EcstoreResult};
use crate::object_api::ObjectOptions;
@@ -38,10 +34,10 @@ use crate::store::ECStore;
pub const MANUAL_TRANSITION_JOB_SCHEMA: &str = "rustfs-manual-transition-job-v1";
pub const MANUAL_TRANSITION_TASK_SCHEMA: &str = "rustfs-manual-transition-task-v1";
pub const MANUAL_TRANSITION_WORKER_RESULT_SCHEMA: &str = "rustfs-manual-transition-worker-result-v1";
pub const MANUAL_TRANSITION_JOB_RECORD_PREFIX: &str = MANUAL_TRANSITION_JOB_NAMESPACE.prefix;
pub const MANUAL_TRANSITION_SCOPE_RECORD_PREFIX: &str = MANUAL_TRANSITION_SCOPE_NAMESPACE.prefix;
pub const MANUAL_TRANSITION_TASK_PREFIX: &str = MANUAL_TRANSITION_TASK_NAMESPACE.prefix;
pub const MANUAL_TRANSITION_WORKER_RESULT_PREFIX: &str = MANUAL_TRANSITION_WORKER_RESULT_NAMESPACE.prefix;
pub const MANUAL_TRANSITION_JOB_RECORD_PREFIX: &str = "ilm/manual-transition/jobs";
pub const MANUAL_TRANSITION_SCOPE_RECORD_PREFIX: &str = "ilm/manual-transition/scopes";
pub const MANUAL_TRANSITION_TASK_PREFIX: &str = "ilm/manual-transition/tasks";
pub const MANUAL_TRANSITION_WORKER_RESULT_PREFIX: &str = "ilm/manual-transition/results";
pub const MAX_MANUAL_TRANSITION_JOB_RECORD_SIZE: usize = 64 * 1024;
pub const MAX_MANUAL_TRANSITION_TASK_RECORD_SIZE: usize = 16 * 1024;
pub const MAX_MANUAL_TRANSITION_WORKER_RESULT_RECORD_SIZE: usize = 8 * 1024;
@@ -199,8 +195,6 @@ pub struct ManualTransitionJobRecord {
pub updated_at_unix_nanos: i128,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub completed_at_unix_nanos: Option<i128>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub cursor_revision: Option<u64>,
pub report: ManualTransitionRunReport,
pub queue_snapshot: ManualTransitionQueueSnapshot,
#[serde(default, skip_serializing_if = "Option::is_none")]
@@ -230,7 +224,6 @@ impl ManualTransitionJobRecord {
created_at_unix_nanos: now,
updated_at_unix_nanos: now,
completed_at_unix_nanos: None,
cursor_revision: Some(0),
report: ManualTransitionRunReport {
bucket: bucket.to_string(),
prefix: options.prefix.clone(),
@@ -245,7 +238,7 @@ impl ManualTransitionJobRecord {
pub fn complete(&mut self, report: ManualTransitionRunReport, queue_snapshot: ManualTransitionQueueSnapshot) {
self.scan_completed = true;
self.merge_scan_report(&report);
self.report.merge_scan_report_preserving_worker(&report);
self.queue_snapshot = queue_snapshot;
self.error = None;
self.mark_terminal_if_worker_drained();
@@ -323,7 +316,7 @@ impl ManualTransitionJobRecord {
}
}
self.queue_snapshot = queue_snapshot;
self.advance_updated_at();
self.updated_at_unix_nanos = OffsetDateTime::now_utc().unix_timestamp_nanos();
self.mark_terminal_if_worker_drained();
}
@@ -366,14 +359,14 @@ impl ManualTransitionJobRecord {
self.report.tier_failure = scan_tier_failure.saturating_add(transition_failed);
self.report.tier_failure_by_reason = scan_tier_failure_by_reason;
self.queue_snapshot = queue_snapshot;
self.advance_updated_at();
self.updated_at_unix_nanos = OffsetDateTime::now_utc().unix_timestamp_nanos();
self.mark_terminal_if_worker_drained();
true
}
pub fn mark_cancel_requested(&mut self) {
self.cancel_requested = true;
self.advance_updated_at();
self.updated_at_unix_nanos = OffsetDateTime::now_utc().unix_timestamp_nanos();
}
pub fn claim_recovery_lease(&mut self, owner_id: impl Into<String>, queue_snapshot: ManualTransitionQueueSnapshot) {
@@ -387,7 +380,7 @@ impl ManualTransitionJobRecord {
pub fn abandon_recovery_lease(&mut self, lease_id: Uuid) {
if self.state == ManualTransitionJobState::Running && self.lease_id == lease_id {
self.lease_expires_at_unix_nanos = 0;
self.advance_updated_at();
self.updated_at_unix_nanos = OffsetDateTime::now_utc().unix_timestamp_nanos();
}
}
@@ -405,7 +398,7 @@ impl ManualTransitionJobRecord {
pub fn renew_lease(&mut self, queue_snapshot: ManualTransitionQueueSnapshot) {
let now = OffsetDateTime::now_utc().unix_timestamp_nanos();
self.updated_at_unix_nanos = self.updated_at_unix_nanos.saturating_add(1).max(now);
self.updated_at_unix_nanos = now;
self.lease_expires_at_unix_nanos = manual_transition_job_lease_expires_at(now);
self.queue_snapshot = queue_snapshot;
}
@@ -445,18 +438,11 @@ impl ManualTransitionJobRecord {
pub fn update_running_progress(&mut self, report: ManualTransitionRunReport, queue_snapshot: ManualTransitionQueueSnapshot) {
if self.state == ManualTransitionJobState::Running {
self.merge_scan_report(&report);
self.report.merge_scan_report_preserving_worker(&report);
self.renew_lease(queue_snapshot);
}
}
fn merge_scan_report(&mut self, report: &ManualTransitionRunReport) {
if self.report.continuation_token != report.continuation_token {
self.cursor_revision = Some(self.cursor_revision.unwrap_or(0).saturating_add(1));
}
self.report.merge_scan_report_preserving_worker(report);
}
pub fn mark_unknown_if_unowned(&mut self) {
if self.state == ManualTransitionJobState::Running {
self.state = ManualTransitionJobState::Unknown;
@@ -477,13 +463,9 @@ impl ManualTransitionJobRecord {
}
fn mark_updated_terminal(&mut self) {
self.advance_updated_at();
self.completed_at_unix_nanos = Some(self.updated_at_unix_nanos);
}
fn advance_updated_at(&mut self) {
let now = OffsetDateTime::now_utc().unix_timestamp_nanos();
self.updated_at_unix_nanos = self.updated_at_unix_nanos.saturating_add(1).max(now);
self.updated_at_unix_nanos = now;
self.completed_at_unix_nanos = Some(now);
}
fn mark_terminal_if_worker_drained(&mut self) {
@@ -1127,8 +1109,7 @@ pub fn manual_transition_scope_record_object_name(scope_key: &str) -> Result<Str
pub async fn save_manual_transition_job_record(api: Arc<ECStore>, job: &ManualTransitionJobRecord) -> EcstoreResult<()> {
let object = manual_transition_job_record_object_name(job.job_id).map_err(manual_transition_job_store_error)?;
let data = job.encode().map_err(manual_transition_job_store_error)?;
config_boundary::save_config(api.clone(), &object, data.clone()).await?;
api.record_durable_ilm_decommission_progress(&object, &data).await
config_boundary::save_config(api, &object, data).await
}
pub async fn load_manual_transition_job_record(api: Arc<ECStore>, job_id: Uuid) -> EcstoreResult<ManualTransitionJobRecord> {
@@ -1161,9 +1142,9 @@ pub async fn save_manual_transition_job_record_if_current(
let object = manual_transition_job_record_object_name(job.job_id).map_err(manual_transition_job_store_error)?;
let data = job.encode().map_err(manual_transition_job_store_error)?;
config_boundary::save_config_with_opts_quiet(
api.clone(),
api,
&object,
data.clone(),
data,
&ObjectOptions {
max_parity: true,
http_preconditions: Some(HTTPPreconditions {
@@ -1173,8 +1154,7 @@ pub async fn save_manual_transition_job_record_if_current(
..Default::default()
},
)
.await?;
api.record_durable_ilm_decommission_progress(&object, &data).await
.await
}
/// Applies a job-record mutation with optimistic concurrency control.
@@ -1612,9 +1592,9 @@ pub async fn save_manual_transition_scope_admission_if_absent(
let object = manual_transition_scope_record_object_name(&admission.scope_key).map_err(manual_transition_job_store_error)?;
let data = serde_json::to_vec(admission).map_err(Error::other)?;
config_boundary::save_config_with_opts(
api.clone(),
api,
&object,
data.clone(),
data,
&ObjectOptions {
max_parity: true,
http_preconditions: Some(HTTPPreconditions {
@@ -1624,8 +1604,7 @@ pub async fn save_manual_transition_scope_admission_if_absent(
..Default::default()
},
)
.await?;
api.record_durable_ilm_decommission_progress(&object, &data).await
.await
}
pub async fn load_manual_transition_scope_admission(
@@ -1663,9 +1642,9 @@ pub async fn save_manual_transition_scope_admission_if_current(
let object = manual_transition_scope_record_object_name(&admission.scope_key).map_err(manual_transition_job_store_error)?;
let data = serde_json::to_vec(admission).map_err(Error::other)?;
match config_boundary::save_config_with_opts(
api.clone(),
api,
&object,
data.clone(),
data,
&ObjectOptions {
max_parity: true,
http_preconditions: Some(HTTPPreconditions {
@@ -1681,8 +1660,7 @@ pub async fn save_manual_transition_scope_admission_if_current(
Err(Error::PreconditionFailed)
}
result => result,
}?;
api.record_durable_ilm_decommission_progress(&object, &data).await
}
}
pub async fn claim_manual_transition_scope_admission(
@@ -1975,15 +1953,13 @@ pub async fn delete_manual_transition_scope_admission_if_current(
job_id: Uuid,
lease_id: Uuid,
) -> EcstoreResult<bool> {
let (admission, etag) = match load_manual_transition_scope_admission_with_etag(api.clone(), scope_key).await {
Ok((admission, etag)) if admission.job_id == job_id && admission.lease_id == lease_id => (admission, etag),
let etag = match load_manual_transition_scope_admission_with_etag(api.clone(), scope_key).await {
Ok((admission, etag)) if admission.job_id == job_id && admission.lease_id == lease_id => etag,
Ok(_) => return Ok(false),
Err(Error::ConfigNotFound) => return Ok(true),
Err(err) => return Err(err),
};
let object = manual_transition_scope_record_object_name(scope_key).map_err(manual_transition_job_store_error)?;
let data = serde_json::to_vec(&admission).map_err(Error::other)?;
api.record_durable_ilm_decommission_terminal(&object, &data).await?;
match config_boundary::delete_config_if_match(api, &object, &etag).await {
Ok(()) | Err(Error::ConfigNotFound) => Ok(true),
Err(Error::PreconditionFailed) => Ok(false),
@@ -2601,25 +2577,6 @@ mod tests {
assert!(decoded.report.tier_failure_by_reason.is_empty());
}
#[test]
fn manual_transition_job_record_decodes_legacy_cursor_without_revision() {
let options = ManualTransitionRunOptions::default();
let record = ManualTransitionJobRecord::new(Uuid::new_v4(), "bucket", &options, TEST_OWNER);
let encoded = record.encode().expect("job record should encode");
let mut value: serde_json::Value = serde_json::from_slice(&encoded).expect("encoded job should be json");
value["job"]
.as_object_mut()
.expect("job should be object")
.remove("cursor_revision");
let record_bytes = serde_json::to_vec(&value["job"]).expect("legacy job should encode");
value["content_sha256"] = serde_json::Value::String(hex_sha256(&record_bytes, ToOwned::to_owned));
let legacy = serde_json::to_vec(&value).expect("legacy envelope should encode");
let decoded = ManualTransitionJobRecord::decode(record.job_id, &legacy).expect("legacy job should decode");
assert_eq!(decoded.cursor_revision, None);
}
#[test]
fn manual_transition_job_record_rejects_unknown_report_fields() {
let options = ManualTransitionRunOptions::default();
@@ -16,7 +16,6 @@ pub mod bucket_lifecycle_audit;
pub mod bucket_lifecycle_ops;
mod config_boundary;
pub mod core;
mod durable_namespace;
pub mod evaluator;
pub mod manual_transition_job;
mod metadata_boundary;
@@ -32,8 +31,3 @@ pub mod tier_free_version_recovery;
pub mod tier_last_day_stats;
pub mod tier_sweeper;
pub mod transition_transaction;
pub(crate) use durable_namespace::{
DurableIlmRecordCheckpoint, ILM_META_PREFIX, ValidatedDurableIlmRecord, classify_durable_ilm_record,
validate_durable_ilm_record,
};
@@ -20,7 +20,6 @@ use tokio_util::sync::CancellationToken;
use tracing::{debug, warn};
use crate::bucket::lifecycle::config_boundary;
use crate::bucket::lifecycle::durable_namespace::TIER_DELETE_JOURNAL_NAMESPACE;
use crate::bucket::lifecycle::runtime_boundary;
use crate::bucket::lifecycle::tier_sweeper::{
Jentry, TierDeleteJournalState, TierDeleteSourceIdentity,
@@ -50,7 +49,7 @@ const TIER_DELETE_JOURNAL_VERSION: u8 = 2;
const TIER_DELETE_JOURNAL_EXACT_VERSION: u8 = 3;
const TIER_DELETE_JOURNAL_STATE_VERSION: u8 = 4;
const TIER_DELETE_JOURNAL_TRANSACTION_VERSION: u8 = 5;
pub(crate) const TIER_DELETE_JOURNAL_PREFIX: &str = TIER_DELETE_JOURNAL_NAMESPACE.prefix;
pub(crate) const TIER_DELETE_JOURNAL_PREFIX: &str = "ilm/tier-delete-journal/";
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
@@ -433,21 +432,6 @@ async fn process_committed_tier_delete_journal_entry(api: Arc<ECStore>, je: &Jen
)
.await?;
}
let path = tier_delete_journal_object_name(je);
let data = encode_tier_delete_journal_entry(je).map_err(std::io::Error::other)?;
let target_pool_indices = api
.record_durable_ilm_decommission_terminal_target_pools(&path, &data)
.await
.map_err(std::io::Error::other)?;
if let Some(target_pool_indices) = target_pool_indices {
for target_pool_idx in target_pool_indices {
match config_boundary::delete_config(api.pools[target_pool_idx].clone(), &path).await {
Ok(()) | Err(Error::ConfigNotFound) => {}
Err(err) => return Err(std::io::Error::other(err)),
}
}
return Ok(());
}
remove_tier_delete_journal_entry(api, je).await
}
@@ -21,7 +21,6 @@ use tracing::{debug, warn};
use uuid::Uuid;
use crate::bucket::lifecycle::config_boundary;
use crate::bucket::lifecycle::durable_namespace::TRANSITION_TRANSACTION_NAMESPACE;
use crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE;
use crate::bucket::lifecycle::tier_sweeper::{
delete_confirmed_transition_candidate_exact_with_lease_idempotent,
@@ -43,7 +42,7 @@ const TRANSITION_TRANSACTION_RECOVERY_INTERVAL: Duration = Duration::from_secs(6
const TRANSITION_TRANSACTION_RECOVERY_TIMEOUT: Duration = Duration::from_secs(300);
pub const TRANSITION_TRANSACTION_SCHEMA: &str = "rustfs-transition-transaction-v1";
pub const TRANSITION_TRANSACTION_PREFIX: &str = "ilm/transition-transactions";
pub const TRANSITION_TRANSACTION_RECORD_PREFIX: &str = TRANSITION_TRANSACTION_NAMESPACE.prefix;
pub const TRANSITION_TRANSACTION_RECORD_PREFIX: &str = "ilm/transition-transactions/records";
pub const MAX_TRANSITION_TRANSACTION_SIZE: usize = 64 * 1024;
pub type Result<T> = std::result::Result<T, TransitionTransactionError>;
@@ -585,8 +584,7 @@ pub(crate) async fn save_transition_transaction_record(
let object =
transition_transaction_record_object_name(transaction.transaction_id).map_err(transition_transaction_store_error)?;
let data = transaction.encode().map_err(transition_transaction_store_error)?;
config_boundary::save_config(api.clone(), &object, data.clone()).await?;
api.record_durable_ilm_decommission_progress(&object, &data).await
config_boundary::save_config(api, &object, data).await
}
pub(crate) async fn load_transition_transaction_record(
@@ -598,14 +596,8 @@ pub(crate) async fn load_transition_transaction_record(
TransitionTransaction::decode(transaction_id, &data).map_err(transition_transaction_store_error)
}
pub(crate) async fn delete_transition_transaction_record(
api: Arc<ECStore>,
transaction: &TransitionTransaction,
) -> EcstoreResult<()> {
let object =
transition_transaction_record_object_name(transaction.transaction_id).map_err(transition_transaction_store_error)?;
let data = transaction.encode().map_err(transition_transaction_store_error)?;
api.record_durable_ilm_decommission_terminal(&object, &data).await?;
pub(crate) async fn delete_transition_transaction_record(api: Arc<ECStore>, transaction_id: Uuid) -> EcstoreResult<()> {
let object = transition_transaction_record_object_name(transaction_id).map_err(transition_transaction_store_error)?;
match config_boundary::delete_config(api, &object).await {
Ok(()) | Err(Error::ConfigNotFound) => Ok(()),
Err(err) => Err(err),
@@ -821,7 +813,7 @@ pub async fn finalize_missing_transition_transaction_for_operator(
if probe != TransitionOperatorProbe::Missing {
return Err(TransitionOperatorError::CandidateNotMissing(probe));
}
delete_transition_transaction_record(api, &transaction)
delete_transition_transaction_record(api, transaction_id)
.await
.map_err(TransitionOperatorError::Store)
}
@@ -857,22 +849,22 @@ pub async fn process_transition_transaction_record(
match transaction.state {
TransitionTransactionState::Uploaded => {
delete_transition_remote_candidate(api.clone(), transaction).await?;
delete_transition_transaction_record(api, transaction).await?;
delete_transition_transaction_record(api, transaction.transaction_id).await?;
Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted)
}
TransitionTransactionState::CleanupPending => match local_commit_matches_transaction(api.clone(), transaction).await {
Ok(true) => {
delete_transition_transaction_record(api, transaction).await?;
delete_transition_transaction_record(api, transaction.transaction_id).await?;
Ok(TransitionTransactionRecoveryOutcome::RecordDeleted)
}
Ok(false) => {
delete_transition_remote_candidate(api.clone(), transaction).await?;
delete_transition_transaction_record(api, transaction).await?;
delete_transition_transaction_record(api, transaction.transaction_id).await?;
Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted)
}
Err(err) if transition_source_is_missing(&err) => {
delete_transition_remote_candidate(api.clone(), transaction).await?;
delete_transition_transaction_record(api, transaction).await?;
delete_transition_transaction_record(api, transaction.transaction_id).await?;
Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted)
}
Err(err) => Err(err),
@@ -880,7 +872,7 @@ pub async fn process_transition_transaction_record(
TransitionTransactionState::LocalCommitStarted => {
match local_commit_matches_transaction(api.clone(), transaction).await {
Ok(true) => {
delete_transition_transaction_record(api, transaction).await?;
delete_transition_transaction_record(api, transaction.transaction_id).await?;
Ok(TransitionTransactionRecoveryOutcome::RecordDeleted)
}
Ok(false) => Ok(TransitionTransactionRecoveryOutcome::Retained),
@@ -889,7 +881,7 @@ pub async fn process_transition_transaction_record(
}
}
TransitionTransactionState::AbortedNoRemote | TransitionTransactionState::Committed => {
delete_transition_transaction_record(api, transaction).await?;
delete_transition_transaction_record(api, transaction.transaction_id).await?;
Ok(TransitionTransactionRecoveryOutcome::RecordDeleted)
}
TransitionTransactionState::UploadOutcomeUnknown => recover_unknown_upload_outcome(api, transaction).await,
@@ -915,7 +907,7 @@ async fn recover_unknown_upload_outcome(
.map_err(Error::other)?
{
TransitionCandidateProbe::Missing => {
delete_transition_transaction_record(api, transaction).await?;
delete_transition_transaction_record(api, transaction.transaction_id).await?;
Ok(TransitionTransactionRecoveryOutcome::RecordDeleted)
}
TransitionCandidateProbe::UnversionedPresent => {
@@ -933,7 +925,7 @@ async fn recover_unknown_upload_outcome(
)
.await
.map_err(Error::other)?;
delete_transition_transaction_record(api, transaction).await?;
delete_transition_transaction_record(api, transaction.transaction_id).await?;
Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted)
}
TransitionCandidateProbe::VersionedPresent(version_id) => {
@@ -966,7 +958,7 @@ async fn cleanup_recovered_unknown_upload_candidate(
.map_err(transition_transaction_store_error)?;
save_transition_transaction_record(api.clone(), &cleanup).await?;
delete_transition_remote_candidate(api.clone(), &cleanup).await?;
delete_transition_transaction_record(api, &cleanup).await?;
delete_transition_transaction_record(api, cleanup.transaction_id).await?;
Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted)
}
-19
View File
@@ -406,25 +406,6 @@ where
Ok(data)
}
pub(crate) async fn read_config_limited_preserve_empty<S>(api: Arc<S>, file: &str, max_bytes: usize) -> Result<Vec<u8>>
where
S: EcstoreObjectIO,
{
let (data, _obj) = read_config_limited_preserve_empty_with_metadata(api, file, max_bytes).await?;
Ok(data)
}
pub(crate) async fn read_config_limited_preserve_empty_with_metadata<S>(
api: Arc<S>,
file: &str,
max_bytes: usize,
) -> Result<(Vec<u8>, ObjectInfo)>
where
S: EcstoreObjectIO,
{
read_config_with_metadata_inner(api, file, &ObjectOptions::default(), true, Some(max_bytes)).await
}
/// Read an existing config object without treating an empty payload as absent.
/// Callers that validate their own payload format need to distinguish corruption
/// from `ConfigNotFound`.
File diff suppressed because it is too large Load Diff
+16 -8
View File
@@ -37,7 +37,7 @@ use crate::bucket::lifecycle::{
transition_transaction::{
TransitionRemoteVersion, TransitionSourceIdentity, TransitionSourceVersionMode, TransitionTransaction,
TransitionTransactionInit, TransitionTransactionState, delete_transition_transaction_record,
load_transition_transaction_record, save_transition_transaction_record,
save_transition_transaction_record,
},
};
use crate::bucket::quota::reservation;
@@ -2124,13 +2124,26 @@ impl SetDisks {
let put_object_size = known_put_object_storage_size(data.size());
let shard_file_size_raw = erasure.shard_file_size(put_object_size);
let is_inline_buffer = storage_class_config.should_inline(shard_file_size_raw, erasure.data_shards, opts.versioned);
let is_inline_buffer =
storage_class_config.should_inline(shard_file_size_raw, erasure.data_shards, opts.versioned);
let collect_stage_timing = rustfs_io_metrics::put_stage_metrics_enabled() || issue3031_diag_enabled();
let shard_file_size = shard_file_size_raw;
let shard_size = erasure.shard_size();
let write_path = classify_put_write_path(is_inline_buffer, put_object_size, fi.erasure.block_size);
let direct_inline_commit = matches!(write_path, SmallWritePath::Inline);
{
use std::io::Write;
let msg = format!(
"INLINE_DEBUG: bucket={} obj={} size={} shard_fs={} ds={} bs={} inline={} direct={} path={} iblock={} ver={}\n",
bucket, object, put_object_size, shard_file_size_raw, erasure.data_shards, fi.erasure.block_size,
is_inline_buffer, direct_inline_commit, write_path.metric_label(), storage_class_config.inline_block(), opts.versioned
);
if let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open("/tmp/rustfs_inline_debug.log") {
let _ = f.write_all(msg.as_bytes());
}
let _ = std::io::stderr().write_all(msg.as_bytes());
}
rustfs_io_metrics::record_put_object_path(write_path.metric_label());
let writer_setup_stage_start = collect_stage_timing.then(Instant::now);
let (mut writers, errors) = if direct_inline_commit {
@@ -4248,12 +4261,7 @@ fn record_transition_uploaded_save_attempt(transaction: &TransitionTransaction,
async fn delete_transition_transaction_if_available(api: Option<&Arc<ECStore>>, transaction_id: Uuid) -> Result<()> {
if let Some(api) = api {
let transaction = match load_transition_transaction_record(api.clone(), transaction_id).await {
Ok(transaction) => transaction,
Err(Error::ConfigNotFound) => return Ok(()),
Err(err) => return Err(err),
};
return delete_transition_transaction_record(api.clone(), &transaction).await;
return delete_transition_transaction_record(api.clone(), transaction_id).await;
}
Ok(())
}
File diff suppressed because it is too large Load Diff
+1 -1
View File
@@ -3194,7 +3194,7 @@ impl ECStore {
// Default return value
let mut del_objects = vec![DeletedObject::default(); objects.len()];
let accounting = vec![None; objects.len()];
let mut accounting = vec![None; objects.len()];
let mut del_errs = Vec::with_capacity(objects.len());
for _ in 0..objects.len() {
@@ -271,7 +271,7 @@ pub(super) fn resolve_latest_object_info_candidates(
.filter(|candidate| latest_candidate_mod_time(candidate) == Some(latest_mod_time))
.collect::<Vec<_>>();
latest_candidates.sort_by_key(|candidate| std::cmp::Reverse(candidate.idx));
latest_candidates.sort_by(|left, right| right.idx.cmp(&left.idx));
let Some(winner) = latest_candidates.first() else {
return Err(Error::ErasureReadQuorum);
+3
View File
@@ -43,6 +43,9 @@ allow-git = [
# RustFS fork carrying presigned expiry and constant-time authentication fixes.
# owner: rustfs-maintainers review: 2026-10
"https://github.com/rustfs/s3s.git",
# MiMalloc fork pinned for hotpath allocation counting support.
# owner: houseme review: 2026-10
"https://github.com/xonatius/mimalloc_rust.git",
]
[bans]
-149
View File
@@ -1,149 +0,0 @@
# CI gate matrix
This file is the source of truth for which validation runs on each event, its
configured wall-clock budget, and whether it can block a merge. Test taxonomy,
naming, and nextest serialization rules remain in [README.md](README.md); e2e
membership and counts remain in
[e2e-suite-inventory.md](e2e-suite-inventory.md).
The distinction between **required** and **report-only** is load-bearing:
a failing job blocks a merge only when its exact check name is present in the
live `main` ruleset. A workflow name, a `merge_group` trigger, or a red PR check
does not make a job required by itself.
## Required merge checks
The live `main` ruleset (`6436880`) currently requires exactly these contexts:
| Required context | Producer | Validation |
|---|---|---|
| `CLA Check` | `.github/workflows/cla.yml` | Contributor agreement |
| `Quick Checks` | `.github/workflows/ci.yml` | Formatting and repository guard scripts |
| `Test and Lint` | `.github/workflows/ci.yml` | Clippy, workspace nextest excluding `e2e_test`, doctests, and migration proofs |
For pull requests limited to the paths excluded by the main CI workflow,
`.github/workflows/ci-docs-only.yml` reports `Quick Checks` and
`Test and Lint` under the same names. It runs the real quick checks and the
planning-document guard; it does not claim that Rust compilation or runtime
tests ran. Despite the workflow name, these paths also include selected deploy,
workflow, and lock files.
Verify the live rule rather than trusting this snapshot before changing merge
policy:
```bash
gh api repos/rustfs/rustfs/rulesets/6436880 \
--jq '.rules[] | select(.type == "required_status_checks") | .parameters'
```
The ruleset currently has `strict_required_status_checks_policy=false`.
`Continuous Integration` accepts `merge_group` events and runs `e2e-full` for
them, but `End-to-End Tests (full merge gate)` is not currently a required
context. Therefore the repository is prepared to test a merge-queue SHA, but
the workflow alone does not prove that every merge passed that lane.
## Pull request and merge matrix
Budgets below are job `timeout-minutes`, not typical runtimes. “Report-only”
means the result is visible and actionable but is not in the live required
context list.
| Event | Validation | Budget | Merge status | Reproduction |
|---|---|---:|---|---|
| PR, non-doc change | `Quick Checks` | 10 min | Required | `make pre-commit` (broader local umbrella) |
| PR, non-doc change | `Test and Lint` | 90 min | Required | `cargo nextest run --profile ci --all --exclude e2e_test` |
| PR, non-doc change | `Typos` | 10 min | Report-only | `typos` |
| PR, non-doc change | `ILM Integration (serial)` | 90 min | Report-only | Use the exact command in `.github/workflows/ci.yml` |
| PR, non-doc change | rio-v2 / swift / sftp test-and-lint variants | 90 min each | Report-only | `cargo nextest run` with the workflow's feature set |
| PR, non-doc change | `Build RustFS Debug Binary` | 30 min | Report-only; prerequisite for black-box lanes | `cargo build -p rustfs --bins` |
| PR, non-doc change | `io_uring Integration (real)` | 30 min | Report-only | `cargo test -p rustfs-ecstore --lib uring_ -- --test-threads=1 --nocapture` |
| PR, non-doc change | `End-to-End Tests` (`e2e-smoke` plus `s3s-e2e`) | 30 min | Report-only | `cargo nextest run --profile e2e-smoke -p e2e_test`; then `./scripts/e2e-run.sh ./target/debug/rustfs <data-dir>` |
| PR, non-doc change | `S3 Implemented Tests` | 60 min | Report-only | Build `rustfs`, then run `scripts/s3-tests/run.sh` with `DEPLOY_MODE=binary`, `TEST_MODE=single`, and `MAXFAIL=0` |
| PR, non-doc change | `S3 Lifecycle Behavior Tests` | 30 min | Report-only | Use the accelerated scanner environment in `.github/workflows/ci.yml` with `scripts/s3-tests/run.sh` |
| PR touching dependency or workflow inputs | Cargo Deny / Workflow Pin Report / Dependency Review | 20 / 5 / 30 min | Report-only | `cargo deny check`; `scripts/security/check_workflow_pins.sh` |
| PR touching architecture rules or architecture docs | `Architecture Migration Rules` | 10 min | Report-only | `scripts/check_architecture_migration_rules.sh` |
| PR touching Nix or workspace manifests | `Nix Build & Check` | 60 min | Report-only | `nix flake check` |
| PR limited to main-CI-excluded paths | companion `Quick Checks` and `Test and Lint` | 10 min each | Required | `git diff --check`; `make doc-paths-check` when documentation paths changed |
| `merge_group` | Standard CI plus `e2e-full` | 55 min for `e2e-full` | Standard required contexts only; `e2e-full` report-only | `cargo nextest run --profile e2e-full -p e2e_test` |
| Push to `main` | Standard CI plus `e2e-full` | 55 min for `e2e-full` | Post-merge detection | Same as `merge_group` |
| PR touching fuzz inputs or harness paths | Build plus five 60-second fuzz smoke targets | 60 min build; 30 min per target | Report-only | `MAX_TOTAL_TIME=60 ./scripts/fuzz/run.sh` |
| PR touching selected ecstore disk/format paths | `Rename Safety` on Windows | 60 min | Report-only | Run the four `cargo test -p rustfs-ecstore --lib <filter>` commands in `windows-filesystem.yml` on Windows |
The authoritative e2e filters live in `.config/nextest.toml`; extend a profile
instead of adding a second ad-hoc selector. Before a profile runs,
`scripts/check_test_wiring.py` compares its exact membership to the committed
digest so a silent test drop fails closed.
## Scheduled and manual validation
Scheduled lanes are independent fault domains. They do not block a pull
request, but their workflow-local gate can fail the run and scheduled failures
are routed to the shared failure-issue action. The scheduled-validation
watchdog and freshness workflow separately detect incomplete runs and missing
schedules.
| Cadence (UTC unless noted) | Workflow / validation | Budget | Verdict and artifacts | Reproduction |
|---|---|---:|---|---|
| Daily 02:17 | Fuzz: five nightly corpus targets | 60 min build; 60 min per target | Gate; corpus/crash artifacts, scheduled failure alert | `MAX_TOTAL_TIME=<seconds> ./scripts/fuzz/run.sh` |
| Daily 03:17 | MinIO interop (EC + SSE read parity) | 40 min | Gate; scheduled failure alert | Dispatch `minio-interop.yml` or follow its pinned Docker fixture steps |
| Daily 04:29 | Replication / cluster-fault / protocol e2e | 45 / 90 / 90 min | Three independent gates; JUnit, membership, and server logs | `cargo nextest run --profile e2e-repl-nightly -p e2e_test`; `--profile e2e-nightly`; `-j 1 --profile e2e-protocols` |
| Daily 06:31 | Warp performance A/B | 180 min | Regression budget gate; A/B summaries and server logs | `bash scripts/run_hotpath_warp_abba.sh --help` |
| Daily 00:07 Asia/Shanghai (16:07 UTC previous day) | Nightly GNU build and Vault lanes | 150 / 90 / 60 min | Build, live Vault, and HA failover gates | Use the commands and pinned Vault images in `nightly-gnu.yml` |
| Daily 03:23 | Security Audit | 20 / 5 min, plus 30 min on PR dependency review | Cargo Deny and workflow-pin gates; scheduled failure alert | `cargo deny check`; `scripts/security/check_workflow_pins.sh` |
| Daily 23:47 | Scheduled Validation Freshness | 10 min | Fails when a critical schedule was never created or is stale | Dispatch `scheduled-validation-freshness.yml` |
| Sunday 00:11 | Full `Continuous Integration` matrix | Per-job budgets above | Weekly variant coverage, including dormant rio-v2 binary/e2e lanes | Dispatch `ci.yml` |
| Sunday 01:13 | Seven-platform build matrix | 150 min per platform | Build/package integrity; scheduled failure alert | Dispatch `build.yml` with an exact platform set |
| Sunday 02:19 | Ceph s3-tests full sweep: single and real four-node, four shards each | 180 min per shard | Compatibility gate; report, JUnit, exact node IDs, and server logs | `scripts/s3-tests/run.sh` against an existing single or distributed target |
| Sunday 06:41 | Mint | 120 min | **Report-only by design**; per-suite PASS/FAIL/NA and raw `log.json` | Reproduce the pinned Docker sequence in `mint.yml` or dispatch it |
| Sunday 07:43 | Workspace line coverage | 120 min | Report-only trend; lcov and JSON retained 90 days | `make coverage` |
| Monthly, day 1 06:37 | Runner Hygiene | 15 min | Validates runner ephemerality; scheduled failure alert | Dispatch `runner-hygiene.yml` |
Manual `workflow_dispatch` exists for the scheduled workflows above. Manual
runs are debugging evidence and intentionally do not open scheduled-failure
issues. A manual performance run may explicitly allow a known regression; that
override must not be treated as an ordinary passing baseline.
## Release validation
Release validation is post-merge and tag-driven; it does not substitute for a
pull-request gate.
| Event | Validation | Budget | Result |
|---|---|---:|---|
| Push to `main` or weekly schedule | `Build and Release` platform matrix | 150 min per platform | Build artifacts for all selected targets; no release publication on a main push |
| Valid release or preview tag | `Build and Release` plus asset checks | 150 min per platform | Draft release, checksummed assets, and publish step |
| Successful non-preview release-tag build | Docker image build and image scan | 60 min build; 30 min scan | Multi-architecture images plus vulnerability report |
| Successful release-tag build | DEB/RPM packaging | 30 min per architecture | Packages and checksum files uploaded to the release |
| Successful non-preview release-tag build | Helm template test and package | 30 min build; 30 min publish | Versioned chart and repository index |
Use an exact preview tag for end-to-end release rehearsal. Manual dispatches
are backfill/debug paths and do not prove the automatic `workflow_run` chain.
## Evidence requirements
A green check is useful only when it proves the intended behavior ran:
- Record the exact commit SHA and run URL.
- Separate product failure from runner prerequisites, service readiness, and
cancellation. Repair the precondition, then rerun the exact workload.
- Preserve membership manifests, JUnit, raw compatibility logs, seeds, and
server logs where the workflow provides them.
- For a bug fix or a new fault checker, provide sensitivity evidence: the old
behavior or an intentional mutation must fail the new oracle, and the fixed
behavior must pass it.
- Never promote a report-only lane to required from one green run. Require at
least 14 days and 30 representative pull requests with at least 99% complete
execution, then update the ruleset and this table together.
## Change checklist
Update this file in the same pull request when any of these change:
- workflow triggers, job names, timeouts, or nextest profile ownership;
- required status contexts or strict/merge-queue policy;
- scheduled cadence, alert routing, artifact contract, or local reproduction;
- report-only versus gating semantics.
Do not copy per-module test counts here. Update
[e2e-suite-inventory.md](e2e-suite-inventory.md) and its enforced membership
digest instead.
+1 -1
View File
@@ -84,7 +84,7 @@
| protocols | 16 | 🌙 |
| quota_test | 14 | |
| reliability_disk_fault_test | 4 | |
| reliant | 25 | 19 ✅ |
| reliant | 29 | 19 ✅ |
| replication_extension_test | 75 | 20 ✅ +55 🌙 |
| security_boundary_test | 4 | |
| server_startup_failfast_test | 1 | |
+2 -2
View File
@@ -336,13 +336,13 @@ opentelemetry = { workspace = true }
tracing-opentelemetry = { workspace = true }
# Data structures
hashbrown = { workspace = true, features = ["serde", "rayon"] }
rustfs-mimalloc = { workspace = true }
mimalloc = { workspace = true }
[target.'cfg(target_os = "linux")'.dependencies]
libsystemd.workspace = true
[target.'cfg(not(target_os = "windows"))'.dependencies]
rustfs-mimalloc-sys.workspace = true
libmimalloc-sys.workspace = true
[dev-dependencies]
uuid = { workspace = true, features = ["v4", "v5", "fast-rng", "macro-diagnostics"] }
+7 -1
View File
@@ -369,8 +369,14 @@ pub fn allocator_reclaim_controller_snapshot(ctx: &CancellationToken) -> Allocat
}
#[cfg(not(target_os = "windows"))]
#[allow(unsafe_code)]
fn collect_allocator_memory(force: bool) -> Result<(), String> {
rustfs_mimalloc::MiMalloc::collect(force);
// SAFETY: `mi_collect` is provided by the active global allocator backend
// on this target family. It is explicitly intended to reclaim retained
// pages/segments and does not require additional invariants from the caller.
unsafe {
libmimalloc_sys::mi_collect(force);
}
Ok(())
}
+8 -10
View File
@@ -26,22 +26,22 @@ struct MiMallocAllocator;
unsafe impl GlobalAlloc for MiMallocAllocator {
unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
// SAFETY: the caller upholds GlobalAlloc's contract for layout.
unsafe { rustfs_mimalloc::MiMalloc.alloc(layout) }
unsafe { mimalloc::MiMalloc.alloc(layout) }
}
unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 {
// SAFETY: the caller upholds GlobalAlloc's contract for layout.
unsafe { rustfs_mimalloc::MiMalloc.alloc_zeroed(layout) }
unsafe { mimalloc::MiMalloc.alloc_zeroed(layout) }
}
unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
// SAFETY: ptr and layout came from this allocator and are forwarded unchanged.
unsafe { rustfs_mimalloc::MiMalloc.dealloc(ptr, layout) }
unsafe { mimalloc::MiMalloc.dealloc(ptr, layout) }
}
unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, new_size: usize) -> *mut u8 {
// SAFETY: ptr and layout came from this allocator and are forwarded unchanged.
unsafe { rustfs_mimalloc::MiMalloc.realloc(ptr, layout, new_size) }
unsafe { mimalloc::MiMalloc.realloc(ptr, layout, new_size) }
}
}
@@ -51,7 +51,7 @@ static GLOBAL: hotpath::CountingAllocator<MiMallocAllocator> = hotpath::Counting
#[cfg(not(all(feature = "hotpath", feature = "hotpath-alloc")))]
#[global_allocator]
static GLOBAL: rustfs_mimalloc::MiMalloc = rustfs_mimalloc::MiMalloc;
static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
fn main() {
let _hotpath_guard = hotpath::HotpathGuardBuilder::new("main").build();
@@ -71,9 +71,8 @@ mod tests {
allocation.extend_from_slice(&[7_u8; 64]);
assert_eq!(allocation.len(), 64);
let heap = rustfs_mimalloc::heap::Heap::main();
// SAFETY: the live Vec pointer is valid to inspect for heap ownership.
assert!(unsafe { heap.contains(allocation.as_ptr()) });
assert!(unsafe { libmimalloc_sys::mi_is_in_heap_region(allocation.as_ptr().cast()) });
}
#[test]
@@ -86,13 +85,12 @@ mod tests {
let layout = Layout::from_size_align(32, 8).expect("valid test allocation layout");
let grown_layout = Layout::from_size_align(64, 8).expect("valid grown test allocation layout");
let allocator = super::MiMallocAllocator;
let heap = rustfs_mimalloc::heap::Heap::main();
// SAFETY: The pointer is checked for null before use and later released
// through the same allocator with the corresponding layout.
let ptr = unsafe { allocator.alloc_zeroed(layout) };
assert!(!ptr.is_null());
assert!(unsafe { heap.contains(ptr) });
assert!(unsafe { libmimalloc_sys::mi_is_in_heap_region(ptr.cast()) });
assert!(unsafe { std::slice::from_raw_parts(ptr, 32).iter().all(|byte| *byte == 0) });
// SAFETY: `ptr` was allocated by `allocator` with `layout`; on failure
@@ -104,7 +102,7 @@ mod tests {
panic!("mimalloc realloc failed in allocator smoke test");
}
assert!(unsafe { heap.contains(grown_ptr) });
assert!(unsafe { libmimalloc_sys::mi_is_in_heap_region(grown_ptr.cast()) });
// SAFETY: `grown_ptr` was reallocated by `allocator` and is released
// with the matching grown layout.
unsafe { allocator.dealloc(grown_ptr, grown_layout) };
+36 -19
View File
@@ -17,7 +17,10 @@ use rustfs_io_metrics::{
record_cpu_usage, record_memory_usage, record_process_memory_split,
};
use serde::Serialize;
#[cfg(any(test, not(target_os = "windows")))]
use serde_json::Value;
#[cfg(not(target_os = "windows"))]
use std::ffi::CStr;
use std::path::Path;
use std::sync::{Arc, Mutex, OnceLock};
use std::time::Duration;
@@ -228,18 +231,7 @@ fn read_cgroup_memory_snapshot() -> Option<CgroupMemorySnapshot> {
read_cgroup_v2().or_else(read_cgroup_v1)
}
fn read_allocator_memory_snapshot() -> Option<AllocatorMemorySnapshot> {
let json = rustfs_mimalloc::MiMalloc::stats_json();
if json.is_empty() {
return None;
}
let observation = parse_mimalloc_stats_json(&json)?;
Some(AllocatorMemorySnapshot {
backend: crate::allocator_reclaim::allocator_backend(),
observation,
})
}
#[cfg(any(test, not(target_os = "windows")))]
fn numeric_json_value(value: &Value) -> Option<u64> {
match value {
Value::Number(number) => number
@@ -250,6 +242,7 @@ fn numeric_json_value(value: &Value) -> Option<u64> {
}
}
#[cfg(any(test, not(target_os = "windows")))]
fn numeric_json_field(value: &Value, field: &str) -> Option<u64> {
match value {
Value::Object(fields) => fields
@@ -261,6 +254,7 @@ fn numeric_json_field(value: &Value, field: &str) -> Option<u64> {
}
}
#[cfg(any(test, not(target_os = "windows")))]
fn mimalloc_stat_field(value: &Value, metric: &str, field: &str) -> Option<u64> {
match value {
Value::Object(fields) => {
@@ -277,10 +271,12 @@ fn mimalloc_stat_field(value: &Value, metric: &str, field: &str) -> Option<u64>
}
}
#[cfg(any(test, not(target_os = "windows")))]
fn mimalloc_stat_current(value: &Value, metric: &str) -> Option<u64> {
mimalloc_stat_field(value, metric, "current")
}
#[cfg(any(test, not(target_os = "windows")))]
fn mimalloc_stat_sum(value: &Value, metrics: &[&str], field: &str) -> Option<u64> {
metrics
.iter()
@@ -289,6 +285,7 @@ fn mimalloc_stat_sum(value: &Value, metrics: &[&str], field: &str) -> Option<u64
.filter(|value| *value > 0)
}
#[cfg(any(test, not(target_os = "windows")))]
fn parse_mimalloc_stats_json(stats_json: &str) -> Option<AllocatorMemoryObservation> {
let value = serde_json::from_str::<Value>(stats_json).ok()?;
let malloc_metrics = ["malloc_normal", "malloc_huge"];
@@ -315,6 +312,33 @@ fn parse_mimalloc_stats_json(stats_json: &str) -> Option<AllocatorMemoryObservat
}
}
#[cfg(not(target_os = "windows"))]
#[allow(unsafe_code)]
fn read_allocator_memory_snapshot() -> Option<AllocatorMemorySnapshot> {
// SAFETY: `mi_stats_get_json` returns a null-terminated JSON buffer owned by
// mimalloc when called with a null input buffer. The mimalloc API requires
// freeing that buffer with `mi_free`; parsing finishes before the buffer is freed.
let observation = unsafe {
let stats_ptr = libmimalloc_sys::mi_stats_get_json(0, std::ptr::null_mut());
if stats_ptr.is_null() {
return None;
}
let observation = CStr::from_ptr(stats_ptr).to_str().ok().and_then(parse_mimalloc_stats_json);
libmimalloc_sys::mi_free(stats_ptr.cast());
observation?
};
Some(AllocatorMemorySnapshot {
backend: crate::allocator_reclaim::allocator_backend(),
observation,
})
}
#[cfg(target_os = "windows")]
fn read_allocator_memory_snapshot() -> Option<AllocatorMemorySnapshot> {
None
}
fn configured_memory_observability_interval_secs() -> u64 {
rustfs_utils::get_env_u64(ENV_MEMORY_OBSERVABILITY_INTERVAL_SECS, DEFAULT_MEMORY_OBSERVABILITY_INTERVAL_SECS).max(1)
}
@@ -542,13 +566,6 @@ mod tests {
assert_eq!(parse_mimalloc_stats_json(r#"{ "allocator": "unknown" }"#), None);
}
#[test]
fn read_allocator_memory_snapshot_uses_mimalloc_stats_json() {
let snapshot = super::read_allocator_memory_snapshot();
#[cfg(not(target_os = "windows"))]
assert!(snapshot.is_some(), "allocator snapshot should be available on non-Windows");
}
#[test]
fn memory_observability_snapshot_reports_disabled_when_metrics_are_disabled() {
let snapshot = build_memory_observability_status_snapshot(false, 15, false);