mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-23 04:39:04 +00:00
Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| f4da36b364 |
@@ -1,2 +1,2 @@
|
||||
sha256-darwin=9f767b37ed8b1c82da62ea441462d75487785c8086e56f08fb6f6cd89c6e2e52
|
||||
sha256-linux=fbdaf42b220958d4b1e8880e0f8b5a7992d38e21051bb60596dd4538424757d6
|
||||
sha256-darwin=03bdfb6a9d6e25d744c385f1461e651f05e4da78e5b0ead2adb3e8b2463e3834
|
||||
sha256-linux=78c46adad135231017fb8679fb91877b5ae3ec6ce1f75d3d045d2c1076c12a49
|
||||
|
||||
@@ -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]]
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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
@@ -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
@@ -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" }
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
|
||||
@@ -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`.
|
||||
|
||||
+44
-1625
File diff suppressed because it is too large
Load Diff
@@ -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
@@ -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);
|
||||
|
||||
@@ -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]
|
||||
|
||||
@@ -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.
|
||||
@@ -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
@@ -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"] }
|
||||
|
||||
@@ -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
@@ -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) };
|
||||
|
||||
@@ -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);
|
||||
|
||||
Reference in New Issue
Block a user