Compare commits

..

17 Commits

Author SHA1 Message Date
houseme 405237ff04 Merge branch 'main' into overtrue/docs-1923-free-version-disposition 2026-08-23 12:38:06 +08:00
Zhengchao An 23a2c7d776 test(kms): stabilize Vault failover validation (#6385)
* test(kms): bound Vault failover progress wait

* ci(nightly): honor manual dispatch ref

* test(kms): preserve Vault worker failures

* test(kms): validate Vault circuit recovery
2026-08-23 12:32:11 +08:00
唐小鸭 5f72209446 fix(ecstore): keep unknown-size sentinel in create_bitrot_writer (#6380)
SSE and compression wrap the payload so its length is unknown and
advertise HashReader::SIZE_PRESERVE_LAYER (-1). Every layer preserved
that sentinel except create_bitrot_writer, which clamped it to 0 before
calling DiskAPI::create_file. RemoteDisk forwards that size verbatim in
the put_file_stream query, so remote peers were told the body was empty.

Since the authenticated put-file trailer (#5868) the receiver used the
declared size to split body from trailer, turning the clamp into a fatal
"auth trailer has trailing data" failure for every SSE PUT on multi-node
deployments (rc.2). #6320 relaxed the receiver to only trust size > 0;
this change fixes the sender so the sentinel survives end to end and the
wire no longer conflates empty objects with unknown-length streams.

Refs #6331
2026-08-23 12:29:52 +08:00
Zhengchao An b6ba89d9e4 docs(testing): document CI gate matrix (#6412) 2026-08-23 12:09:06 +08:00
houseme 648d5166e2 feat(allocator): replace mimalloc/libmimalloc-sys with rustfs-mimalloc/rustfs-mimalloc-sys (#6404)
Replace the upstream xonatius/mimalloc_rust.git fork (mimalloc + libmimalloc-sys)
with the published rustfs-mimalloc (v0.5.0) and rustfs-mimalloc-sys (v0.5.0) crates
from crates.io.

The new crates are based on mimalloc V3 (v3.5.0) and provide:
- MiMalloc global allocator with safe API (collect, stats_json, process_info)
- Heap management and arena operations (heap module)
- Full FFI bindings to mimalloc V3

Changes:
- Workspace deps: mimalloc + libmimalloc-sys (git) → rustfs-mimalloc + rustfs-mimalloc-sys (crates.io)
- allocator_reclaim.rs: libmimalloc_sys::mi_collect → rustfs_mimalloc::MiMalloc::collect
- memory_observability.rs: raw FFI mi_stats_get_json → MiMalloc::stats_json()
- main.rs: heap ownership tests use Heap::contains() (V3 API)
- deny.toml: remove xonatius/mimalloc_rust.git from allow-git

Co-authored-by: heihutu <heihutu@gmail.com>
2026-08-23 12:07:25 +08:00
houseme 84eb5aebef fix(ecstore): remove inline write debug noise (#6408)
* fix(ecstore): remove inline write debug noise

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

* fix(ecstore): satisfy warning-as-error lints

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

---------

Co-authored-by: heihutu <heihutu@gmail.com>
2026-08-23 12:07:20 +08:00
overtrue 76aad3772b test(ecstore): make free-version migration fixtures decode-faithful
Decoded tier free versions always carry both the on-disk
free-version suffix and the in-memory tier marker, and real
decommission targets already contain the bucket volume. Mirror both
invariants in the migration regression fixtures so write-path metadata
validation and rename publication behave exactly as in production.
2026-08-23 11:21:16 +08:00
overtrue 06d00e2502 test(ecstore): cover consumed free-version race 2026-08-23 11:03:38 +08:00
overtrue 3be5efab67 fix(ecstore): handle free-version migration races 2026-08-23 10:52:13 +08:00
overtrue dfbe165882 fix(ecstore): fence free-version decommission replay 2026-08-23 10:34:11 +08:00
overtrue 5360b87459 test(ecstore): cite production free-version regressions 2026-08-23 10:04:32 +08:00
overtrue 646de337c5 fix(ecstore): repair free-version guard visibility 2026-08-23 08:20:47 +08:00
overtrue 7985251b67 fix(ecstore): guard free-version decommission conflicts 2026-08-23 08:01:50 +08:00
overtrue bef7e5bb68 Merge origin/main into overtrue/docs-1923-free-version-disposition 2026-08-23 06:59:55 +08:00
overtrue e1b5c665e8 fix(ecstore): migrate tier free versions during decommission 2026-08-23 06:45:57 +08:00
overtrue a206895fad Merge remote-tracking branch 'origin/main' into overtrue/docs-1923-free-version-disposition 2026-08-22 21:24:49 +08:00
overtrue 3cee88f313 feat(ecstore): account for tier free versions in decommission sweep
Tier free versions (xl.meta cleanup records for deleted transitioned
versions) are not migrated as free versions during decommission: the
exact inventory keeps them inline in versions and the migration loop
routes them through the generic delete-marker path, dropping the flag
and remote-tier identity. Reference audit across GET, heal, ILM,
transition, replication, and restore found no cluster-local consumer
that resolves a free version after decommission; on user-facing delete
paths the remote-delete obligation is also carried by a committed
tier-journal entry, leaving only journal-less records (transition
state unknown) exposed to remote orphaning.

- count and log skipped free versions per decommission entry with
  disposition reason tier_free_version_not_migrated instead of
  omitting them silently
- document free-version lifecycle, non-migration invariant, allowed
  physical-delete timing, and the reference-audit result in
  docs/architecture/decommission-compatibility.md
- state the invariant in doc comments at the filemeta free-version
  sites
- guard the accounting with
  decommission_free_version_accounting_reports_skipped_records

Closes rustfs/backlog#1923
2026-08-22 18:37:54 +08:00
28 changed files with 1714 additions and 240 deletions
+2 -2
View File
@@ -1,2 +1,2 @@
sha256-darwin=af59a0519bfb651bfb6cb57105d7e4eb68bca13cc24887c945908cbf4cb102ff
sha256-linux=5c2d3aeddaaa48e7807039a72e55dad427f971955c8251feec6d0bb66ac25181
sha256-darwin=9f767b37ed8b1c82da62ea441462d75487785c8086e56f08fb6f6cd89c6e2e52
sha256-linux=fbdaf42b220958d4b1e8880e0f8b5a7992d38e21051bb60596dd4538424757d6
+3 -6
View File
@@ -39,11 +39,10 @@ jobs:
env:
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
steps:
- name: Checkout main branch
- name: Checkout repository
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
with:
persist-credentials: false
ref: main
- name: Setup Rust environment
uses: ./.github/actions/setup
@@ -89,11 +88,10 @@ jobs:
# either casing.
NO_PROXY: 127.0.0.1,localhost
steps:
- name: Checkout main branch
- name: Checkout repository
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
with:
persist-credentials: false
ref: main
- name: Setup Rust environment
uses: ./.github/actions/setup
@@ -178,11 +176,10 @@ jobs:
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
NO_PROXY: 127.0.0.1,localhost
steps:
- name: Checkout main branch
- name: Checkout repository
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
with:
persist-credentials: false
ref: main
- name: Setup Rust environment
uses: ./.github/actions/setup
+2 -1
View File
@@ -30,7 +30,8 @@ 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 gates: `.github/workflows/ci.yml` (source of truth; never copy its steps into docs)
- CI workflow steps: `.github/workflows/`; event, timeout, and required-status
matrix: [docs/testing/ci-gates.md](docs/testing/ci-gates.md)
- 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,6 +70,8 @@ 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
+22 -27
View File
@@ -1858,9 +1858,9 @@ dependencies = [
[[package]]
name = "cc"
version = "1.4.3"
version = "1.4.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "509591b7bcd67f4ef775afad7662703b4935daaa6ec0e5605cfb1090b32a2b6d"
checksum = "0ad534f4357a5264cce5019c989cf66a4f0dc4e0d1b1d15f8aacec0ff7360273"
dependencies = [
"find-msvc-tools",
"jobserver",
@@ -2522,12 +2522,6 @@ 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"
@@ -5988,15 +5982,6 @@ 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"
@@ -6397,14 +6382,6 @@ 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"
@@ -9162,13 +9139,11 @@ 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",
@@ -9204,6 +9179,8 @@ dependencies = [
"rustfs-lock",
"rustfs-log-analyzer",
"rustfs-madmin",
"rustfs-mimalloc",
"rustfs-mimalloc-sys",
"rustfs-notify",
"rustfs-object-capacity",
"rustfs-object-data-cache",
@@ -9875,6 +9852,24 @@ 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
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"] }
rustfs-mimalloc = { version = "0.5.0" }
rustfs-mimalloc-sys = { version = "0.5.0" }
hotpath = { version = "0.23.3", default-features = false }
# Snapshot testing for output format regression detection
insta = { version = "1.48" }
+82 -67
View File
@@ -13,37 +13,55 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::common::{RustFSTestEnvironment, init_logging};
use aws_config::meta::region::RegionProviderChain;
use aws_sdk_s3::Client;
use aws_sdk_s3::error::ProvideErrorMetadata;
use aws_sdk_s3::config::{Credentials, Region};
use aws_sdk_s3::types::{
CsvInput, CsvOutput, ExpressionType, FileHeaderInfo, InputSerialization, JsonInput, JsonOutput, JsonType, OutputSerialization,
};
use bytes::Bytes;
use std::error::Error;
use std::time::Duration;
const ENDPOINT: &str = "http://localhost:9000";
const ACCESS_KEY: &str = "rustfsadmin";
const SECRET_KEY: &str = "rustfsadmin";
const BUCKET: &str = "test-sql-bucket";
const CSV_OBJECT: &str = "test-data.csv";
const JSON_OBJECT: &str = "test-data.json";
const SELECT_RESPONSE_TIMEOUT: Duration = Duration::from_secs(30);
type TestResult<T> = Result<T, 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;
async fn create_test_environment() -> TestResult<(RustFSTestEnvironment, Client)> {
init_logging();
let mut env = RustFSTestEnvironment::new().await?;
env.start_rustfs_server(vec![]).await?;
let client = env.create_s3_client();
Ok((env, client))
let client = Client::from_conf(
aws_sdk_s3::Config::from(&shared_config)
.to_builder()
.force_path_style(true) // Important for S3-compatible services
.build(),
);
Ok(client)
}
async fn setup_test_bucket(client: &Client) -> TestResult<()> {
client.create_bucket().bucket(BUCKET).send().await?;
async fn setup_test_bucket(client: &Client) -> Result<(), Box<dyn Error>> {
match client.create_bucket().bucket(BUCKET).send().await {
Ok(_) => {}
Err(e) => {
let error_str = e.to_string();
if !error_str.contains("BucketAlreadyOwnedByYou") && !error_str.contains("BucketAlreadyExists") {
return Err(e.into());
}
}
}
Ok(())
}
async fn upload_test_csv(client: &Client) -> TestResult<()> {
async fn upload_test_csv(client: &Client) -> Result<(), Box<dyn Error>> {
let csv_data = "name,age,city\nAlice,30,New York\nBob,25,Los Angeles\nCharlie,35,Chicago\nDiana,28,Boston";
client
@@ -57,7 +75,7 @@ async fn upload_test_csv(client: &Client) -> TestResult<()> {
Ok(())
}
async fn upload_test_json(client: &Client) -> TestResult<()> {
async fn upload_test_json(client: &Client) -> Result<(), Box<dyn Error>> {
let json_data = r#"{"name":"Alice","age":30,"city":"New York"}
{"name":"Bob","age":25,"city":"Los Angeles"}
{"name":"Charlie","age":35,"city":"Chicago"}
@@ -75,38 +93,33 @@ async fn upload_test_json(client: &Client) -> TestResult<()> {
async fn process_select_response(
mut event_stream: aws_sdk_s3::operation::select_object_content::SelectObjectContentOutput,
) -> TestResult<String> {
tokio::time::timeout(SELECT_RESPONSE_TIMEOUT, async move {
let mut total_data = Vec::new();
let mut saw_end = false;
) -> Result<String, Box<dyn Error>> {
let mut total_data = Vec::new();
while let Some(event) = event_stream.payload.recv().await? {
match event {
aws_sdk_s3::types::SelectObjectContentEventStream::Records(records_event) => {
if let Some(payload) = records_event.payload {
total_data.extend_from_slice(payload.as_ref());
}
while let Ok(Some(event)) = event_stream.payload.recv().await {
match event {
aws_sdk_s3::types::SelectObjectContentEventStream::Records(records_event) => {
if let Some(payload) = records_event.payload {
let data = payload.into_inner();
total_data.extend_from_slice(&data);
}
aws_sdk_s3::types::SelectObjectContentEventStream::End(_) => {
saw_end = true;
break;
}
_ => {}
}
aws_sdk_s3::types::SelectObjectContentEventStream::End(_) => {
break;
}
_ => {
// Handle other event types (Stats, Progress, Cont, etc.)
}
}
}
if !saw_end {
return Err("Select response ended without an End event".into());
}
Ok(String::from_utf8(total_data)?)
})
.await
.map_err(|_| -> Box<dyn Error + Send + Sync> { "Select response timed out".into() })?
Ok(String::from_utf8(total_data)?)
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn test_select_object_content_csv_basic() -> TestResult<()> {
let (_env, client) = create_test_environment().await?;
#[ignore = "requires running RustFS server at localhost:9000"]
async fn test_select_object_content_csv_basic() -> Result<(), Box<dyn Error>> {
let client = create_aws_s3_client().await?;
setup_test_bucket(&client).await?;
upload_test_csv(&client).await?;
@@ -145,8 +158,9 @@ async fn test_select_object_content_csv_basic() -> TestResult<()> {
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn test_select_object_content_csv_aggregation() -> TestResult<()> {
let (_env, client) = create_test_environment().await?;
#[ignore = "requires running RustFS server at localhost:9000"]
async fn test_select_object_content_csv_aggregation() -> Result<(), Box<dyn Error>> {
let client = create_aws_s3_client().await?;
setup_test_bucket(&client).await?;
upload_test_csv(&client).await?;
@@ -189,15 +203,16 @@ async fn test_select_object_content_csv_aggregation() -> TestResult<()> {
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn test_select_object_content_json_basic() -> TestResult<()> {
let (_env, client) = create_test_environment().await?;
#[ignore = "requires running RustFS server at localhost:9000"]
async fn test_select_object_content_json_basic() -> Result<(), Box<dyn Error>> {
let client = create_aws_s3_client().await?;
setup_test_bucket(&client).await?;
upload_test_json(&client).await?;
// Construct JSON query
let sql = "SELECT s.name, s.age FROM S3Object s WHERE s.age > 28";
let json_input = JsonInput::builder().set_type(Some(JsonType::Lines)).build();
let json_input = JsonInput::builder().set_type(Some(JsonType::Document)).build();
let input_serialization = InputSerialization::builder().json(json_input).build();
@@ -229,8 +244,9 @@ async fn test_select_object_content_json_basic() -> TestResult<()> {
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn test_select_object_content_csv_limit() -> TestResult<()> {
let (_env, client) = create_test_environment().await?;
#[ignore = "requires running RustFS server at localhost:9000"]
async fn test_select_object_content_csv_limit() -> Result<(), Box<dyn Error>> {
let client = create_aws_s3_client().await?;
setup_test_bucket(&client).await?;
upload_test_csv(&client).await?;
@@ -270,8 +286,9 @@ async fn test_select_object_content_csv_limit() -> TestResult<()> {
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn test_select_object_content_csv_order_by() -> TestResult<()> {
let (_env, client) = create_test_environment().await?;
#[ignore = "requires running RustFS server at localhost:9000"]
async fn test_select_object_content_csv_order_by() -> Result<(), Box<dyn Error>> {
let client = create_aws_s3_client().await?;
setup_test_bucket(&client).await?;
upload_test_csv(&client).await?;
@@ -301,10 +318,9 @@ async fn test_select_object_content_csv_order_by() -> TestResult<()> {
println!("CSV Order By result: {result_str}");
// Verify ordered by age descending
assert_eq!(
result_str.lines().filter(|line| !line.trim().is_empty()).count(),
2,
"Should return exactly 2 records"
assert!(
result_str.lines().filter(|line| !line.trim().is_empty()).count() >= 2,
"Should return at least 2 records"
);
// Check if contains highest age records
@@ -315,8 +331,9 @@ async fn test_select_object_content_csv_order_by() -> TestResult<()> {
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn test_select_object_content_error_handling() -> TestResult<()> {
let (_env, client) = create_test_environment().await?;
#[ignore = "requires running RustFS server at localhost:9000"]
async fn test_select_object_content_error_handling() -> Result<(), Box<dyn Error>> {
let client = create_aws_s3_client().await?;
setup_test_bucket(&client).await?;
upload_test_csv(&client).await?;
@@ -331,7 +348,7 @@ async fn test_select_object_content_error_handling() -> TestResult<()> {
let output_serialization = OutputSerialization::builder().csv(csv_output).build();
// This query should fail because invalid_column doesn't exist
let error = client
let result = client
.select_object_content()
.bucket(BUCKET)
.key(CSV_OBJECT)
@@ -340,20 +357,18 @@ async fn test_select_object_content_error_handling() -> TestResult<()> {
.input_serialization(input_serialization)
.output_serialization(output_serialization)
.send()
.await
.expect_err("a query referencing an unknown column must fail");
.await;
assert_eq!(
error.as_service_error().and_then(ProvideErrorMetadata::code),
Some("EvaluatorBindingDoesNotExist")
);
// Verify query fails (expected behavior)
assert!(result.is_err(), "Query with invalid column should fail");
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn test_select_object_content_nonexistent_object() -> TestResult<()> {
let (_env, client) = create_test_environment().await?;
#[ignore = "requires running RustFS server at localhost:9000"]
async fn test_select_object_content_nonexistent_object() -> Result<(), Box<dyn Error>> {
let client = create_aws_s3_client().await?;
setup_test_bucket(&client).await?;
// Test query on nonexistent object
@@ -366,7 +381,7 @@ async fn test_select_object_content_nonexistent_object() -> TestResult<()> {
let csv_output = CsvOutput::builder().build();
let output_serialization = OutputSerialization::builder().csv(csv_output).build();
let error = client
let result = client
.select_object_content()
.bucket(BUCKET)
.key("nonexistent.csv")
@@ -375,10 +390,10 @@ async fn test_select_object_content_nonexistent_object() -> TestResult<()> {
.input_serialization(input_serialization)
.output_serialization(output_serialization)
.send()
.await
.expect_err("selecting a missing object must fail");
.await;
assert_eq!(error.as_service_error().and_then(ProvideErrorMetadata::code), Some("NoSuchKey"));
// Verify query fails (expected behavior)
assert!(result.is_err(), "Query on nonexistent object should fail");
Ok(())
}
+263 -13
View File
@@ -13,6 +13,8 @@
// limitations under the License.
use crate::bucket::replication::replication_state_from_filemeta;
#[cfg(test)]
use crate::bucket::utils::is_meta_bucketname;
use crate::bucket::versioning_sys::BucketVersioningSys;
use crate::bucket::{
lifecycle::{
@@ -1347,6 +1349,53 @@ fn should_cleanup_decommission_source_entry(decommissioned: usize, total_version
decommissioned.saturating_add(expired) == total_versions
}
const DECOMMISSION_FREE_VERSION_MIGRATED_REASON: &str = "tier_free_version_migrated";
const DECOMMISSION_FREE_VERSION_CONSUMED_REASON: &str = "tier_free_version_already_consumed";
const DECOMMISSION_FREE_VERSION_RETAINED_REASON: &str = "tier_free_version_migration_failed";
const DECOMMISSION_FREE_VERSION_SWEEP_REASON: &str = "tier_free_version_unresolved_after_decommission";
const DECOMMISSION_FREE_VERSION_DISPOSITION_REASON: &str = "tier_free_version_disposition_recorded";
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
struct DecommissionFreeVersionDisposition {
migrated: usize,
consumed: usize,
retained: usize,
}
impl DecommissionFreeVersionDisposition {
fn record_migrated(&mut self) {
self.migrated += 1;
}
fn record_consumed(&mut self) {
self.consumed += 1;
}
fn record_retained(&mut self) {
self.retained += 1;
}
fn total(self) -> usize {
self.migrated.saturating_add(self.consumed).saturating_add(self.retained)
}
}
enum DecommissionFreeVersionAttempt {
Migrated,
Consumed,
CapacityFailure(Error),
Retry(Error),
}
fn classify_decommission_free_version_attempt(result: Result<()>) -> DecommissionFreeVersionAttempt {
match result {
Ok(()) => DecommissionFreeVersionAttempt::Migrated,
Err(err) if is_decommission_copy_cleanup_safe_error(&err) => DecommissionFreeVersionAttempt::Consumed,
Err(err) if is_decommission_target_capacity_error(&err) => DecommissionFreeVersionAttempt::CapacityFailure(err),
Err(err) => DecommissionFreeVersionAttempt::Retry(err),
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[allow(
dead_code,
@@ -2615,8 +2664,12 @@ fn determine_decommission_final_state(items_failed: usize, was_cancelled: bool)
}
}
fn decommission_remaining_version_count(total_versions: usize, expired: usize) -> usize {
total_versions.saturating_sub(expired)
fn decommission_remaining_version_count(versions: &[rustfs_filemeta::FileInfo], expired: usize) -> usize {
versions
.iter()
.filter(|version| !version.tier_free_version())
.count()
.saturating_sub(expired)
}
fn should_skip_decommission_delete_marker(
@@ -2679,6 +2732,7 @@ fn decommission_remote_tiered_opts(
user_defined: version.metadata.clone(),
src_pool_idx,
data_movement: true,
incl_free_versions: version.tier_free_version(),
include_part_checksums: true,
http_preconditions: Some(crate::data_movement::data_movement_target_precondition()),
expected_bucket_incarnation_id,
@@ -3822,6 +3876,7 @@ impl ECStore {
let mut decommissioned: usize = 0;
let mut expired: usize = 0;
let mut free_version_disposition = DecommissionFreeVersionDisposition::default();
let mut cleanup_preflight_allowed_missing = Vec::new();
for version in fivs.versions.iter() {
@@ -3830,6 +3885,115 @@ impl ECStore {
}
decommission_cancel_signal_result(rx.is_cancelled())?;
if version.tier_free_version() {
let version_id = version.version_id.map(|v| v.to_string());
let mut migration_error = None;
let mut migrated = false;
let mut consumed = false;
let mut capacity_failure = false;
for _ in 0..3 {
match classify_decommission_free_version_attempt(
run_decommission_side_effect(&rx, &operation_gate, || async {
self.decommission_tiered_object(
bucket.as_str(),
&version.name,
version,
&decommission_remote_tiered_opts(
version,
version_id.clone(),
idx,
expected_bucket_incarnation_id,
),
)
.await
})
.await,
) {
DecommissionFreeVersionAttempt::Migrated => {
migrated = true;
migration_error = None;
break;
}
DecommissionFreeVersionAttempt::Consumed => {
consumed = true;
migration_error = None;
break;
}
DecommissionFreeVersionAttempt::CapacityFailure(err) => {
capacity_failure = true;
migration_error = Some(err);
break;
}
DecommissionFreeVersionAttempt::Retry(err) => migration_error = Some(err),
}
}
{
let mut pool_meta = self.pool_meta.write().await;
ensure_decommission_generation(&pool_meta, idx, generation)?;
if let Err(err) = count_decommission_item(&mut pool_meta, idx, 0, !migrated && !consumed) {
return Err(with_decommission_entry_context(
"count_decommission_item",
bucket.as_str(),
entry.name.as_str(),
err,
));
}
}
if migrated || consumed {
decommissioned += 1;
cleanup_preflight_allowed_missing.push(data_movement::source_cleanup_version_identity(version));
}
if migrated {
free_version_disposition.record_migrated();
} else if consumed {
free_version_disposition.record_consumed();
} else {
free_version_disposition.record_retained();
}
debug!(
event = EVENT_DECOMMISSION_ENTRY,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_POOLS,
pool_index = idx,
bucket = %bucket,
object = %version.name,
version_id = ?version_id,
result = ?migration_error,
reason = if migrated {
DECOMMISSION_FREE_VERSION_MIGRATED_REASON
} else if consumed {
DECOMMISSION_FREE_VERSION_CONSUMED_REASON
} else {
DECOMMISSION_FREE_VERSION_RETAINED_REASON
},
state = if migrated {
"free_version_migrated"
} else if consumed {
"free_version_consumed"
} else {
"free_version_retained"
},
"Decommission free-version disposition recorded"
);
if capacity_failure {
return Err(with_decommission_entry_context(
"decommission_tier_free_version",
bucket.as_str(),
version.name.as_str(),
migration_error.expect("capacity failure must retain its error"),
));
}
if !migrated && !consumed {
break;
}
continue;
}
if run_decommission_side_effect(&rx, &operation_gate, || async {
should_skip_lifecycle_for_data_movement(
self.clone(),
@@ -3850,7 +4014,7 @@ impl ECStore {
continue;
}
let remaining_versions = decommission_remaining_version_count(fivs.versions.len(), expired);
let remaining_versions = decommission_remaining_version_count(&fivs.versions, expired);
if should_skip_decommission_delete_marker(version, remaining_versions, replication_config.is_some()) {
//
decommissioned += 1;
@@ -4129,6 +4293,24 @@ impl ECStore {
}
}
if free_version_disposition.total() > 0 {
debug!(
event = EVENT_DECOMMISSION_ENTRY,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_POOLS,
pool_index = idx,
bucket = %bucket,
object = %entry.name,
free_versions_migrated = free_version_disposition.migrated,
free_versions_consumed = free_version_disposition.consumed,
free_versions_retained = free_version_disposition.retained,
free_versions_total = free_version_disposition.total(),
reason = DECOMMISSION_FREE_VERSION_DISPOSITION_REASON,
state = "free_version_disposition",
"Decommission free-version disposition summary"
);
}
if should_cleanup_decommission_source_entry(decommissioned, fivs.versions.len(), expired) {
if bucket_incarnation_fence.as_ref().is_some_and(|guard| guard.is_lock_lost()) {
return Err(Error::other("decommission bucket incarnation fence was lost before source cleanup"));
@@ -4298,6 +4480,39 @@ impl ECStore {
.await
}
#[cfg(test)]
pub(crate) async fn decommission_entry_for_test_with_bucket_incarnation(
self: &Arc<Self>,
idx: usize,
entry: MetaCacheEntry,
bucket: String,
set: Arc<SetDisks>,
) -> Result<()> {
let expected_bucket_incarnation_id = if is_meta_bucketname(&bucket) {
None
} else {
Some(self.bucket_incarnation_id_from_disk(&bucket).await?)
};
self.decommission_entry(
CancellationToken::new(),
idx,
OffsetDateTime::now_utc(),
entry,
bucket,
set,
None,
None,
None,
expected_bucket_incarnation_id,
)
.await
}
#[cfg(test)]
pub(crate) async fn check_after_decommission_for_test(self: &Arc<Self>, idx: usize) -> Result<()> {
self.check_after_decommission(idx).await
}
#[tracing::instrument(skip(self, rx))]
async fn decommission_pool(
self: &Arc<Self>,
@@ -5162,6 +5377,7 @@ impl ECStore {
let lifecycle_config_cb = lifecycle_config.clone();
let object_lock_config_cb = object_lock_config.clone();
let store = Arc::clone(self);
let set_cb = Arc::clone(set);
let callback_rx_cb = callback_rx.clone();
let callback: ListCallback = Arc::new(move |entry: MetaCacheEntry| {
@@ -5171,6 +5387,7 @@ impl ECStore {
let lifecycle_config = lifecycle_config_cb.clone();
let object_lock_config = object_lock_config_cb.clone();
let store = Arc::clone(&store);
let set = Arc::clone(&set_cb);
let callback_rx = callback_rx_cb.clone();
Box::pin(async move {
if callback_rx.is_cancelled() {
@@ -5185,11 +5402,14 @@ impl ECStore {
return;
}
let fivs = match load_decommission_entry_versions(
let fivs = match load_decommission_entry_exact_versions(
&set,
&entry,
&bucket_name,
"check_after_decommission.file_info_versions",
) {
)
.await
{
Ok(fivs) => fivs,
Err(err) => {
let mut first_err = entry_error.lock().await;
@@ -5202,7 +5422,23 @@ impl ECStore {
};
let mut remaining = 0;
for version in &fivs.versions {
for version in fivs.versions.iter().chain(fivs.free_versions.iter()) {
if version.tier_free_version() {
remaining += 1;
debug!(
event = EVENT_DECOMMISSION_ENTRY,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_POOLS,
pool_index = idx,
bucket = %bucket_name,
object = %entry.name,
version_id = ?version.version_id,
reason = DECOMMISSION_FREE_VERSION_SWEEP_REASON,
state = "free_version_retained",
"Decommission final sweep retained a free version"
);
continue;
}
if version.deleted {
continue;
}
@@ -5317,13 +5553,6 @@ mod tests {
assert_eq!(determine_decommission_final_state(0, true), DecommissionFinalState::Failed);
}
#[test]
fn decommission_remaining_version_count_excludes_only_expired_versions() {
assert_eq!(decommission_remaining_version_count(1, 0), 1);
assert_eq!(decommission_remaining_version_count(2, 1), 1);
assert_eq!(decommission_remaining_version_count(1, 1), 0);
}
#[test]
fn lifecycle_action_removes_data_movement_version_rejects_delete_marker_action() {
assert!(!lifecycle_action_removes_data_movement_version(IlmAction::DeleteAction));
@@ -5380,6 +5609,21 @@ mod tests {
)));
}
#[test]
fn decommission_free_version_attempt_treats_missing_source_as_consumed() {
let attempt =
classify_decommission_free_version_attempt(Err(Error::ObjectNotFound("bucket".to_string(), "object".to_string())));
assert!(matches!(attempt, DecommissionFreeVersionAttempt::Consumed));
}
#[test]
fn decommission_free_version_attempt_preserves_capacity_failure() {
let attempt = classify_decommission_free_version_attempt(Err(Error::DiskFull));
assert!(matches!(attempt, DecommissionFreeVersionAttempt::CapacityFailure(Error::DiskFull)));
}
#[test]
fn decommission_delete_marker_copy_error_rejects_data_movement_overwrite() {
let err = Error::DataMovementOverwriteErr("bucket".to_string(), "object".to_string(), "version".to_string());
@@ -5561,6 +5805,12 @@ mod tests {
assert!(opts.include_part_checksums);
assert!(opts.http_preconditions.is_some());
assert_eq!(opts.expected_bucket_incarnation_id, Some(incarnation));
assert!(!opts.incl_free_versions);
let mut free_version = version;
free_version.set_tier_free_version();
let free_opts = decommission_remote_tiered_opts(&free_version, Some("free-version-id".to_string()), 9, Some(incarnation));
assert!(free_opts.incl_free_versions);
}
#[test]
+13
View File
@@ -1950,6 +1950,19 @@ mod tests {
assert!(source_cleanup_versions_match_with_allowed_missing(&expected, &current, &allowed_missing));
}
#[test]
fn test_decommission_cleanup_preflight_accepts_migrated_free_version_consumed_from_source() {
let migrated = cleanup_test_file_info("object.txt", Uuid::from_u128(1), "migrated");
let mut free_version = cleanup_test_file_info("object.txt", Uuid::from_u128(2), "tier-cleanup");
free_version.deleted = true;
free_version.set_tier_free_version();
let expected = cleanup_test_versions(vec![migrated.clone(), free_version.clone()]);
let current = cleanup_test_versions(vec![migrated]);
let allowed_missing = vec![source_cleanup_version_identity(&free_version)];
assert!(source_cleanup_versions_match_with_allowed_missing(&expected, &current, &allowed_missing));
}
#[test]
fn test_decommission_cleanup_preflight_rejects_unexpected_missing_version() {
let migrated = cleanup_test_file_info("object.txt", Uuid::from_u128(1), "migrated");
+38 -6
View File
@@ -784,6 +784,24 @@ pub(crate) fn create_deferred_bitrot_reader_with_stripe_handle(
///
/// # Returns
/// A Result containing the BitrotWriterWrapper or an error
/// Size hint handed to `DiskAPI::create_file` for a bitrot-wrapped shard.
///
/// A known length is grown by one checksum per shard so the on-disk file size
/// matches what the bitrot writer emits. A negative length is the
/// unknown-size sentinel (`HashReader::SIZE_PRESERVE_LAYER`, used by SSE and
/// compression) and must be preserved: `RemoteDisk::create_file` forwards it
/// in the `put_file_stream` query, and the receiver only treats `size > 0` as
/// a fixed body length when locating the authenticated trailer. Clamping it
/// to `0` would claim an empty body and misframe the stream. `0` stays `0`
/// because a genuinely empty object still means an empty body.
fn bitrot_create_file_size(length: i64, shard_size: usize, checksum_algo: &HashAlgorithm) -> i64 {
if length <= 0 {
return length;
}
let length = length as usize;
(length.div_ceil(shard_size) * checksum_algo.size() + length) as i64
}
pub async fn create_bitrot_writer(
is_inline_buffer: bool,
disk: Option<&DiskStore>,
@@ -796,12 +814,7 @@ pub async fn create_bitrot_writer(
let writer = if is_inline_buffer {
CustomWriter::new_inline_buffer()
} else if let Some(disk) = disk {
let length = if length > 0 {
let length = length as usize;
(length.div_ceil(shard_size) * checksum_algo.size() + length) as i64
} else {
0
};
let length = bitrot_create_file_size(length, shard_size, &checksum_algo);
let file = disk.create_file("", volume, path, length).await?;
#[cfg(feature = "hotpath")]
@@ -820,6 +833,25 @@ mod tests {
use rustfs_rio::ChunkReader;
use std::collections::VecDeque;
#[test]
fn bitrot_create_file_size_grows_known_length_by_checksums() {
// 10 bytes over 4-byte shards = 3 shards, each followed by a 32-byte hash.
assert_eq!(bitrot_create_file_size(10, 4, &HashAlgorithm::HighwayHash256), 10 + 3 * 32);
assert_eq!(bitrot_create_file_size(10, 4, &HashAlgorithm::None), 10);
}
#[test]
fn bitrot_create_file_size_keeps_empty_and_unknown_distinct() {
assert_eq!(bitrot_create_file_size(0, 4, &HashAlgorithm::HighwayHash256), 0);
// SSE/compression streams advertise SIZE_PRESERVE_LAYER (-1); the remote
// put_file_stream receiver relies on a non-positive size to parse the auth
// trailer from the stream tail, so the sentinel must survive untouched.
assert_eq!(
bitrot_create_file_size(rustfs_rio::HashReader::SIZE_PRESERVE_LAYER, 4, &HashAlgorithm::HighwayHash256),
rustfs_rio::HashReader::SIZE_PRESERVE_LAYER
);
}
struct TestChunkReader {
chunks: VecDeque<Bytes>,
}
+297
View File
@@ -227,6 +227,70 @@ pub(super) fn restore_commit_operation_id_from_metadata(metadata: &HashMap<Strin
restore_operation_id_from_metadata(metadata)
}
async fn inspect_decommission_tier_free_version_target(
disk: &DiskStore,
bucket: &str,
object: &str,
source: &FileInfo,
) -> Result<bool> {
let raw = match disk.read_xl(bucket, object, false).await {
Ok(raw) => raw,
Err(DiskError::FileNotFound | DiskError::FileVersionNotFound | DiskError::VolumeNotFound) => return Ok(false),
Err(err) => return Err(err.into()),
};
let meta = FileMeta::load(&raw.buf)?;
let source_version_id = source.version_id.filter(|version_id| !version_id.is_nil());
let mut matching_count = 0;
let mut all_matching_versions_equivalent = true;
for existing in meta
.versions
.iter()
.filter(|version| version.header.version_id.filter(|version_id| !version_id.is_nil()) == source_version_id)
{
matching_count += 1;
let existing = existing.into_fileinfo(bucket, object, true)?;
existing.validate_for_metadata_read()?;
if !existing.tier_free_version() || !crate::store::tiered_data_movement_source_matches(source, &existing)? {
all_matching_versions_equivalent = false;
}
}
if matching_count == 0 {
return Ok(false);
}
if matching_count == 1 && all_matching_versions_equivalent {
return Ok(true);
}
Err(StorageError::DataMovementOverwriteErr(
bucket.to_owned(),
object.to_owned(),
source_version_id.map(|version_id| version_id.to_string()).unwrap_or_default(),
)
.into())
}
fn ensure_decommission_tier_free_version_commit_fence(bucket: &str, object: &str, opts: &ObjectOptions) -> Result<()> {
if opts
.namespace_lock_fence
.as_ref()
.is_some_and(NamespaceLockFence::is_lock_lost)
|| opts
.bucket_lifecycle_lock_fence
.as_ref()
.is_some_and(NamespaceLockFence::is_lock_lost)
{
return Err(StorageError::NamespaceLockQuorumUnavailable {
mode: "decommission_tier_free_version_commit",
bucket: bucket.to_string(),
object: object.to_string(),
required: 1,
achieved: 0,
});
}
Ok(())
}
impl SetDisks {
pub(super) async fn require_current_restore_operation_id(
&self,
@@ -4665,6 +4729,98 @@ fn resolve_delete_version_state(opts: &ObjectOptions, goi: &ObjectInfo, version_
}
impl SetDisks {
/// Publish an internal tier free-version record without changing its
/// delete-marker shape or remote-tier identity. The caller holds the
/// source and target object locks; a write quorum is required before the
/// source cleanup may remove the original record.
#[tracing::instrument(skip(self, fi, opts))]
pub(crate) async fn decommission_tier_free_version(
&self,
bucket: &str,
object: &str,
fi: &FileInfo,
opts: &ObjectOptions,
) -> Result<()> {
if !fi.deleted || !fi.tier_free_version() {
return Err(Error::other("decommission tier free-version write requires a free version record"));
}
ensure_decommission_tier_free_version_commit_fence(bucket, object, opts)?;
self.validate_decommission_tier_free_version_target(bucket, object, fi)
.await?;
ensure_decommission_tier_free_version_commit_fence(bucket, object, opts)?;
let disks = self.disks.read().await.clone();
let write_quorum = self.default_write_quorum();
let futures = disks.into_iter().map(|disk| {
let file_info = fi.clone();
async move {
if let Some(disk) = disk {
disk.write_metadata("", bucket, object, file_info).await
} else {
Err(DiskError::DiskNotFound)
}
}
});
let mut errs = Vec::new();
for result in join_all(futures).await {
match result {
Ok(_) => errs.push(None),
Err(err) => errs.push(Some(err)),
}
}
ensure_decommission_tier_free_version_commit_fence(bucket, object, opts)?;
resolve_tiered_decommission_write_quorum_result(&errs, write_quorum, bucket, object)
}
pub(crate) async fn validate_decommission_tier_free_version_target(
&self,
bucket: &str,
object: &str,
fi: &FileInfo,
) -> Result<()> {
// The caller holds the source and target object locks. Inspect every
// target disk before an idempotent return or metadata fan-out so a
// sub-quorum conflict cannot be hidden by a successful quorum.
let disks = self.disks.read().await.clone();
let preflight = disks
.iter()
.flatten()
.map(|disk| inspect_decommission_tier_free_version_target(disk, bucket, object, fi));
for result in join_all(preflight).await {
result?;
}
Ok(())
}
pub(crate) async fn has_decommission_tier_free_version_write_quorum(
&self,
bucket: &str,
object: &str,
fi: &FileInfo,
opts: &ObjectOptions,
) -> Result<bool> {
ensure_decommission_tier_free_version_commit_fence(bucket, object, opts)?;
let disks = self.disks.read().await.clone();
let preflight = disks.iter().map(|disk| async {
match disk {
Some(disk) => inspect_decommission_tier_free_version_target(disk, bucket, object, fi).await,
None => Ok(false),
}
});
let mut equivalent = 0;
for result in join_all(preflight).await {
if result? {
equivalent += 1;
}
}
ensure_decommission_tier_free_version_commit_fence(bucket, object, opts)?;
Ok(equivalent >= self.default_write_quorum())
}
#[tracing::instrument(skip(self, fi, opts))]
pub(crate) async fn decommission_tiered_object(
&self,
@@ -9887,6 +10043,147 @@ mod tests {
assert_ne!(updated.erasure.distribution, original.erasure.distribution);
}
#[tokio::test]
async fn decommission_tier_free_version_preserves_remote_identity() {
let set_disks = make_local_bucket_test_set_disks().await;
let bucket = "free-version-decommission";
let object = "object.txt";
set_disks
.make_bucket(bucket, &MakeBucketOptions::default())
.await
.expect("target bucket should exist before free-version migration");
let version_id = Uuid::new_v4();
let mut free_version = FileInfo {
name: object.to_string(),
volume: bucket.to_string(),
version_id: Some(version_id),
mod_time: Some(time::OffsetDateTime::now_utc()),
deleted: true,
transition_tier: "WARM-TIER".to_string(),
transitioned_objname: "remote/object".to_string(),
..Default::default()
};
free_version.set_tier_free_version();
// Decoded free versions always carry the on-disk free-version
// suffix alongside the in-memory tier marker; mirror that here so
// the record satisfies delete-marker metadata validation.
rustfs_utils::http::metadata_compat::insert_str(
&mut free_version.metadata,
rustfs_utils::http::metadata_compat::SUFFIX_FREE_VERSION,
String::new(),
);
set_disks
.decommission_tier_free_version(bucket, object, &free_version, &ObjectOptions::default())
.await
.expect("free-version metadata should reach the target quorum");
set_disks
.decommission_tier_free_version(bucket, object, &free_version, &ObjectOptions::default())
.await
.expect("replaying the same free-version metadata should be idempotent");
let versions = set_disks
.load_file_info_versions_exact(bucket, object)
.await
.expect("migrated free-version metadata should decode")
.expect("migrated free-version metadata should exist");
let migrated = versions
.versions
.iter()
.find(|version| version.version_id == Some(version_id))
.expect("free version should be present on the target");
assert_eq!(
versions
.versions
.iter()
.filter(|version| version.version_id == Some(version_id))
.count(),
1
);
assert!(migrated.tier_free_version());
assert_eq!(migrated.transition_tier, "WARM-TIER");
assert_eq!(migrated.transitioned_objname, "remote/object");
}
#[tokio::test]
async fn decommission_tier_free_version_resume_requires_write_quorum() {
let set_disks = make_local_bucket_test_set_disks_with_drive_count(4).await;
let bucket = "free-version-decommission-resume";
let object = "object.txt";
set_disks
.make_bucket(bucket, &MakeBucketOptions::default())
.await
.expect("target bucket should exist before free-version migration");
let mut free_version = FileInfo {
name: object.to_string(),
volume: bucket.to_string(),
version_id: Some(Uuid::new_v4()),
mod_time: Some(time::OffsetDateTime::now_utc()),
deleted: true,
transition_tier: "WARM-TIER".to_string(),
transitioned_objname: "remote/object".to_string(),
..Default::default()
};
free_version.set_tier_free_version();
// Decoded free versions always carry the on-disk free-version
// suffix alongside the in-memory tier marker; mirror that here so
// the record satisfies delete-marker metadata validation.
rustfs_utils::http::metadata_compat::insert_str(
&mut free_version.metadata,
rustfs_utils::http::metadata_compat::SUFFIX_FREE_VERSION,
String::new(),
);
let opts = ObjectOptions::default();
let disks = set_disks.get_disks_internal().await;
for disk in disks.iter().take(2).flatten() {
disk.write_metadata("", bucket, object, free_version.clone())
.await
.expect("partial first attempt should leave equivalent metadata");
}
assert!(
!set_disks
.has_decommission_tier_free_version_write_quorum(bucket, object, &free_version, &opts)
.await
.expect("partial target metadata should remain valid"),
"write-quorum-minus-one must not be accepted as an idempotent migration"
);
disks[2]
.as_ref()
.expect("third target disk should be online")
.write_metadata("", bucket, object, free_version.clone())
.await
.expect("third equivalent target write should complete quorum");
assert!(
set_disks
.has_decommission_tier_free_version_write_quorum(bucket, object, &free_version, &opts)
.await
.expect("write-quorum target metadata should remain valid")
);
}
#[test]
fn decommission_tier_free_version_commit_rejects_lost_fence() {
let opts = ObjectOptions {
namespace_lock_fence: Some(NamespaceLockFence::lost_for_test()),
..Default::default()
};
let err = ensure_decommission_tier_free_version_commit_fence("bucket", "object", &opts)
.expect_err("lost target lock must fail the free-version commit");
assert!(matches!(
err,
Error::NamespaceLockQuorumUnavailable {
mode: "decommission_tier_free_version_commit",
required: 1,
achieved: 0,
..
}
));
}
#[test]
fn test_resolve_tiered_decommission_write_quorum_result_allows_successful_quorum() {
let errs = vec![None, None, Some(DiskError::DiskNotFound), None];
+1 -14
View File
@@ -2124,26 +2124,13 @@ 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 {
+450
View File
@@ -553,6 +553,8 @@ mod tests {
should_retry_format_load, should_retry_local_decommission_resume, wait_for_local_decommission_resume_delay,
};
#[cfg(feature = "test-util")]
use crate::disk::DiskAPI;
#[cfg(feature = "test-util")]
use crate::{
bucket::lifecycle::{
lifecycle::{TRANSITION_PENDING, TransitionOptions},
@@ -1307,6 +1309,77 @@ mod tests {
});
}
#[cfg(feature = "test-util")]
async fn seed_transitioned_free_version(
ctx: &Arc<crate::runtime::instance::InstanceContext>,
store: &Arc<crate::store::ECStore>,
bucket: &str,
object: &str,
) -> (uuid::Uuid, uuid::Uuid) {
let tier_name = format!("DECOMFREE{}", uuid::Uuid::new_v4().simple());
register_mock_tier(&ctx.tier_config_mgr(), &tier_name).await;
let mut reader = PutObjReader::from_vec(b"transitioned source bytes".to_vec());
let source = store.pools[0]
.put_object(
bucket,
object,
&mut reader,
&ObjectOptions {
versioned: true,
..Default::default()
},
)
.await
.expect("write transitioned decommission source");
let source_version = source.version_id.expect("transitioned source must be versioned");
store.pools[0]
.transition_object(
bucket,
object,
&ObjectOptions {
versioned: true,
version_id: Some(source_version.to_string()),
transition: TransitionOptions {
status: TRANSITION_PENDING.to_string(),
tier: tier_name,
etag: source.etag.clone().expect("transitioned source must have an ETag"),
..Default::default()
},
mod_time: source.mod_time,
..Default::default()
},
)
.await
.expect("transition source before decommission");
store.pools[0]
.delete_object(
bucket,
object,
ObjectOptions {
versioned: true,
version_id: Some(source_version.to_string()),
..Default::default()
},
)
.await
.expect("delete transitioned source version");
let versions = store.pools[0]
.get_disks_by_key(object)
.load_file_info_versions_exact(bucket, object)
.await
.expect("source versions should decode after transition delete")
.expect("source free version should remain after transition delete");
let free_version = versions
.versions
.iter()
.find(|version| version.tier_free_version())
.and_then(|version| version.version_id)
.expect("transition delete should create a free version");
(source_version, free_version)
}
async fn write_decommission_test_multipart_source(
store: &Arc<crate::store::ECStore>,
pool_idx: usize,
@@ -3846,6 +3919,383 @@ mod tests {
shutdown.cancel();
}
#[cfg(feature = "test-util")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial_test::serial(storage_class_env)]
async fn decommission_entry_skips_cleanup_only_marker_when_free_version_is_present() {
let temp_dir = tempfile::tempdir().expect("create free-version decommission store dir");
let (ctx, store, shutdown) =
without_storage_class_env(build_isolated_test_store(temp_dir.path(), "decommission-free-marker", &[4, 4])).await;
crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
let bucket = format!("decom-free-marker-{}", uuid::Uuid::new_v4());
let object = "free-marker-object";
store
.make_bucket(&bucket, &MakeBucketOptions::default())
.await
.expect("create free-version decommission bucket");
let (_, free_version) = seed_transitioned_free_version(&ctx, &store, &bucket, object).await;
let source_free = store.pools[0]
.get_disks_by_key(object)
.load_file_info_versions_exact(&bucket, object)
.await
.expect("source free-version metadata should decode")
.and_then(|versions| {
versions
.versions
.into_iter()
.find(|version| version.version_id == Some(free_version) && version.tier_free_version())
})
.expect("source free-version identity should be present before decommission");
let mut target_reader = PutObjReader::from_vec(b"target ordinary bytes".to_vec());
let target_w = store.pools[1]
.put_object(
&bucket,
object,
&mut target_reader,
&ObjectOptions {
versioned: true,
..Default::default()
},
)
.await
.expect("write unrelated target version");
let target_w_version = target_w.version_id.expect("target version should have an id");
let marker = store.pools[0]
.delete_object(
&bucket,
object,
ObjectOptions {
versioned: true,
..Default::default()
},
)
.await
.expect("write cleanup-only delete marker");
assert!(marker.delete_marker);
mark_test_pool_decommissioning(&store, 0).await;
let source_set = store.pools[0].get_disks_by_key(object);
store
.decommission_entry_for_test_with_bucket_incarnation(
0,
MetaCacheEntry {
name: object.to_string(),
..Default::default()
},
bucket.clone(),
source_set.clone(),
)
.await
.expect("real decommission entry should migrate the free version");
let target_versions = store.pools[1]
.get_disks_by_key(object)
.load_file_info_versions_exact(&bucket, object)
.await
.expect("target versions should decode")
.expect("target free version should be present");
assert!(
target_versions
.versions
.iter()
.any(|version| { version.version_id == Some(free_version) && version.tier_free_version() })
);
let migrated_free = target_versions
.versions
.iter()
.find(|version| version.version_id == Some(free_version) && version.tier_free_version())
.expect("migrated free-version identity should remain readable from target disks");
assert!(
crate::store::tiered_data_movement_source_matches(&source_free, migrated_free)
.expect("migrated free-version identity should decode")
);
let retained_w = target_versions
.versions
.iter()
.find(|version| version.version_id == Some(target_w_version))
.expect("unrelated target version should remain");
assert!(!retained_w.deleted && !retained_w.tier_free_version());
assert_eq!(retained_w.size, target_w.size);
assert_eq!(retained_w.get_etag(), target_w.etag);
assert!(
target_versions
.versions
.iter()
.all(|version| { version.tier_free_version() || !version.deleted })
);
assert!(
source_set
.load_file_info_versions_exact(&bucket, object)
.await
.expect("source versions should be readable after cleanup")
.is_none(),
"successful free-version migration should permit source cleanup"
);
let (heal_versions, _, _) = store
.heal_walk_versions_page(1, 0, &bucket, "", None, 2, 16, true)
.await
.expect("heal walk should decode the migrated free version");
let free_version_string = free_version.to_string();
let healed_free = heal_versions
.iter()
.find(|version| version.version_id.as_deref() == Some(free_version_string.as_str()))
.expect("heal walk should surface the migrated free version");
let healed_info = healed_free
.lifecycle_object_info
.as_ref()
.expect("heal walk should retain lifecycle identity for the migrated free version");
assert!(healed_info.transitioned_object.free_version);
assert_eq!(healed_info.transitioned_object.tier, source_free.transition_tier);
assert_eq!(healed_info.transitioned_object.name, source_free.transitioned_objname);
shutdown.cancel();
}
#[cfg(feature = "test-util")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial_test::serial(storage_class_env)]
async fn decommission_entry_allows_free_version_consumed_before_source_lock() {
let temp_dir = tempfile::tempdir().expect("create consumed free-version store dir");
let (ctx, store, shutdown) =
without_storage_class_env(build_isolated_test_store(temp_dir.path(), "decommission-free-consumed", &[4, 4])).await;
crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
let bucket = format!("decom-free-consumed-{}", uuid::Uuid::new_v4());
let object = "free-consumed-object";
store
.make_bucket(&bucket, &MakeBucketOptions::default())
.await
.expect("create consumed free-version bucket");
let (_, free_version) = seed_transitioned_free_version(&ctx, &store, &bucket, object).await;
mark_test_pool_decommissioning(&store, 0).await;
let source_set = store.pools[0].get_disks_by_key(object);
let barrier = crate::store::object::DecommissionFreeVersionSourceRaceBarrier::install(&bucket, object);
let decommission = tokio::spawn({
let store = store.clone();
let bucket = bucket.clone();
let source_set = source_set.clone();
async move {
store
.decommission_entry_for_test_with_bucket_incarnation(
0,
MetaCacheEntry {
name: object.to_string(),
..Default::default()
},
bucket,
source_set,
)
.await
}
});
barrier.wait_until_paused().await;
store.pools[0]
.delete_object(
&bucket,
object,
ObjectOptions {
versioned: true,
version_id: Some(free_version.to_string()),
incl_free_versions: true,
..Default::default()
},
)
.await
.expect("lifecycle should consume the source free version before decommission locks it");
assert!(
source_set
.load_file_info_versions_exact(&bucket, object)
.await
.expect("consumed source metadata should remain readable")
.is_none(),
"the lifecycle delete should remove the source free version"
);
barrier.release();
decommission
.await
.expect("decommission task should join")
.expect("a concurrently consumed free version should not fail source cleanup");
assert!(
store.pools[1]
.get_disks_by_key(object)
.load_file_info_versions_exact(&bucket, object)
.await
.expect("target metadata should remain readable")
.is_none(),
"an already consumed free version should not be recreated on the target"
);
shutdown.cancel();
}
#[cfg(feature = "test-util")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial_test::serial(storage_class_env)]
async fn decommission_entry_rejects_subquorum_free_version_conflict_and_retains_source() {
let temp_dir = tempfile::tempdir().expect("create sub-quorum free-version store dir");
let (ctx, store, shutdown) =
without_storage_class_env(build_isolated_test_store(temp_dir.path(), "decommission-free-conflict", &[4, 4])).await;
crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
let bucket = format!("decom-free-conflict-{}", uuid::Uuid::new_v4());
let object = "free-conflict-object";
store
.make_bucket(&bucket, &MakeBucketOptions::default())
.await
.expect("create sub-quorum conflict bucket");
let (_, free_version) = seed_transitioned_free_version(&ctx, &store, &bucket, object).await;
let source_free = store.pools[0]
.get_disks_by_key(object)
.load_file_info_versions_exact(&bucket, object)
.await
.expect("source free version should decode before crash replay setup")
.and_then(|versions| {
versions
.versions
.into_iter()
.find(|version| version.version_id == Some(free_version))
})
.expect("source free version should be available for crash replay setup");
let mut target_reader = PutObjReader::from_vec(b"target ordinary bytes".to_vec());
let target = store.pools[1]
.put_object(
&bucket,
object,
&mut target_reader,
&ObjectOptions {
versioned: true,
..Default::default()
},
)
.await
.expect("seed ordinary target version");
let target_version = target.version_id.expect("target version must have an ID");
let target_disks = store.pools[1].get_disks_by_key(object).disks.read().await.clone();
for disk in target_disks.iter().skip(1) {
disk.as_ref()
.expect("target crash replay quorum disk should be online")
.write_metadata("", &bucket, object, source_free.clone())
.await
.expect("seed an equivalent free version on the target quorum");
}
let conflict_path = temp_dir
.path()
.join(format!("pool1/set0/disk0/{bucket}/{object}/{STORAGE_FORMAT_FILE}"));
let encoded = tokio::fs::read(&conflict_path)
.await
.expect("target metadata should be readable");
let mut metadata = FileMeta::load(&encoded).expect("target metadata should decode");
let target_index = metadata
.versions
.iter()
.position(|version| version.header.version_id == Some(target_version))
.expect("target version should be present on the conflict disk");
let mut target_meta = metadata.versions[target_index]
.parse_version_meta()
.expect("target version metadata should decode");
target_meta
.object
.as_mut()
.expect("target conflict must remain an ordinary object")
.version_id = Some(free_version);
metadata.versions[target_index] = target_meta.try_into().expect("conflict metadata should encode");
let expected_conflict_meta = metadata.versions[target_index].meta.clone();
let expected_conflict = metadata.versions[target_index]
.into_fileinfo(&bucket, object, true)
.expect("conflict metadata should decode as an ordinary object");
let duplicate_free: rustfs_filemeta::FileMetaShallowVersion = rustfs_filemeta::FileMetaVersion::from(source_free.clone())
.try_into()
.expect("duplicate free metadata should encode");
metadata.versions.insert(target_index, duplicate_free);
tokio::fs::write(&conflict_path, metadata.marshal_msg().expect("conflict metadata should encode"))
.await
.expect("write sub-quorum conflict metadata");
mark_test_pool_decommissioning(&store, 0).await;
let source_set = store.pools[0].get_disks_by_key(object);
store
.decommission_entry_for_test_with_bucket_incarnation(
0,
MetaCacheEntry {
name: object.to_string(),
..Default::default()
},
bucket.clone(),
source_set.clone(),
)
.await
.expect("conflicted decommission entry should retain the source and retry later");
let source_versions = source_set
.load_file_info_versions_exact(&bucket, object)
.await
.expect("retained source versions should decode")
.expect("source free version should be retained after conflict");
assert!(
source_versions
.versions
.iter()
.any(|version| { version.version_id == Some(free_version) && version.tier_free_version() })
);
let post_encoded = tokio::fs::read(&conflict_path)
.await
.expect("conflict metadata should remain readable");
let post_metadata = FileMeta::load(&post_encoded).expect("post-conflict metadata should decode");
let same_id = post_metadata
.versions
.iter()
.filter(|version| version.header.version_id == Some(free_version))
.collect::<Vec<_>>();
assert_eq!(same_id.len(), 2, "conflict metadata should retain both same-ID records");
assert_eq!(same_id.iter().filter(|version| version.header.free_version()).count(), 1);
assert_eq!(same_id.iter().filter(|version| !version.header.free_version()).count(), 1);
let post_conflict = same_id
.into_iter()
.find(|version| !version.header.free_version())
.expect("ordinary conflict version must remain addressable by the source ID");
let post_conflict_info = post_conflict
.into_fileinfo(&bucket, object, true)
.expect("post-conflict ordinary metadata should decode");
assert!(!post_conflict_info.deleted && !post_conflict_info.tier_free_version());
assert_eq!(post_conflict.meta, expected_conflict_meta);
assert_eq!(post_conflict_info.size, expected_conflict.size);
assert_eq!(post_conflict_info.data_dir, expected_conflict.data_dir);
assert_eq!(post_conflict_info.metadata, expected_conflict.metadata);
assert_eq!(post_conflict_info.get_etag(), expected_conflict.get_etag());
for disk_index in 0..4 {
let target_path = temp_dir
.path()
.join(format!("pool1/set0/disk{disk_index}/{bucket}/{object}/{STORAGE_FORMAT_FILE}"));
let target_encoded = tokio::fs::read(&target_path)
.await
.expect("target metadata should remain readable");
let target_meta = FileMeta::load(&target_encoded).expect("target metadata should decode");
let same_id = target_meta
.versions
.iter()
.filter(|version| version.header.version_id == Some(free_version))
.collect::<Vec<_>>();
if disk_index == 0 {
assert_eq!(same_id.len(), 2);
assert!(same_id[0].header.free_version());
assert!(!same_id[1].header.free_version());
} else {
assert_eq!(same_id.len(), 1);
assert!(same_id[0].header.free_version());
}
}
let sweep_err = store
.check_after_decommission_for_test(0)
.await
.expect_err("final sweep must report the retained free version");
assert!(
sweep_err.to_string().contains("version(s) were found"),
"unexpected final sweep error: {sweep_err}"
);
shutdown.cancel();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial_test::serial(storage_class_env)]
async fn versioned_batch_delete_marker_skips_decommission_source() {
+1 -1
View File
@@ -151,7 +151,7 @@ pub(crate) mod init_format;
pub(crate) mod list_objects;
mod multipart;
mod object;
pub(crate) use object::{ObjectLockDiagGuard, SourceCleanupMutationFence};
pub(crate) use object::{ObjectLockDiagGuard, SourceCleanupMutationFence, tiered_data_movement_source_matches};
pub use object::{
PrepareSelectObjectSnapshotError, PreparedGetObjectReader, SelectObjectSnapshot, SelectObjectSnapshotReadError,
SnapshotConsistencyError,
+144 -16
View File
@@ -490,6 +490,82 @@ fn decommission_mutation_fence_for_test(
.map(|hook| hook.fence.clone())
}
#[cfg(test)]
struct DecommissionFreeVersionSourceRaceState {
bucket: String,
object: String,
arrived: tokio::sync::Notify,
release: tokio::sync::Notify,
}
#[cfg(test)]
pub(crate) struct DecommissionFreeVersionSourceRaceBarrier {
state: Arc<DecommissionFreeVersionSourceRaceState>,
}
#[cfg(test)]
static DECOMMISSION_FREE_VERSION_SOURCE_RACE_BARRIER: std::sync::OnceLock<
std::sync::Mutex<Option<Arc<DecommissionFreeVersionSourceRaceState>>>,
> = std::sync::OnceLock::new();
#[cfg(test)]
impl DecommissionFreeVersionSourceRaceBarrier {
pub(crate) fn install(bucket: &str, object: &str) -> Self {
let state = Arc::new(DecommissionFreeVersionSourceRaceState {
bucket: bucket.to_string(),
object: object.to_string(),
arrived: tokio::sync::Notify::new(),
release: tokio::sync::Notify::new(),
});
let mut slot = DECOMMISSION_FREE_VERSION_SOURCE_RACE_BARRIER
.get_or_init(|| std::sync::Mutex::new(None))
.lock()
.expect("decommission free-version source race barrier should not poison");
assert!(slot.is_none(), "decommission free-version source race barrier must be unique");
*slot = Some(Arc::clone(&state));
Self { state }
}
pub(crate) async fn wait_until_paused(&self) {
tokio::time::timeout(Duration::from_secs(30), self.state.arrived.notified())
.await
.expect("decommission should pause before acquiring the free-version source lock");
}
pub(crate) fn release(&self) {
self.state.release.notify_one();
}
}
#[cfg(test)]
impl Drop for DecommissionFreeVersionSourceRaceBarrier {
fn drop(&mut self) {
self.state.release.notify_one();
let mut slot = DECOMMISSION_FREE_VERSION_SOURCE_RACE_BARRIER
.get_or_init(|| std::sync::Mutex::new(None))
.lock()
.expect("decommission free-version source race barrier should not poison");
if slot.as_ref().is_some_and(|state| Arc::ptr_eq(state, &self.state)) {
*slot = None;
}
}
}
#[cfg(test)]
async fn pause_decommission_free_version_before_source_lock(bucket: &str, object: &str) {
let state = DECOMMISSION_FREE_VERSION_SOURCE_RACE_BARRIER
.get_or_init(|| std::sync::Mutex::new(None))
.lock()
.expect("decommission free-version source race barrier should not poison")
.as_ref()
.filter(|state| state.bucket == bucket && state.object == object)
.cloned();
if let Some(state) = state {
state.arrived.notify_one();
state.release.notified().await;
}
}
pub(crate) struct SourceCleanupMutationFence {
guard: ObjectLockDiagGuard,
source_lock_covered: bool,
@@ -1494,13 +1570,15 @@ fn is_equivalent_data_movement_tiered_object(source: &rustfs_filemeta::FileInfo,
&& source_actual_size == target_actual_size
}
fn tiered_data_movement_source_matches(
pub(crate) fn tiered_data_movement_source_matches(
expected: &rustfs_filemeta::FileInfo,
current: &rustfs_filemeta::FileInfo,
) -> Result<bool> {
let expected_backend = crate::services::tier::tier::tier_destination_id_from_metadata(&expected.metadata)?;
let current_backend = crate::services::tier::tier::tier_destination_id_from_metadata(&current.metadata)?;
Ok(expected.version_id == current.version_id
&& expected.deleted == current.deleted
&& expected.tier_free_version() == current.tier_free_version()
&& expected.data_dir == current.data_dir
&& expected.mod_time == current.mod_time
&& expected.size == current.size
@@ -1514,6 +1592,15 @@ fn tiered_data_movement_source_matches(
&& expected_backend == current_backend)
}
fn decommission_free_version_overwrite_error(bucket: &str, object: &str, version_id: Option<Uuid>) -> Error {
StorageError::DataMovementOverwriteErr(
bucket.to_owned(),
object.to_owned(),
version_id.map(|id| id.to_string()).unwrap_or_default(),
)
.into()
}
fn should_check_data_movement_resume_target(src_pool_idx: usize, target_pool_idx: usize) -> bool {
target_pool_idx != src_pool_idx
}
@@ -2221,6 +2308,23 @@ impl ECStore {
)
}
async fn has_equivalent_data_movement_tier_free_version(
&self,
bucket: &str,
object: &str,
source: &rustfs_filemeta::FileInfo,
opts: &ObjectOptions,
target_pool_idx: usize,
) -> Result<bool> {
let pool = self
.pools
.get(target_pool_idx)
.ok_or_else(|| Error::other(format!("invalid tiered data movement target pool {target_pool_idx}")))?;
pool.get_disks_by_key(object)
.has_decommission_tier_free_version_write_quorum(bucket, object, source, opts)
.await
}
fn resolve_decommission_target_pool_idx_result(result: Result<usize>, bucket: &str, object: &str) -> Result<usize> {
result.map_err(|err| Error::other(format!("failed to select decommission target pool for {bucket}/{object}: {err}")))
}
@@ -2240,6 +2344,10 @@ impl ECStore {
check_put_object_args(bucket, object)?;
let mut opts = opts.clone();
let is_free_version = fi.tier_free_version();
if is_free_version {
opts.incl_free_versions = true;
}
let bucket_incarnation_fence = if is_meta_bucketname(bucket) {
None
} else {
@@ -2277,6 +2385,10 @@ impl ECStore {
&object,
)?
};
#[cfg(test)]
if is_free_version {
pause_decommission_free_version_before_source_lock(bucket, logical_object).await;
}
let _object_guards = self
.acquire_data_movement_object_write_locks(bucket, &object, opts.src_pool_idx, idx, &mut opts)
.await?;
@@ -2294,7 +2406,7 @@ impl ECStore {
versions
.versions
.iter()
.find(|current| current.version_id == fi.version_id && !current.tier_free_version())
.find(|current| current.version_id == fi.version_id && current.tier_free_version() == is_free_version)
})
.ok_or_else(|| to_object_err(StorageError::FileNotFound, vec![bucket, object.as_str()]))?;
if !tiered_data_movement_source_matches(fi, current_source)? {
@@ -2309,24 +2421,40 @@ impl ECStore {
.get_available_pool_idx_excluding(bucket, &object, fi.size, opts.src_pool_idx)
.await;
let target_pool_idx = resolve_data_movement_resume_target_pool(idx, resume_target_pool_idx, opts.src_pool_idx);
if is_free_version && target_pool_idx == opts.src_pool_idx {
return Err(Error::DiskFull);
}
let equivalent = if is_free_version {
self.has_equivalent_data_movement_tier_free_version(bucket, &object, &fi, &opts, target_pool_idx)
.await?
} else {
self.has_equivalent_data_movement_tiered_object(bucket, &object, &fi, &opts, target_pool_idx)
.await?
};
if equivalent {
return Ok(());
}
return Err(decommission_free_version_overwrite_error(bucket, &object, fi.version_id));
}
let result = if is_free_version {
if self
.has_equivalent_data_movement_tiered_object(bucket, &object, &fi, &opts, target_pool_idx)
.has_equivalent_data_movement_tier_free_version(bucket, &object, &fi, &opts, idx)
.await?
{
return Ok(());
}
return Err(StorageError::DataMovementOverwriteErr(
bucket.to_owned(),
object.to_owned(),
opts.version_id.clone().unwrap_or_default(),
));
}
let result = self.pools[idx]
.get_disks_by_key(&object)
.decommission_tiered_object(bucket, &object, &fi, &opts)
.await;
self.pools[idx]
.get_disks_by_key(&object)
.decommission_tier_free_version(bucket, &object, &fi, &opts)
.await
} else {
self.pools[idx]
.get_disks_by_key(&object)
.decommission_tiered_object(bucket, &object, &fi, &opts)
.await
};
if matches!(result, Err(Error::PreconditionFailed)) {
if self
.has_equivalent_data_movement_tiered_object(bucket, &object, &fi, &opts, idx)
@@ -3194,7 +3322,7 @@ impl ECStore {
// Default return value
let mut del_objects = vec![DeletedObject::default(); objects.len()];
let mut accounting = vec![None; objects.len()];
let 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(|left, right| right.idx.cmp(&left.idx));
latest_candidates.sort_by_key(|candidate| std::cmp::Reverse(candidate.idx));
let Some(winner) = latest_candidates.first() else {
return Err(Error::ErasureReadQuorum);
+19
View File
@@ -90,6 +90,21 @@ fn legacy_data_key_for_version(version_id: Option<Uuid>) -> Option<String> {
pub const TRANSITION_COMPLETE: &str = "complete";
pub const TRANSITION_PENDING: &str = "pending";
/// xl.meta key marking a tier free-version record.
///
/// A free version is a delete-marker-shaped cleanup hint appended by
/// [`MetaObject::delete_version`] when a version whose remote transition
/// completed is removed from xl.meta; it carries the remote tier identity for
/// an idempotent remote delete and is never a user-visible version
/// (`num_versions` excludes it). While the record exists it is consumed by the
/// lifecycle free-version recovery scan and the usage scanner, which re-enqueue
/// the pending remote delete, and by heal metadata walks. On S3 and lifecycle
/// delete paths the same obligation is also carried by a committed tier-journal
/// entry; deletes without such an entry (for example a removed version whose
/// transition state decodes as unknown) rely on this record alone until the
/// worker removes it after a successful remote delete. Decommission preserves
/// the record and its remote identity on the target pool before source cleanup
/// — see docs/architecture/decommission-compatibility.md.
pub const FREE_VERSION: &str = "free-version";
pub const TRANSITION_STATUS: &str = "transition-status";
@@ -447,6 +462,10 @@ impl FileMeta {
};
if let Some(fidx) = existing_idx {
let existing = self.versions[fidx].parse_version_meta()?;
if existing.free_version() != version.free_version() {
return Err(Error::other("cannot replace a free version with a non-free version"));
}
return self.set_idx(fidx, version);
}
+9
View File
@@ -2725,6 +2725,15 @@ impl MetaObject {
self.meta_sys.retain(|k, _| !k.starts_with("X-Amz-Restore"));
}
/// Builds the free-version cleanup record appended when a transitioned
/// version is removed from xl.meta. The record keeps the remote tier
/// identity so the lifecycle worker can issue the idempotent remote delete
/// and only then remove the record; until then the recovery scan and the
/// usage scanner keep re-enqueueing it. S3 and lifecycle deletes also
/// persist a committed tier-journal entry for the same remote delete. The
/// decommission path copies this record unchanged before source cleanup,
/// including when the transition state is unknown — see
/// docs/architecture/decommission-compatibility.md.
pub fn init_free_version(&self, fi: &FileInfo) -> Result<(FileMetaVersion, bool)> {
if fi.skip_tier_free_version() {
return Ok((FileMetaVersion::default(), false));
+86 -25
View File
@@ -16,14 +16,14 @@
//!
//! `scripts/test/vault_ha_kms_live.sh` owns the official Vault containers and
//! kills the active node while this test continuously decrypts through a
//! surviving standby. KV2 and Transit requests must remain successful, use a
//! bounded number of attempts, and leave the circuit and in-flight gauges at
//! zero after a new leader is elected.
//! surviving standby. KV2 and Transit must recover after the bounded circuit
//! interval, use a bounded number of attempts, and leave the circuit and
//! in-flight gauges at zero after a new leader is elected.
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use metrics_util::MetricKind;
@@ -43,6 +43,11 @@ const OPERATION_ATTEMPTS: &str = "rustfs_kms_backend_operation_attempts";
const IN_FLIGHT: &str = "rustfs_kms_backend_in_flight";
const CIRCUIT_OPEN: &str = "rustfs_kms_backend_circuit_open";
const MAX_ATTEMPTS: u32 = 10;
const ATTEMPT_TIMEOUT: Duration = Duration::from_secs(2);
const HEALTHY_PROGRESS_TIMEOUT: Duration = Duration::from_secs(20);
// The circuit remains open for 30s after five failed attempts.
const POST_FAILOVER_PROGRESS_TIMEOUT: Duration = Duration::from_secs(35);
const FAILOVER_ERROR_POLL_INTERVAL: Duration = Duration::from_millis(100);
type MetricEntry = (
metrics_util::CompositeKey,
@@ -64,7 +69,7 @@ fn config(backend: KmsBackend, backend_config: BackendConfig) -> KmsConfig {
backend,
backend_config,
allow_insecure_dev_defaults: true,
timeout: Duration::from_secs(2),
timeout: ATTEMPT_TIMEOUT,
retry_attempts: MAX_ATTEMPTS,
enable_cache: false,
..KmsConfig::default()
@@ -164,14 +169,31 @@ fn retryable_failures(snapshot: &[MetricEntry], operation: &str) -> u64 {
.sum()
}
async fn wait_for_count(counter: &AtomicU64, minimum: u64, description: &str) {
tokio::time::timeout(Duration::from_secs(20), async {
async fn wait_for_count(
counter: &AtomicU64,
failure: &Mutex<Option<String>>,
minimum: u64,
description: &str,
timeout: Duration,
) {
tokio::time::timeout(timeout, async {
while counter.load(Ordering::SeqCst) < minimum {
if let Some(error) = failure.lock().expect("decrypt failure lock poisoned").as_ref() {
panic!(
"{description} worker failed after {} successful decrypts: {error}",
counter.load(Ordering::SeqCst)
);
}
tokio::time::sleep(Duration::from_millis(25)).await;
}
})
.await
.unwrap_or_else(|_| panic!("timed out waiting for {description}"));
.unwrap_or_else(|_| {
panic!(
"timed out after {timeout:?} waiting for {description}: completed {}, expected {minimum}",
counter.load(Ordering::SeqCst)
)
});
}
async fn wait_for_file(path: &Path, description: &str) {
@@ -189,7 +211,8 @@ async fn decrypt_loop<B: KmsBackendTrait + Send + Sync + 'static>(
request: DecryptRequest,
expected: Vec<u8>,
completed: Arc<AtomicU64>,
failed: Arc<AtomicBool>,
allow_failover_errors: Arc<AtomicBool>,
failure: Arc<Mutex<Option<String>>>,
stop: CancellationToken,
) {
while !stop.is_cancelled() {
@@ -197,8 +220,18 @@ async fn decrypt_loop<B: KmsBackendTrait + Send + Sync + 'static>(
Ok(response) if response.plaintext == expected => {
completed.fetch_add(1, Ordering::SeqCst);
}
Ok(_) | Err(_) => {
failed.store(true, Ordering::SeqCst);
Ok(_) => {
*failure.lock().expect("decrypt failure lock poisoned") =
Some("decrypt returned unexpected plaintext".to_string());
return;
}
Err(rustfs_kms::KmsError::BackendError { .. } | rustfs_kms::KmsError::OperationTimedOut { .. })
if allow_failover_errors.load(Ordering::SeqCst) =>
{
tokio::time::sleep(FAILOVER_ERROR_POLL_INTERVAL).await;
}
Err(error) => {
*failure.lock().expect("decrypt failure lock poisoned") = Some(error.to_string());
return;
}
}
@@ -296,7 +329,9 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
);
let stop = CancellationToken::new();
let failed = Arc::new(AtomicBool::new(false));
let allow_failover_errors = Arc::new(AtomicBool::new(false));
let kv2_failure = Arc::new(Mutex::new(None));
let transit_failure = Arc::new(Mutex::new(None));
let kv2_completed = Arc::new(AtomicU64::new(0));
let transit_completed = Arc::new(AtomicU64::new(0));
let kv2_worker = tokio::spawn(decrypt_loop(
@@ -304,7 +339,8 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
kv2_request,
kv2_data_key.plaintext_key,
Arc::clone(&kv2_completed),
Arc::clone(&failed),
Arc::clone(&allow_failover_errors),
Arc::clone(&kv2_failure),
stop.clone(),
));
let transit_worker = tokio::spawn(decrypt_loop(
@@ -312,12 +348,21 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
transit_request,
transit_data_key.plaintext_key,
Arc::clone(&transit_completed),
Arc::clone(&failed),
Arc::clone(&allow_failover_errors),
Arc::clone(&transit_failure),
stop.clone(),
));
wait_for_count(&kv2_completed, 2, "two healthy KV2 decrypts").await;
wait_for_count(&transit_completed, 2, "two healthy Transit decrypts").await;
wait_for_count(&kv2_completed, &kv2_failure, 2, "two healthy KV2 decrypts", HEALTHY_PROGRESS_TIMEOUT).await;
wait_for_count(
&transit_completed,
&transit_failure,
2,
"two healthy Transit decrypts",
HEALTHY_PROGRESS_TIMEOUT,
)
.await;
allow_failover_errors.store(true, Ordering::SeqCst);
std::fs::write(&marker, b"ready").expect("publish failover readiness marker");
wait_for_file(&elected, "the replacement Vault leader").await;
@@ -326,18 +371,39 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
let kv2_after_election = kv2_completed.load(Ordering::SeqCst) + 2;
let transit_after_election = transit_completed.load(Ordering::SeqCst) + 2;
wait_for_count(&kv2_completed, kv2_after_election, "post-failover KV2 decrypts").await;
wait_for_count(&transit_completed, transit_after_election, "post-failover Transit decrypts").await;
wait_for_count(
&kv2_completed,
&kv2_failure,
kv2_after_election,
"post-failover KV2 decrypts",
POST_FAILOVER_PROGRESS_TIMEOUT,
)
.await;
wait_for_count(
&transit_completed,
&transit_failure,
transit_after_election,
"post-failover Transit decrypts",
POST_FAILOVER_PROGRESS_TIMEOUT,
)
.await;
stop.cancel();
kv2_worker.await.expect("KV2 decrypt worker must join");
transit_worker.await.expect("Transit decrypt worker must join");
assert!(!failed.load(Ordering::SeqCst), "no decrypt may fail or return different plaintext");
assert!(
kv2_failure.lock().expect("KV2 failure lock poisoned").is_none(),
"no KV2 decrypt may fail or return different plaintext"
);
assert!(
transit_failure.lock().expect("Transit failure lock poisoned").is_none(),
"no Transit decrypt may fail or return different plaintext"
);
}
#[test]
#[ignore = "requires a real three-node Vault Raft cluster; run scripts/test/vault_ha_kms_live.sh"]
fn vault_raft_leader_failure_preserves_kv2_and_transit_decrypts() {
fn vault_raft_leader_failure_recovers_kv2_and_transit_decrypts() {
let recorder = DebuggingRecorder::new();
let snapshotter = recorder.snapshotter();
metrics::with_local_recorder(&recorder, || {
@@ -349,11 +415,6 @@ fn vault_raft_leader_failure_preserves_kv2_and_transit_decrypts() {
});
let snapshot = snapshotter.snapshot().into_vec();
assert_eq!(
counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "circuit_open")]),
0,
"a bounded leader election must not open the circuit"
);
assert_eq!(
counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "budget_exhausted")]),
0,
-3
View File
@@ -43,9 +43,6 @@ 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]
@@ -153,6 +153,99 @@ No migration step is required for these decisions because this note documents th
current RustFS behavior. Changing either decision later requires an operator
compatibility note and updated characterization tests.
## Tier Free Versions During Decommission
A tier free version is an internal xl.meta record (`rustfs_filemeta::FREE_VERSION`,
flagged `XL_FLAG_FREE_VERSION`) shaped like a delete marker. It is created by
`MetaObject::init_free_version` when a version whose remote transition completed is
deleted locally: the visible version is removed and the record keeps the remote-tier
identity (tier, object name, version id, state, destination id) needed for an
idempotent remote delete. Free versions are not user-visible versions; `num_versions`
and all listing/GET paths exclude them.
### Lifecycle And Consumers
Creation: any local delete that removes a version whose transition status is
`complete` appends the record via `MetaObject::delete_version`
`init_free_version` (skipped only when `skip_tier_free_version` is set, as on
data-movement copies). The same deletes also persist a durable tier-journal
entry on every user-facing path: S3 single deletes (`execute_delete_object`
`delete_object_with_tier_delete_journal`), S3 batch deletes, lifecycle expiry,
and lifecycle delete-all all prepare and commit a journal entry around the
delete. A journal entry is omitted when the removed version's transition state
decodes as `TransitionVersionState::Unknown`, or on internal journal-less
delete paths that never touch transitioned user objects.
Consumption while the record exists: the background recovery loop started by
`init_background_expiry` (spawned by `spawn_tier_free_version_recovery_once`,
enabled by default) scans disks for pending records and re-enqueues them; the
usage scanner does the same; the lifecycle worker then deletes the remote tier
object idempotently and only afterwards removes the local record. Heal walks
include free-version records in metadata healing. Transition planning,
replication, restore, GET, listings, and usage aggregation never depend on
them.
### Decommission Handling
The exact decommission inventory loader (`load_file_info_versions_exact` via
`get_all_file_info_versions`) keeps free-version records inline in `versions`.
The migration loop handles them before lifecycle expiry and delete-marker
shortcuts. It selects a target pool using the free-version-aware lookup, then
writes the original free record to every target disk with the normal metadata
write quorum. The free-version marker, local version id, transition identity,
transition state, and destination id are preserved at the FileInfo/metadata
boundary.
The source record is physically removed only after the target write quorum has
committed and the source cleanup preflight still matches the exact inventory.
If the lifecycle worker has already completed the remote delete and removed the
source record before decommission acquires the source lock, decommission records
that identity as already consumed and treats the missing source record as safe.
If target capacity, metadata validation, lock fencing, or quorum fails, the
source record remains and the entry records `state = "free_version_retained"`
with reason `tier_free_version_migration_failed`; the worker retries the
operation on a later pass. A target record with the same version id is accepted
only when its free-version identity matches; a conflicting ordinary version or
different free record is an overwrite error. This makes retries idempotent and
prevents a free record from replacing a user-visible version.
### Reference-Audit Result
After migration, user-facing GET/list/transition/replication/restore paths still
exclude the record. Recovery, usage scanning, lifecycle tier cleanup, and heal
continue to see it when they request free versions, so an unresolved remote
delete remains actionable on the target pool. The committed tier journal remains
an independent retry source where one exists; it is not used as a reason to drop
the xl.meta record. In particular, `Unknown` transition state records are
migrated unchanged rather than discarded: the lifecycle worker retains them if
remote identity validation cannot make a delete request.
Each migrated record emits `state = "free_version_migrated"` with reason
`tier_free_version_migrated`. A record consumed before migration emits
`state = "free_version_consumed"` with reason
`tier_free_version_already_consumed`. Each failed record emits the retained state
and failure reason above. The entry also emits a disposition summary with
migrated, consumed, retained, and total counts. The final decommission sweep uses
the exact loader, counts free records still present, and emits one retained
record/reason for each unresolved free version before failing the sweep. This
makes successful migration, completed cleanup, and retained cleanup obligations
visible instead of silently omitting free records.
No new S3-visible version or admin response field is needed: free versions remain
internal and are never counted as user-visible versions. The structured
`decommission_entry` events are the operational status surface for the
free-version disposition; the existing decommission item/failed counters still
report the enclosing object migration result.
Regression guard:
- `decommission_tier_free_version_preserves_remote_identity`
- `decommission_tier_free_version_resume_requires_write_quorum`
- `decommission_tier_free_version_commit_rejects_lost_fence`
- `test_decommission_cleanup_preflight_accepts_migrated_free_version_consumed_from_source`
- `decommission_entry_skips_cleanup_only_marker_when_free_version_is_present`
- `decommission_entry_rejects_subquorum_free_version_conflict_and_retains_source`
## Regression Guard
The queued multi-pool contract is guarded by:
+149
View File
@@ -0,0 +1,149 @@
# 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.
+2 -2
View File
@@ -84,7 +84,7 @@
| protocols | 16 | 🌙 |
| quota_test | 14 | |
| reliability_disk_fault_test | 4 | |
| reliant | 32 | 19 ✅ |
| reliant | 25 | 19 ✅ |
| replication_extension_test | 75 | 20 ✅ +55 🌙 |
| security_boundary_test | 4 | |
| server_startup_failfast_test | 1 | |
@@ -99,4 +99,4 @@
| tls_hot_reload_test | 1 | ✅ |
| version_id_regression_test | 10 | ✅ |
**Total listed: 582 tests across 82 modules · PR smoke: 163 tests / 36 modules · merge/main full: 460 tests / 73 modules · nightly replication: 55 tests · nightly cluster faults: 28 tests / 7 modules · nightly protocols: 16 tests** · updated 2026-08-23.
**Total listed: 575 tests across 82 modules · PR smoke: 163 tests / 36 modules · merge/main full: 453 tests / 73 modules · nightly replication: 55 tests · nightly cluster faults: 28 tests / 7 modules · nightly protocols: 16 tests** · updated 2026-08-23.
+2 -2
View File
@@ -336,13 +336,13 @@ opentelemetry = { workspace = true }
tracing-opentelemetry = { workspace = true }
# Data structures
hashbrown = { workspace = true, features = ["serde", "rayon"] }
mimalloc = { workspace = true }
rustfs-mimalloc = { workspace = true }
[target.'cfg(target_os = "linux")'.dependencies]
libsystemd.workspace = true
[target.'cfg(not(target_os = "windows"))'.dependencies]
libmimalloc-sys.workspace = true
rustfs-mimalloc-sys.workspace = true
[dev-dependencies]
uuid = { workspace = true, features = ["v4", "v5", "fast-rng", "macro-diagnostics"] }
+1 -7
View File
@@ -369,14 +369,8 @@ 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> {
// 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);
}
rustfs_mimalloc::MiMalloc::collect(force);
Ok(())
}
+10 -8
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 { mimalloc::MiMalloc.alloc(layout) }
unsafe { rustfs_mimalloc::MiMalloc.alloc(layout) }
}
unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 {
// SAFETY: the caller upholds GlobalAlloc's contract for layout.
unsafe { mimalloc::MiMalloc.alloc_zeroed(layout) }
unsafe { rustfs_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 { mimalloc::MiMalloc.dealloc(ptr, layout) }
unsafe { rustfs_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 { mimalloc::MiMalloc.realloc(ptr, layout, new_size) }
unsafe { rustfs_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: mimalloc::MiMalloc = mimalloc::MiMalloc;
static GLOBAL: rustfs_mimalloc::MiMalloc = rustfs_mimalloc::MiMalloc;
fn main() {
let _hotpath_guard = hotpath::HotpathGuardBuilder::new("main").build();
@@ -71,8 +71,9 @@ 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 { libmimalloc_sys::mi_is_in_heap_region(allocation.as_ptr().cast()) });
assert!(unsafe { heap.contains(allocation.as_ptr()) });
}
#[test]
@@ -85,12 +86,13 @@ 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 { libmimalloc_sys::mi_is_in_heap_region(ptr.cast()) });
assert!(unsafe { heap.contains(ptr) });
assert!(unsafe { std::slice::from_raw_parts(ptr, 32).iter().all(|byte| *byte == 0) });
// SAFETY: `ptr` was allocated by `allocator` with `layout`; on failure
@@ -102,7 +104,7 @@ mod tests {
panic!("mimalloc realloc failed in allocator smoke test");
}
assert!(unsafe { libmimalloc_sys::mi_is_in_heap_region(grown_ptr.cast()) });
assert!(unsafe { heap.contains(grown_ptr) });
// SAFETY: `grown_ptr` was reallocated by `allocator` and is released
// with the matching grown layout.
unsafe { allocator.dealloc(grown_ptr, grown_layout) };
+19 -36
View File
@@ -17,10 +17,7 @@ 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;
@@ -231,7 +228,18 @@ fn read_cgroup_memory_snapshot() -> Option<CgroupMemorySnapshot> {
read_cgroup_v2().or_else(read_cgroup_v1)
}
#[cfg(any(test, not(target_os = "windows")))]
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,
})
}
fn numeric_json_value(value: &Value) -> Option<u64> {
match value {
Value::Number(number) => number
@@ -242,7 +250,6 @@ 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
@@ -254,7 +261,6 @@ 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) => {
@@ -271,12 +277,10 @@ 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()
@@ -285,7 +289,6 @@ 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"];
@@ -312,33 +315,6 @@ 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)
}
@@ -566,6 +542,13 @@ 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);
+1 -1
View File
@@ -241,7 +241,7 @@ env \
RUSTFS_TEST_VAULT_FAILOVER_MARKER="$MARKER" \
RUSTFS_TEST_VAULT_OLD_LEADER="$OLD_LEADER" \
cargo test -p rustfs-kms --test vault_ha_failover_live \
vault_raft_leader_failure_preserves_kv2_and_transit_decrypts -- \
vault_raft_leader_failure_recovers_kv2_and_transit_decrypts -- \
--ignored --nocapture --test-threads=1 &
TEST_PID=$!