Compare commits

..

6 Commits

Author SHA1 Message Date
houseme 16fc3ff0e7 Merge branch 'main' into cxymds/fix-1935-size-resolution-unknown-run2
Signed-off-by: houseme <housemecn@gmail.com>
2026-08-23 12:32:56 +08:00
马登山 6dbff1c036 fix(scanner): close lifecycle review gaps 2026-08-23 09:45:36 +08:00
马登山 854b6dfb0c fix(ci): resolve ecstore clippy warnings 2026-08-23 09:20:07 +08:00
马登山 893a4a11c1 Merge remote-tracking branch 'origin/main' into HEAD 2026-08-23 09:17:13 +08:00
马登山 2086ade967 fix(scanner): preserve restore expiry semantics 2026-08-23 08:41:31 +08:00
马登山 23b85b792e fix(scanner): unify unknown metadata size accounting 2026-08-22 22:58:05 +08:00
22 changed files with 1387 additions and 129 deletions
+2 -2
View File
@@ -325,8 +325,8 @@ slow-timeout = { period = "60s", terminate-after = 2, grace-period = "10s" }
#
# Wired by .github/workflows/e2e-replication-nightly.yml (schedule +
# workflow_dispatch), which builds the rustfs binary once, installs awscurl so
# the STS dual-node test actually exercises its path (the test fails when
# awscurl is absent), and routes scheduled failures
# the STS dual-node test actually exercises its path (it skips gracefully with
# a visible log line when awscurl is absent), and routes scheduled failures
# through .github/actions/schedule-failure-issue (ci-8). Explicit division of
# labor with e2e-full: these tests run only in the consolidated nightly
# workflow, not in the merge/main lane.
-27
View File
@@ -681,19 +681,6 @@ jobs:
cache-save-if: 'false'
install-build-packaging-tools: 'false'
- name: Set up Python
uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6.3.0
with:
python-version: "3.12"
- name: Install awscurl
run: |
python3 -m pip install --user --upgrade pip "awscurl==0.44"
echo "AWSCURL_PATH=$HOME/.local/bin/awscurl" >> "$GITHUB_ENV"
- name: Verify awscurl
run: test -x "$AWSCURL_PATH"
# Download after the cache restore so the freshly built binary from the
# build job always wins over anything restored into target/debug.
- name: Download debug binary
@@ -816,20 +803,6 @@ jobs:
- name: Verify awscurl
run: test -x "$AWSCURL_PATH"
- name: Install mc
env:
MC_VERSION: RELEASE.2025-08-13T08-35-41Z
MC_SHA256: 01f866e9c5f9b87c2b09116fa5d7c06695b106242d829a8bb32990c00312e891
run: |
MC_BINARY="mc.linux-amd64.${MC_VERSION}"
curl -fsSLo "$RUNNER_TEMP/mc" "https://github.com/minio/mc/releases/download/${MC_VERSION}/${MC_BINARY}"
echo "${MC_SHA256} $RUNNER_TEMP/mc" | sha256sum --check --status
chmod +x "$RUNNER_TEMP/mc"
echo "$RUNNER_TEMP" >> "$GITHUB_PATH"
- name: Verify mc
run: mc --version
- name: Install Vault
run: |
VAULT_VERSION="1.17.6"
@@ -75,7 +75,11 @@ jobs:
cache-save-if: ${{ github.ref == 'refs/heads/main' }}
install-build-packaging-tools: 'false'
# The STS dual-node test requires awscurl and fails if it is unavailable.
# awscurl lets the STS dual-node test actually exercise its path. Without
# it the test skips gracefully with a visible log line
# (`awscurl_available()` in crates/e2e_test/src/common.rs), so the lane
# still passes — installing it just upgrades that one test from skip to
# real coverage.
- name: Set up Python
uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6.3.0
with:
@@ -83,7 +87,7 @@ jobs:
- name: Install awscurl
run: |
python3 -m pip install --user --upgrade pip "awscurl==0.44"
python3 -m pip install --user --upgrade pip awscurl
echo "AWSCURL_PATH=$HOME/.local/bin/awscurl" >> "$GITHUB_ENV"
- name: Verify awscurl
+89 -1
View File
@@ -307,6 +307,38 @@ pub struct DiskUsageStatus {
pub snapshot_exists: bool,
}
/// A bounded reconciliation record for an object whose logical size could not
/// be trusted at the scanner boundary. The scanner persists these records in
/// its cache; keeping the model here avoids a second, incompatible accounting
/// representation in storage-facing crates.
#[derive(Debug, Default, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct SizeReconciliationEntry {
/// Stable object/version identity key (not a metrics label).
pub key: String,
pub bucket: String,
pub object: String,
#[serde(default)]
pub version_id: Option<String>,
#[serde(default)]
pub generation: Option<String>,
/// Structured reason label; raw metadata values must never be stored here.
pub reason: String,
#[serde(default)]
pub physical_size: Option<u64>,
#[serde(default)]
pub first_seen: u64,
#[serde(default)]
pub attempts: u32,
}
/// Object scope refreshed by one scanner pass. Existing debts in this scope
/// are removed before the pass's unresolved records are inserted.
#[derive(Debug, Default, Clone, PartialEq, Eq)]
pub struct SizeReconciliationScope {
pub bucket: String,
pub object: String,
}
/// Size summary for a single object or group of objects
#[derive(Debug, Default, Clone)]
pub struct SizeSummary {
@@ -336,6 +368,16 @@ pub struct SizeSummary {
pub repl_target_stats: HashMap<String, ReplTargetSizeSummary>,
/// Per-tier accounting, keyed by storage class or remote tier name
pub tier_stats: HashMap<String, TierStats>,
/// Size-resolution debts observed while scanning this summary.
pub size_reconciliation: Vec<SizeReconciliationEntry>,
/// True when the per-object summary exceeded its bounded debt buffer.
/// Callers must retain prior ledger entries rather than treating the
/// partial list as a complete refresh.
pub size_reconciliation_truncated: bool,
/// Object scopes refreshed by this summary. They let the durable ledger
/// remove versions that resolved without allocating one key per healthy
/// version on the hot path.
pub reconciliation_scopes: Vec<SizeReconciliationScope>,
}
/// Replication target size summary
@@ -833,7 +875,8 @@ impl DataUsageEntry {
///
/// The canonical wire format is written by the hand-written map-encoded
/// `Serialize` on the scanner-side `DataUsageCacheInfo`
/// (`crates/scanner/src/data_usage_define.rs`), which carries 16 fields.
/// (`crates/scanner/src/data_usage_define.rs`), which carries the original 16
/// fields plus an optional reconciliation field.
/// This type decodes only the shared subset and is deliberately not
/// `Serialize`: a derived (array) encoding of this 6-field subset would
/// corrupt the cache for scanner readers, so no write path may exist here.
@@ -1777,6 +1820,51 @@ impl SizeSummary {
entry.pending_count = entry.pending_count.saturating_add(stats.pending_count);
entry.failed_count = entry.failed_count.saturating_add(stats.failed_count);
}
for entry in &other.size_reconciliation {
self.record_size_reconciliation(entry.clone());
}
self.size_reconciliation_truncated |= other.size_reconciliation_truncated;
for scope in &other.reconciliation_scopes {
self.record_reconciliation_scope(&scope.bucket, &scope.object);
}
}
/// Add one reconciliation debt, coalescing repeated observations in the
/// same object summary. The scanner cache applies its own larger bound.
pub fn record_size_reconciliation(&mut self, entry: SizeReconciliationEntry) {
const MAX_SUMMARY_RECONCILIATION_ENTRIES: usize = 1024;
if let Some(existing) = self.size_reconciliation.iter_mut().find(|value| value.key == entry.key) {
existing.reason = entry.reason;
existing.physical_size = entry.physical_size;
existing.generation = entry.generation;
existing.version_id = entry.version_id;
return;
}
if self.size_reconciliation.len() < MAX_SUMMARY_RECONCILIATION_ENTRIES {
self.size_reconciliation.push(entry);
} else {
self.size_reconciliation_truncated = true;
}
}
/// Mark one object scope as refreshed. Duplicate scopes are suppressed so
/// merging summaries remains bounded and deterministic.
pub fn record_reconciliation_scope(&mut self, bucket: &str, object: &str) {
if !self
.reconciliation_scopes
.iter()
.any(|scope| scope.bucket == bucket && scope.object == object)
{
if self.reconciliation_scopes.len() >= 1024 {
self.size_reconciliation_truncated = true;
return;
}
self.reconciliation_scopes.push(SizeReconciliationScope {
bucket: bucket.to_string(),
object: object.to_string(),
});
}
}
}
+9 -7
View File
@@ -123,7 +123,7 @@ via `create_s3_client(idx)` / `create_all_clients()`. See
| `find_available_port` | Random free port (isolation primitive) |
| `rustfs_binary_path` / `_with_features` | Locate/build the binary; honors `RUSTFS_BUILD_FEATURES` |
| `requested_rustfs_build_features` / `rustfs_build_feature_enabled` | Feature-gate a test to what the binary was built with |
| `execute_awscurl` / `awscurl_post` / `_get` / `_put` / `_delete` / `awscurl_post_sts_form_urlencoded` | Admin/STS API calls via `awscurl`; missing binaries are test failures |
| `awscurl_available` + `execute_awscurl` / `awscurl_post` / `_get` / `_put` / `_delete` / `awscurl_post_sts_form_urlencoded` | Admin/STS API calls via `awscurl` (skip gracefully when absent) |
| `replication_fast_env` | Env vars that shrink replication timers (from repl-4); pass to `start_rustfs_server_with_env` |
| `local_http_client` / `init_logging` | Loopback HTTP client; idempotent tracing init |
| `RustFSTestClusterEnvironment` (`new`/`start`/`start_node`/`stop_node`/`create_all_clients`) | Multi-node harness |
@@ -189,7 +189,7 @@ cargo nextest run --profile e2e-smoke -p e2e_test
cargo nextest run --profile e2e-full -p e2e_test
# Cluster fault nightly lane
cargo nextest run --profile e2e-nightly -p e2e_test
# Replication nightly lane; awscurl is required for STS paths
# Replication nightly lane; install awscurl so STS paths do not skip
cargo nextest run --profile e2e-repl-nightly -p e2e_test
# Fixed-port protocol nightly lane
RUSTFS_BUILD_FEATURES=ftps,webdav,sftp \
@@ -221,8 +221,9 @@ The `s3s-e2e` CI job selects a random `RUSTFS_TEST_PORT` (see the `e2e-tests`
job) to dodge this; local single-node tests already use random ports, so a
lingering orphan is usually the cause of a spurious bind failure.
**`awscurl` not found.** `awscurl`-dependent tests fail closed with a process
spawn error. Install the pinned CI version before running their profiles.
**`awscurl` not found.** `awscurl`-dependent tests skip gracefully with a
visible log line (`awscurl_available()`); install `awscurl` to actually run
them.
## Related
@@ -257,9 +258,10 @@ A test module may join the smoke filter only if every test in it is:
2. **Single-node** — spawns its own server via
`RustFSTestEnvironment`/`start_rustfs_server` on a random port with an
isolated temp dir. No `RustFSTestClusterEnvironment`, no fixed ports.
3. **Hermetic dependencies** — no pre-started server at `localhost:9000`, no
Vault, and no fixed protocol ports. Any required CLI must be pinned and
installed by the workflow; a missing CLI must fail the test.
3. **Dependency-free** — no pre-started server at `localhost:9000`, no Vault,
no fixed protocol ports. Tools that may be absent on the runner (e.g.
`awscurl`) are acceptable only when the test skips gracefully with a
visible log line (see `bucket_policy_check_test.rs`).
4. **Not `#[ignore]`** — ignored tests are activation work (backlog#1149
ci-13 / backlog#1148 ilm-3), not smoke candidates.
@@ -52,6 +52,10 @@ fn create_user_client(env: &RustFSTestEnvironment, access_key: &str, secret_key:
#[tokio::test]
async fn test_bucket_policy_authenticated_user() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
init_logging();
if !crate::common::awscurl_available() {
info!("Skipping test_bucket_policy_authenticated_user because awscurl is not available");
return Ok(());
}
info!("Starting test_bucket_policy_authenticated_user...");
let mut env = RustFSTestEnvironment::new().await?;
+11
View File
@@ -494,6 +494,17 @@ fn awscurl_binary_path() -> PathBuf {
.unwrap_or_else(|| PathBuf::from("awscurl"))
}
pub fn awscurl_available() -> bool {
let path = awscurl_binary_path();
if path.components().count() > 1 || path.is_absolute() {
return path.is_file();
}
std::env::var_os("PATH")
.map(|paths| std::env::split_paths(&paths).any(|dir| dir.join(&path).is_file()))
.unwrap_or(false)
}
// Global initialization
static INIT: Once = Once::new();
@@ -16,7 +16,9 @@
//! session policy** (`Policy` parameter) via `awscurl --service sts` with explicit
//! `Content-Type: application/x-www-form-urlencoded` on `POST /`.
use crate::common::{RustFSTestEnvironment, awscurl_delete, awscurl_post_sts_form_urlencoded, awscurl_put, init_logging};
use crate::common::{
RustFSTestEnvironment, awscurl_available, awscurl_delete, awscurl_post_sts_form_urlencoded, awscurl_put, init_logging,
};
use aws_sdk_s3::config::{Credentials, Region};
use aws_sdk_s3::primitives::ByteStream;
use aws_sdk_s3::types::{Delete, ObjectIdentifier, Tag, Tagging};
@@ -173,6 +175,11 @@ async fn cleanup_bucket_and_object(admin: &Client, bucket: &str, key: &str) {
#[tokio::test]
async fn test_e2e_iam_policy_existing_object_tag_get_object() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
init_logging();
if !awscurl_available() {
info!("Skipping test_e2e_iam_policy_existing_object_tag_get_object: awscurl not available");
return Ok(());
}
let suffix = Uuid::new_v4();
let user = format!("e2eiamtag-{suffix}");
let user_secret = "longSecretKeyForTest123!";
@@ -226,6 +233,11 @@ async fn test_e2e_iam_policy_existing_object_tag_get_object() -> Result<(), Box<
#[tokio::test]
async fn test_e2e_bucket_policy_existing_object_tag_get_object() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
init_logging();
if !awscurl_available() {
info!("Skipping test_e2e_bucket_policy_existing_object_tag_get_object: awscurl not available");
return Ok(());
}
let suffix = Uuid::new_v4();
let user = format!("e2ebptag-{suffix}");
let user_secret = "longSecretKeyForTest456!";
@@ -282,6 +294,11 @@ async fn test_e2e_bucket_policy_existing_object_tag_get_object() -> Result<(), B
#[tokio::test]
async fn test_e2e_sts_assume_role_session_policy_existing_object_tag() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
init_logging();
if !awscurl_available() {
info!("Skipping test_e2e_sts_assume_role_session_policy_existing_object_tag: awscurl not available");
return Ok(());
}
let suffix = Uuid::new_v4();
let parent = format!("e2e-sts-par-{suffix}");
let parent_secret = "longSecretKeyForParentSts99!";
@@ -353,6 +370,11 @@ async fn test_e2e_sts_assume_role_session_policy_existing_object_tag() -> Result
#[tokio::test]
async fn test_e2e_sts_session_policy_delete_objects_object_prefix_only() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
init_logging();
if !awscurl_available() {
info!("Skipping test_e2e_sts_session_policy_delete_objects_object_prefix_only: awscurl not available");
return Ok(());
}
let suffix = Uuid::new_v4();
let parent = format!("e2e-sts-del-par-{suffix}");
let parent_secret = "longSecretKeyForParentDelete99!";
+16 -1
View File
@@ -22,7 +22,9 @@
//! - KMS backend configuration (Local and Vault)
//! - SSE encryption testing utilities
use crate::common::{RustFSTestEnvironment, awscurl_get, awscurl_post, init_logging as common_init_logging, local_http_client};
use crate::common::{
RustFSTestEnvironment, awscurl_available, awscurl_get, awscurl_post, init_logging as common_init_logging, local_http_client,
};
use aws_sdk_s3::Client;
use aws_sdk_s3::primitives::ByteStream;
use aws_sdk_s3::types::ServerSideEncryption;
@@ -57,6 +59,15 @@ pub fn init_logging() {
// Additional KMS-specific logging configuration can be added here if needed
}
pub fn skip_if_kms_admin_tool_unavailable(test_name: &str) -> bool {
if awscurl_available() {
return false;
}
info!("Skipping {} because awscurl is not available in PATH", test_name);
true
}
pub fn sse_customer_key_md5_base64(key: &str) -> String {
let mut hasher = Md5::new();
hasher.update(key.as_bytes());
@@ -479,6 +490,10 @@ pub async fn test_kms_key_management(
access_key: &str,
secret_key: &str,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
if skip_if_kms_admin_tool_unavailable("test_kms_key_management") {
return Ok(());
}
info!("Testing KMS key management APIs");
// Test CreateKey
+5 -1
View File
@@ -20,7 +20,8 @@
//! - Complete encryption/decryption lifecycle
use super::common::{
LocalKMSTestEnvironment, get_kms_status, sse_customer_key_md5_base64, test_kms_key_management, test_sse_c_encryption,
LocalKMSTestEnvironment, get_kms_status, skip_if_kms_admin_tool_unavailable, sse_customer_key_md5_base64,
test_kms_key_management, test_sse_c_encryption,
};
use crate::common::{TEST_BUCKET, init_logging};
use tracing::{error, info};
@@ -28,6 +29,9 @@ use tracing::{error, info};
#[tokio::test]
async fn test_local_kms_end_to_end() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
init_logging();
if skip_if_kms_admin_tool_unavailable("test_local_kms_end_to_end") {
return Ok(());
}
info!("Starting Local KMS End-to-End Test");
// Create LocalKMS test environment
+17 -2
View File
@@ -22,8 +22,8 @@ use crate::common::{TEST_BUCKET, init_logging};
use tracing::{error, info};
use super::common::{
VAULT_KEY_NAME, VaultTestEnvironment, get_kms_status, sse_customer_key_md5_base64, start_kms,
test_all_multipart_encryption_types, test_error_scenarios, test_kms_key_management, test_sse_c_encryption,
VAULT_KEY_NAME, VaultTestEnvironment, get_kms_status, skip_if_kms_admin_tool_unavailable, sse_customer_key_md5_base64,
start_kms, test_all_multipart_encryption_types, test_error_scenarios, test_kms_key_management, test_sse_c_encryption,
test_sse_kms_encryption, test_sse_s3_encryption,
};
@@ -62,6 +62,9 @@ impl VaultKmsTestContext {
#[tokio::test]
async fn test_vault_kms_end_to_end() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
init_logging();
if skip_if_kms_admin_tool_unavailable("test_vault_kms_end_to_end") {
return Ok(());
}
info!("Starting Vault KMS End-to-End Test with default key {}", VAULT_KEY_NAME);
let context = VaultKmsTestContext::new().await?;
@@ -114,6 +117,9 @@ async fn test_vault_kms_end_to_end() -> Result<(), Box<dyn std::error::Error + S
#[tokio::test]
async fn test_vault_kms_key_isolation() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
init_logging();
if skip_if_kms_admin_tool_unavailable("test_vault_kms_key_isolation") {
return Ok(());
}
info!("Starting Vault KMS SSE-C key isolation test");
let context = VaultKmsTestContext::new().await?;
@@ -197,6 +203,9 @@ async fn test_vault_kms_key_isolation() -> Result<(), Box<dyn std::error::Error
#[tokio::test]
async fn test_vault_kms_large_file() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
init_logging();
if skip_if_kms_admin_tool_unavailable("test_vault_kms_large_file") {
return Ok(());
}
info!("Starting Vault KMS large file SSE-S3 test");
let context = VaultKmsTestContext::new().await?;
@@ -258,6 +267,9 @@ async fn test_vault_kms_large_file() -> Result<(), Box<dyn std::error::Error + S
#[tokio::test]
async fn test_vault_kms_multipart_upload() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
init_logging();
if skip_if_kms_admin_tool_unavailable("test_vault_kms_multipart_upload") {
return Ok(());
}
info!("Starting Vault KMS multipart upload encryption suite");
let context = VaultKmsTestContext::new().await?;
@@ -285,6 +297,9 @@ async fn test_vault_kms_multipart_upload() -> Result<(), Box<dyn std::error::Err
#[tokio::test]
async fn test_vault_kms_key_operations() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
init_logging();
if skip_if_kms_admin_tool_unavailable("test_vault_kms_key_operations") {
return Ok(());
}
info!("Starting Vault KMS key operations test (CRUD)");
let context = VaultKmsTestContext::new().await?;
@@ -41,6 +41,13 @@ async fn create_issue_3107_fixture(root: &Path) -> TestResult {
Ok(())
}
fn mc_available() -> bool {
Command::new("mc")
.arg("--version")
.output()
.is_ok_and(|output| output.status.success())
}
fn run_mc(args: &[&str]) -> TestResult {
let output = Command::new("mc").args(args).output()?;
if !output.status.success() {
@@ -68,7 +75,10 @@ fn count_files(root: &Path) -> usize {
async fn test_mc_mirror_small_bucket_completes_without_list_timeout() -> TestResult {
crate::common::init_logging();
info!("Starting issue #3107 mc mirror regression test");
run_mc(&["--version"])?;
if !mc_available() {
info!("Skipping issue #3107 mc mirror regression test because mc is not installed");
return Ok(());
}
let mut env = RustFSTestEnvironment::new().await?;
env.start_rustfs_server(vec![]).await?;
@@ -4278,6 +4278,10 @@ async fn test_signed_put_object_extract_preserves_pax_metadata_and_version_id()
async fn test_signed_put_object_extract_authorizes_each_pax_privilege_and_retention_conditions()
-> Result<(), Box<dyn std::error::Error + Send + Sync>> {
init_logging();
if !crate::common::awscurl_available() {
return Ok(());
}
let mut env = RustFSTestEnvironment::new().await?;
env.start_rustfs_server(vec![]).await?;
+51
View File
@@ -18,6 +18,15 @@ use http::{Method, StatusCode};
use tokio::time::{Duration, sleep, timeout};
use tracing::{debug, info};
fn skip_without_awscurl() -> bool {
if crate::common::awscurl_available() {
return false;
}
info!("Skipping quota test because awscurl is not available");
true
}
/// Test environment setup for quota tests
pub struct QuotaTestEnv {
pub env: RustFSTestEnvironment,
@@ -267,6 +276,9 @@ mod integration_tests {
#[tokio::test]
async fn test_quota_basic_operations() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
init_logging();
if skip_without_awscurl() {
return Ok(());
}
let env = QuotaTestEnv::new().await?;
// Create test bucket
@@ -308,6 +320,9 @@ mod integration_tests {
#[tokio::test]
async fn test_quota_admission_aws_chunked_declared_encoding() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
init_logging();
if skip_without_awscurl() {
return Ok(());
}
let env = QuotaTestEnv::new().await?;
env.create_bucket().await?;
@@ -356,6 +371,9 @@ mod integration_tests {
#[tokio::test]
async fn test_quota_update_and_clear() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
init_logging();
if skip_without_awscurl() {
return Ok(());
}
let env = QuotaTestEnv::new().await?;
env.create_bucket().await?;
@@ -388,6 +406,9 @@ mod integration_tests {
#[tokio::test]
async fn test_quota_delete_operations() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
init_logging();
if skip_without_awscurl() {
return Ok(());
}
let env = QuotaTestEnv::new().await?;
env.create_bucket().await?;
@@ -421,6 +442,9 @@ mod integration_tests {
#[tokio::test]
async fn test_quota_usage_tracking() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
init_logging();
if skip_without_awscurl() {
return Ok(());
}
let env = QuotaTestEnv::new().await?;
env.create_bucket().await?;
@@ -456,6 +480,9 @@ mod integration_tests {
#[tokio::test]
async fn test_quota_statistics() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
init_logging();
if skip_without_awscurl() {
return Ok(());
}
let env = QuotaTestEnv::new().await?;
env.create_bucket().await?;
@@ -486,6 +513,9 @@ mod integration_tests {
#[tokio::test]
async fn test_quota_check_api() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
init_logging();
if skip_without_awscurl() {
return Ok(());
}
let env = QuotaTestEnv::new().await?;
env.create_bucket().await?;
@@ -523,6 +553,9 @@ mod integration_tests {
#[tokio::test]
async fn test_quota_multiple_buckets() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
init_logging();
if skip_without_awscurl() {
return Ok(());
}
let env = QuotaTestEnv::new().await?;
// Create two buckets in the same environment
@@ -560,6 +593,9 @@ mod integration_tests {
#[tokio::test]
async fn test_quota_error_handling() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
init_logging();
if skip_without_awscurl() {
return Ok(());
}
let env = QuotaTestEnv::new().await?;
env.create_bucket().await?;
@@ -592,6 +628,9 @@ mod integration_tests {
#[tokio::test]
async fn test_quota_http_endpoints() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
init_logging();
if skip_without_awscurl() {
return Ok(());
}
let env = QuotaTestEnv::new().await?;
env.create_bucket().await?;
@@ -650,6 +689,9 @@ mod integration_tests {
#[tokio::test]
async fn test_quota_normal_user_permissions() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
init_logging();
if skip_without_awscurl() {
return Ok(());
}
let env = QuotaTestEnv::new().await?;
env.create_bucket().await?;
@@ -702,6 +744,9 @@ mod integration_tests {
#[tokio::test]
async fn test_quota_copy_operations() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
init_logging();
if skip_without_awscurl() {
return Ok(());
}
let env = QuotaTestEnv::new().await?;
env.create_bucket().await?;
@@ -744,6 +789,9 @@ mod integration_tests {
#[tokio::test]
async fn test_quota_batch_delete() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
init_logging();
if skip_without_awscurl() {
return Ok(());
}
let env = QuotaTestEnv::new().await?;
env.create_bucket().await?;
@@ -799,6 +847,9 @@ mod integration_tests {
#[tokio::test]
async fn test_quota_multipart_upload() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
init_logging();
if skip_without_awscurl() {
return Ok(());
}
let env = QuotaTestEnv::new().await?;
env.create_bucket().await?;
@@ -13,8 +13,9 @@
// limitations under the License.
use crate::common::{
RustFSTestEnvironment, admin_create_user, awscurl_post_sts_form_urlencoded, init_logging, local_http_client,
replication_fast_env, rustfs_binary_path, signed_request, signed_request_with_client, signed_request_with_session_token,
RustFSTestEnvironment, admin_create_user, awscurl_available, awscurl_post_sts_form_urlencoded, init_logging,
local_http_client, replication_fast_env, rustfs_binary_path, signed_request, signed_request_with_client,
signed_request_with_session_token,
};
use crate::fake_s3_target::{
FAKE_ACCESS_KEY, FAKE_SECRET_KEY, FakeS3Target, FaultAction as FakeTargetFault, Operation as FakeTargetOperation,
@@ -7280,6 +7281,11 @@ async fn test_site_replication_replicates_multiple_service_accounts_real_dual_no
async fn test_site_replication_replicates_service_accounts_created_from_sts_session_real_dual_node() -> TestResult {
init_logging();
if !awscurl_available() {
eprintln!("Skipping STS site replication service-account test because awscurl is unavailable");
return Ok(());
}
let mut source_env = RustFSTestEnvironment::new().await?;
source_env
.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
@@ -21,7 +21,7 @@
//! - SSRF prevention (internal/private endpoints rejected for tiering)
//! - Race condition handling (concurrent writes converge without corruption)
use crate::common::{RustFSTestEnvironment, awscurl_put, init_logging};
use crate::common::{RustFSTestEnvironment, awscurl_available, awscurl_put, init_logging};
use aws_sdk_s3::error::ProvideErrorMetadata;
use aws_sdk_s3::primitives::ByteStream;
use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart, Tag, Tagging};
@@ -225,11 +225,16 @@ async fn test_concurrent_object_operations() -> Result<(), Box<dyn Error + Send
/// outcome — the internal endpoint is not accepted — is asserted here.
///
/// The admin API is exercised via signed `awscurl` requests, matching the
/// pattern used by the other admin-API E2E tests in this crate. The full E2E
/// lane installs and verifies the pinned `awscurl` prerequisite.
/// pattern used by the other admin-API E2E tests in this crate; the test is
/// skipped when `awscurl` is not installed.
#[tokio::test]
async fn test_tiering_url_validation() -> Result<(), Box<dyn Error + Send + Sync>> {
init_logging();
if !awscurl_available() {
info!("Skipping tiering URL validation test because awscurl is not available");
return Ok(());
}
let mut env = RustFSTestEnvironment::new().await?;
env.start_rustfs_server(vec![]).await?;
+33 -4
View File
@@ -54,6 +54,7 @@ const ERR_LIFECYCLE_INVALID_EXPIRED_OBJECT_ALL_VERSIONS: &str =
"Days must be a positive integer and Date must not be specified inside Expiration with ExpiredObjectAllVersions";
const ERR_LIFECYCLE_INVALID_DEL_MARKER_EXPIRATION_DAYS: &str = "Days must be a positive integer with DelMarkerExpiration";
const ERR_LIFECYCLE_INVALID_RULE_ID_TOO_LONG: &str = "Rule ID must be at most 255 characters";
const ERR_LIFECYCLE_INVALID_RULE_ID_EMPTY: &str = "Rule ID must not be empty";
const ERR_LIFECYCLE_INVALID_RULE_STATUS: &str = "Rule status must be either Enabled or Disabled";
const ERR_LIFECYCLE_DEL_MARKER_WITH_TAGS: &str = "Rule with DelMarkerExpiration cannot have tags based filtering";
const ERR_LIFECYCLE_EXPIRED_OBJECT_DELETE_MARKER_WITH_TAGS: &str =
@@ -402,10 +403,13 @@ impl Lifecycle for BucketLifecycleConfiguration {
NoncurrentVersionTransitionOps::validate(transition)?;
}
}
if let Some(id) = &r.id
&& id.len() > 255
{
return Err(std::io::Error::other(ERR_LIFECYCLE_INVALID_RULE_ID_TOO_LONG));
if let Some(id) = &r.id {
if id.is_empty() {
return Err(std::io::Error::other(ERR_LIFECYCLE_INVALID_RULE_ID_EMPTY));
}
if id.len() > 255 {
return Err(std::io::Error::other(ERR_LIFECYCLE_INVALID_RULE_ID_TOO_LONG));
}
}
r.validate()?;
if let Some(object_lock_enabled) = lr.object_lock_enabled.as_ref()
@@ -3730,6 +3734,31 @@ mod tests {
.expect("empty prefix with filter should be valid");
}
#[tokio::test]
async fn validate_rejects_empty_rule_id() {
let lc = BucketLifecycleConfiguration {
expiry_updated_at: None,
rules: vec![LifecycleRule {
status: ExpirationStatus::from_static(ExpirationStatus::ENABLED),
expiration: Some(LifecycleExpiration {
days: Some(30),
..Default::default()
}),
abort_incomplete_multipart_upload: None,
del_marker_expiration: None,
filter: None,
id: Some(String::new()),
noncurrent_version_expiration: None,
noncurrent_version_transitions: None,
prefix: None,
transitions: None,
}],
};
let error = lc.validate(&ObjectLockConfiguration::default()).await.unwrap_err();
assert_eq!(error.to_string(), ERR_LIFECYCLE_INVALID_RULE_ID_EMPTY);
}
// --- TASK-004 tests: ExpiredObjectAllVersions ---
#[tokio::test]
+50 -5
View File
@@ -29,7 +29,8 @@ use rustfs_config::ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS;
pub use rustfs_data_usage::{
AllTierStats, BucketTargetUsageInfo, BucketUsageInfo, DATA_USAGE_OBJECT_NAME, DATA_USAGE_OBSERVED_OBJECT_NAME,
DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageInfo, LEGACY_DATA_USAGE_OBJECT_NAME, PrefixUsageEntry,
PrefixUsageQuery, PrefixUsageSummary, ReplTargetSizeSummary, SizeSummary, TierStats, hash_path, prefix_usage_in_cache,
PrefixUsageQuery, PrefixUsageSummary, ReplTargetSizeSummary, SizeReconciliationEntry, SizeReconciliationScope, SizeSummary,
TierStats, hash_path, prefix_usage_in_cache,
};
use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
use tokio::time::{Duration, Instant, sleep, timeout};
@@ -192,6 +193,10 @@ const MAX_DATA_USAGE_CACHE_DEPTH: usize = 1024;
pub trait ScannerSizeSummaryExt {
/// Fold one object's contribution into the summary, including its tier.
fn actions_accounting(&mut self, oi: &ObjectInfo, size: i64, actual_size: i64);
/// Fold counters and physical tier usage for an object whose metadata is
/// valid but whose logical size is currently unavailable. Logical totals
/// stay unchanged.
fn actions_accounting_unknown(&mut self, oi: &ObjectInfo);
}
impl ScannerSizeSummaryExt for SizeSummary {
@@ -225,6 +230,34 @@ impl ScannerSizeSummaryExt for SizeSummary {
});
}
}
fn actions_accounting_unknown(&mut self, oi: &ObjectInfo) {
if oi.delete_marker {
self.delete_markers = self.delete_markers.saturating_add(1);
return;
}
if oi.version_id.is_some_and(|v| !v.is_nil()) {
self.versions = self.versions.saturating_add(1);
}
if oi.transitioned_object.free_version {
return;
}
let tier = if oi.transitioned_object.status == TRANSITION_COMPLETE {
oi.transitioned_object.tier.clone()
} else {
oi.storage_class.clone().unwrap_or_else(|| storageclass::STANDARD.to_string())
};
if let Some(tier_stats) = self.tier_stats.get_mut(&tier) {
*tier_stats = tier_stats.add(&TierStats {
total_size: u64::try_from(oi.size).unwrap_or(0),
num_versions: 1,
num_objects: u64::from(oi.is_latest),
});
}
}
}
// ===== Cache-related data structures =====
@@ -344,6 +377,10 @@ pub struct DataUsageCacheInfo {
pub scan_plan_digest: Option<DataUsageScanPlanDigest>,
#[serde(default)]
pub cache_key_format: u16,
/// Bounded durable debts for versions whose logical size was not trusted.
/// The map key is an identity key, never a user-controlled metric label.
#[serde(default)]
pub size_reconciliation: HashMap<String, SizeReconciliationEntry>,
}
impl Serialize for DataUsageCacheInfo {
@@ -353,7 +390,8 @@ impl Serialize for DataUsageCacheInfo {
{
// Keep this metadata map-encoded so older readers can ignore fields
// appended by newer scanner versions during rolling upgrades.
let mut state = serializer.serialize_map(Some(16))?;
let field_count = 16 + usize::from(!self.size_reconciliation.is_empty());
let mut state = serializer.serialize_map(Some(field_count))?;
state.serialize_entry("name", &self.name)?;
state.serialize_entry("next_cycle", &self.next_cycle)?;
state.serialize_entry("leader_epoch", &self.leader_epoch)?;
@@ -370,6 +408,9 @@ impl Serialize for DataUsageCacheInfo {
state.serialize_entry("snapshot_complete", &self.snapshot_complete)?;
state.serialize_entry("scan_plan_digest", &self.scan_plan_digest)?;
state.serialize_entry("cache_key_format", &self.cache_key_format)?;
if !self.size_reconciliation.is_empty() {
state.serialize_entry("size_reconciliation", &self.size_reconciliation)?;
}
state.end()
}
}
@@ -428,14 +469,18 @@ impl DataUsageCache {
self.checked_flatten(name).is_some()
});
if !reusable {
let pending_heals = if self.info.name == name {
std::mem::take(&mut self.info.pending_heals)
let (pending_heals, size_reconciliation) = if self.info.name == name {
(
std::mem::take(&mut self.info.pending_heals),
std::mem::take(&mut self.info.size_reconciliation),
)
} else {
Vec::new()
(Vec::new(), HashMap::new())
};
*self = Self::default();
self.info.name = name.to_string();
self.info.pending_heals = pending_heals;
self.info.size_reconciliation = size_reconciliation;
}
self.info.next_cycle = next_cycle;
@@ -673,6 +673,34 @@ fn size_summary_actions_accounting_accumulates_tier_stats() {
);
}
#[test]
fn size_summary_unknown_accounting_keeps_physical_tier_and_version_only() {
let mut summary = SizeSummary::default();
summary
.tier_stats
.insert(storageclass::STANDARD.to_string(), TierStats::default());
let object = ObjectInfo {
size: 12,
storage_class: Some(storageclass::STANDARD.to_string()),
version_id: Some(uuid::Uuid::new_v4()),
is_latest: true,
..Default::default()
};
summary.actions_accounting_unknown(&object);
assert_eq!(summary.total_size, 0, "unknown logical size must not become zero or physical bytes");
assert_eq!(summary.versions, 1);
assert_eq!(
summary.tier_stats.get(storageclass::STANDARD),
Some(&TierStats {
total_size: 12,
num_versions: 1,
num_objects: 1,
})
);
}
#[test]
fn test_data_usage_entry_merge_sums_failed_objects() {
let mut left = DataUsageEntry {
@@ -1079,6 +1107,16 @@ fn data_usage_cache_prepare_for_scan_preserves_pending_heal_only_progress() {
scan_plan_digest: Some(TEST_PLAN_DIGEST),
cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
pending_heals: vec![pending_heal.clone()],
size_reconciliation: HashMap::from([(
"size-key".to_string(),
SizeReconciliationEntry {
key: "size-key".to_string(),
bucket: "bucket".to_string(),
object: "prefix/object".to_string(),
reason: "invalid_declared_size".to_string(),
..Default::default()
},
)]),
..Default::default()
},
..Default::default()
@@ -1088,6 +1126,7 @@ fn data_usage_cache_prepare_for_scan_preserves_pending_heal_only_progress() {
assert_eq!(outcome, DataUsageCachePrepareOutcome::Reused);
assert_eq!(cache.info.pending_heals, vec![pending_heal]);
assert!(cache.info.size_reconciliation.contains_key("size-key"));
assert!(cache.cache.is_empty());
assert!(!cache.info.snapshot_complete);
}
+129 -5
View File
@@ -20,8 +20,9 @@ use std::time::{Duration, Instant, SystemTime};
use crate::ReplTargetSizeSummary;
use crate::data_usage_define::{
DATA_USAGE_SCAN_CHECKPOINT_VERSION, DataUsageCache, DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageScanCheckpoint,
DataUsageScanCheckpointReason, PendingScannerHeal, PendingScannerHealKind, ScannerSizeSummaryExt, SizeSummary, hash_path,
DATA_USAGE_SCAN_CHECKPOINT_VERSION, DataUsageCache, DataUsageCacheInfo, DataUsageEntry, DataUsageHash, DataUsageHashMap,
DataUsageScanCheckpoint, DataUsageScanCheckpointReason, PendingScannerHeal, PendingScannerHealKind, ScannerSizeSummaryExt,
SizeReconciliationEntry, SizeSummary, hash_path,
};
use crate::error::ScannerError;
use crate::runtime_config::{
@@ -97,6 +98,9 @@ const METRIC_SCANNER_EXCESS_FOLDERS_TOTAL: &str = "rustfs_scanner_excess_folders
const METRIC_SCANNER_PENDING_HEAL_PRUNE_TOTAL: &str = "rustfs_scanner_pending_heal_prune_total";
const METRIC_SCANNER_PENDING_HEAL_MALFORMED_TOTAL: &str = "rustfs_scanner_pending_heal_malformed_total";
const MAX_PENDING_SCANNER_HEAL_RETRIES_PER_BUCKET: usize = 128;
const MAX_SIZE_RECONCILIATION_ENTRIES_PER_BUCKET: usize = 10_000;
const MAX_SIZE_RECONCILIATION_BYTES_PER_BUCKET: usize = 8 * 1024 * 1024;
const MAX_SIZE_RECONCILIATION_AGE_SECS: u64 = 7 * 24 * 60 * 60;
// --- scanner excess alerts as S3 notification events (rustfs/backlog#1868) --
//
@@ -364,7 +368,7 @@ impl PendingScannerAccounting<'_> {
fn apply(self, size_summary: &mut SizeSummary, cumulative_size: &mut i64, queued: bool) {
let size = if queued { self.expired_size } else { self.retained_size };
size_summary.actions_accounting(self.object, size, self.retained_size);
*cumulative_size += size;
*cumulative_size = cumulative_size.saturating_add(size);
}
}
@@ -671,10 +675,65 @@ pub struct FolderScanner {
skip_heal: Arc<std::sync::atomic::AtomicBool>,
local_disk: Arc<Disk>,
pending_heals_changed: bool,
pending_size_reconciliation_keys: HashSet<String>,
pending_size_reconciliation_scopes: HashSet<String>,
pending_size_reconciliation_truncated: bool,
#[cfg(test)]
list_path_raw_options_observer: Option<mpsc::UnboundedSender<ListPathRawTimeoutSnapshot>>,
}
fn size_reconciliation_entry_bytes(entry: &SizeReconciliationEntry) -> usize {
entry.key.len()
+ entry.bucket.len()
+ entry.object.len()
+ entry.version_id.as_deref().map_or(0, str::len)
+ entry.generation.as_deref().map_or(0, str::len)
+ entry.reason.len()
+ std::mem::size_of::<u64>()
+ std::mem::size_of::<u32>()
}
fn size_reconciliation_scope_key(bucket: &str, object: &str) -> String {
format!("{}:{}|{}:{}", bucket.len(), bucket, object.len(), object)
}
fn prune_size_reconciliation(info: &mut DataUsageCacheInfo, now: u64) {
info.size_reconciliation.retain(|key, entry| {
if entry.first_seen == 0 || entry.first_seen > now {
entry.first_seen = now;
}
key == &entry.key
&& entry.key.len() <= 4096
&& entry.bucket.len() <= 512
&& entry.object.len() <= 512
&& entry.version_id.as_deref().is_none_or(|value| value.len() <= 64)
&& entry.generation.as_deref().is_none_or(|value| value.len() <= 64)
&& entry.reason.len() <= 64
&& now.saturating_sub(entry.first_seen) <= MAX_SIZE_RECONCILIATION_AGE_SECS
});
while info.size_reconciliation.len() > MAX_SIZE_RECONCILIATION_ENTRIES_PER_BUCKET
|| info
.size_reconciliation
.values()
.map(size_reconciliation_entry_bytes)
.sum::<usize>()
> MAX_SIZE_RECONCILIATION_BYTES_PER_BUCKET
{
let oldest = info
.size_reconciliation
.iter()
.min_by(|(left_key, left), (right_key, right)| {
left.first_seen.cmp(&right.first_seen).then_with(|| left_key.cmp(right_key))
})
.map(|(key, _)| key.clone());
let Some(oldest) = oldest else {
break;
};
info.size_reconciliation.remove(&oldest);
}
}
impl FolderScanner {
fn now_secs() -> u64 {
SystemTime::now()
@@ -748,6 +807,60 @@ impl FolderScanner {
}
}
/// Apply the per-object size-resolution ledger updates in one place. The
/// scanner cache is the durable boundary; both working copies are updated
/// so an incremental publication cannot lose a debt or its resolution.
fn apply_size_reconciliation(&mut self, summary: &SizeSummary) {
let now = Self::now_secs();
self.pending_size_reconciliation_keys
.extend(summary.size_reconciliation.iter().map(|entry| entry.key.clone()));
self.pending_size_reconciliation_scopes.extend(
summary
.reconciliation_scopes
.iter()
.map(|scope| size_reconciliation_scope_key(&scope.bucket, &scope.object)),
);
self.pending_size_reconciliation_truncated |= summary.size_reconciliation_truncated;
for info in [&mut self.new_cache.info, &mut self.update_cache.info] {
for incoming in &summary.size_reconciliation {
if let Some(existing) = info.size_reconciliation.get_mut(&incoming.key) {
existing.reason = incoming.reason.clone();
existing.physical_size = incoming.physical_size;
existing.generation = incoming.generation.clone();
existing.version_id = incoming.version_id.clone();
existing.attempts = existing.attempts.saturating_add(1);
continue;
}
if size_reconciliation_entry_bytes(incoming) > MAX_SIZE_RECONCILIATION_BYTES_PER_BUCKET {
continue;
}
let mut entry = incoming.clone();
entry.first_seen = now;
entry.attempts = 1;
info.size_reconciliation.insert(entry.key.clone(), entry);
}
}
}
fn finish_size_reconciliation_batch(&mut self) {
let now = Self::now_secs();
let current_keys = std::mem::take(&mut self.pending_size_reconciliation_keys);
let scopes = std::mem::take(&mut self.pending_size_reconciliation_scopes);
let truncated = std::mem::replace(&mut self.pending_size_reconciliation_truncated, false);
for info in [&mut self.new_cache.info, &mut self.update_cache.info] {
if !truncated {
info.size_reconciliation.retain(|key, entry| {
!scopes.contains(&size_reconciliation_scope_key(&entry.bucket, &entry.object)) || current_keys.contains(key)
});
}
prune_size_reconciliation(info, now);
}
}
fn record_scan_resume_hint(&mut self, folder: &str) {
self.new_cache.info.scan_resume_after = Some(folder.to_string());
self.update_cache.info.scan_resume_after = Some(folder.to_string());
@@ -1426,6 +1539,7 @@ impl FolderScanner {
abandoned_children.remove(&path_join_buf(&[&item.bucket, &item.object_path()]));
apply_scanner_size_summary(into, &sz);
self.apply_size_reconciliation(&sz);
into.objects += 1;
object_count += 1;
self.budget.record_object_scanned();
@@ -2105,6 +2219,7 @@ impl FolderScanner {
}
}
self.finish_size_reconciliation_batch();
done_folder();
let scanned_objects = u64::try_from(into.objects).unwrap_or(u64::MAX);
emit_scanner_folder_trace(&self.root, &folder.name, scanned_objects, trace_started_at, "completed");
@@ -2190,10 +2305,17 @@ pub async fn scan_data_folder(
skip_heal,
local_disk,
pending_heals_changed: false,
pending_size_reconciliation_keys: HashSet::new(),
pending_size_reconciliation_scopes: HashSet::new(),
pending_size_reconciliation_truncated: false,
#[cfg(test)]
list_path_raw_options_observer: None,
};
let now = FolderScanner::now_secs();
prune_size_reconciliation(&mut scanner.new_cache.info, now);
prune_size_reconciliation(&mut scanner.update_cache.info, now);
// Check if context is cancelled
if ctx.is_cancelled() {
return Err(ScannerError::Other("Operation cancelled".to_string()));
@@ -2217,7 +2339,9 @@ pub async fn scan_data_folder(
new_cache.force_compact(DATA_SCANNER_COMPACT_AT_CHILDREN);
new_cache.info.last_update = Some(SystemTime::now());
new_cache.info.next_cycle = cache.info.next_cycle;
let unresolved_objects = root.failed_objects > 0 || !new_cache.info.failed_objects.is_empty();
let unresolved_objects = root.failed_objects > 0
|| !new_cache.info.failed_objects.is_empty()
|| !new_cache.info.size_reconciliation.is_empty();
new_cache.info.snapshot_complete = !unresolved_objects;
let had_scan_checkpoint = cache.info.scan_checkpoint.is_some() || new_cache.info.scan_checkpoint.is_some();
new_cache.info.scan_resume_after = None;
@@ -2245,7 +2369,7 @@ pub async fn scan_data_folder(
if root_has_progress {
new_cache.replace_hashed(&root_hash, &None, &root);
}
if partial_cache_is_useful(&root, pending_heals_changed) {
if partial_cache_is_useful(&root, pending_heals_changed) || !new_cache.info.size_reconciliation.is_empty() {
if new_cache.root().is_some() {
new_cache.force_compact(DATA_SCANNER_COMPACT_AT_CHILDREN);
}
+809 -65
View File
@@ -13,6 +13,7 @@
// limitations under the License.
/// Per-object scan actions: ScannerItem, the get-size failure policy, and the heal/ILM admission helpers.
use super::*;
use sha2::{Digest as _, Sha256};
/// Cached folder information for scanning
#[derive(Clone, Debug)]
@@ -32,6 +33,263 @@ pub(super) enum GetSizeFailureAction {
HealMetadata { object: String },
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum SizeResolutionReason {
CompressedSizeUnknown,
InvalidPhysicalSize,
UnsupportedCompression,
InvalidObjectSize,
InvalidPartSize,
InvalidDeclaredSize,
SizeOverflowOrMismatch,
}
impl SizeResolutionReason {
fn as_str(self) -> &'static str {
match self {
Self::CompressedSizeUnknown => "compressed_size_unknown",
Self::InvalidPhysicalSize => "invalid_physical_size",
Self::UnsupportedCompression => "unsupported_compression",
Self::InvalidObjectSize => "invalid_object_size",
Self::InvalidPartSize => "invalid_part_size",
Self::InvalidDeclaredSize => "invalid_declared_size",
Self::SizeOverflowOrMismatch => "size_overflow_or_mismatch",
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub(super) enum SizeResolution {
Known { logical: i64, physical: i64 },
Unknown { physical: i64, reason: SizeResolutionReason },
Corrupt { physical: i64, reason: SizeResolutionReason },
}
impl SizeResolution {
fn known_size(&self) -> Option<i64> {
match self {
Self::Known { logical, .. } => Some(*logical),
Self::Unknown { .. } | Self::Corrupt { .. } => None,
}
}
}
fn size_reconciliation_key(oi: &ObjectInfo, reason: SizeResolutionReason) -> String {
let version = oi
.version_id
.filter(|version| !version.is_nil())
.map(|version| version.to_string())
.unwrap_or_default();
let generation = oi
.data_dir
.filter(|generation| !generation.is_nil())
.map(|generation| generation.to_string())
.unwrap_or_default();
// Length-prefix each component so an object key containing the separator
// cannot alias another identity. S3 keys are bounded in normal operation;
// oversized persisted values use a digest so a corrupt metadata record
// cannot grow the ledger without bound.
fn component(value: &str) -> String {
const MAX_COMPONENT_LEN: usize = 512;
if value.len() <= MAX_COMPONENT_LEN {
return format!("{}:{}", value.len(), value);
}
let digest = Sha256::digest(value.as_bytes());
let digest = hex_simd::encode_to_string(digest, hex_simd::AsciiCase::Lower);
format!("hash:{}:{}", value.len(), digest)
}
format!(
"{}|{}|{}|{}|{}",
component(&oi.bucket),
component(&oi.name),
component(&version),
component(&generation),
component(reason.as_str())
)
}
pub(super) fn bounded_reconciliation_field(value: &str) -> String {
const MAX_FIELD_LEN: usize = 512;
if value.len() <= MAX_FIELD_LEN {
return value.to_string();
}
let digest = hex_simd::encode_to_string(Sha256::digest(value.as_bytes()), hex_simd::AsciiCase::Lower);
let prefix_len = MAX_FIELD_LEN - 65;
let prefix = value
.char_indices()
.take_while(|(offset, ch)| offset.saturating_add(ch.len_utf8()) <= prefix_len)
.map(|(_, ch)| ch)
.collect::<String>();
format!("{}~{}", prefix, digest)
}
fn record_size_resolution(summary: &mut SizeSummary, oi: &ObjectInfo, resolution: &SizeResolution) {
match resolution {
SizeResolution::Known { .. } => {}
SizeResolution::Unknown { physical, reason } | SizeResolution::Corrupt { physical, reason } => {
summary.record_size_reconciliation(SizeReconciliationEntry {
key: size_reconciliation_key(oi, *reason),
bucket: bounded_reconciliation_field(&oi.bucket),
object: bounded_reconciliation_field(&oi.name),
version_id: oi
.version_id
.filter(|version| !version.is_nil())
.map(|version| version.to_string()),
generation: oi
.data_dir
.filter(|generation| !generation.is_nil())
.map(|generation| generation.to_string()),
reason: reason.as_str().to_string(),
physical_size: u64::try_from(*physical).ok(),
first_seen: 0,
attempts: 0,
});
}
}
}
/// Resolve the size metadata once at the scanner trust boundary. A compressed
/// -1 sentinel is valid legacy metadata, but it cannot participate in normal
/// logical-size accounting or size-filtered lifecycle rules.
pub(super) fn resolve_size(oi: &ObjectInfo) -> SizeResolution {
let physical = oi.size;
if physical < 0 {
return SizeResolution::Corrupt {
physical,
reason: SizeResolutionReason::InvalidPhysicalSize,
};
}
let compressed = match oi.compression_read_plan() {
Ok((_, _, compressed)) => compressed,
Err(_) => {
return SizeResolution::Corrupt {
physical,
reason: SizeResolutionReason::UnsupportedCompression,
};
}
};
if oi.actual_size < -1 || (oi.actual_size == -1 && !compressed) {
return SizeResolution::Corrupt {
physical,
reason: SizeResolutionReason::InvalidObjectSize,
};
}
// Match ObjectInfo::get_actual_size: a positive in-memory value is the
// authoritative decoded size. Stale declared/part metadata must not turn
// an otherwise valid object into a false corruption report.
if oi.actual_size > 0 {
return SizeResolution::Known {
logical: oi.actual_size,
physical,
};
}
if oi
.parts
.iter()
.any(|part| part.actual_size < -1 || (part.actual_size < 0 && !compressed))
{
return SizeResolution::Corrupt {
physical,
reason: SizeResolutionReason::InvalidPartSize,
};
}
let declared = rustfs_utils::http::get_str(&oi.user_defined, rustfs_utils::http::SUFFIX_ACTUAL_SIZE);
let declared = match declared {
Some(value) if value.is_empty() => {
return SizeResolution::Corrupt {
physical,
reason: SizeResolutionReason::InvalidDeclaredSize,
};
}
Some(value) => match value.parse::<i64>() {
Ok(value) if value >= 0 => Some(value),
_ => {
return SizeResolution::Corrupt {
physical,
reason: SizeResolutionReason::InvalidDeclaredSize,
};
}
},
None => None,
};
let logical = match oi.get_actual_size() {
Ok(size) if size == -1 && compressed && declared.is_none() => {
return SizeResolution::Unknown {
physical,
reason: SizeResolutionReason::CompressedSizeUnknown,
};
}
Ok(size) if size >= 0 => size,
Ok(_) | Err(_) => {
return SizeResolution::Corrupt {
physical,
reason: SizeResolutionReason::SizeOverflowOrMismatch,
};
}
};
if compressed && logical == 0 && physical != 0 && oi.parts.is_empty() && declared.is_none() {
return SizeResolution::Corrupt {
physical,
reason: SizeResolutionReason::SizeOverflowOrMismatch,
};
}
SizeResolution::Known { logical, physical }
}
fn resolve_sizes(object_infos: &[ObjectInfo]) -> Vec<SizeResolution> {
object_infos.iter().map(resolve_size).collect()
}
fn lifecycle_rule_has_size_filter(lifecycle: &BucketLifecycleConfiguration, rule_id: &str) -> bool {
let filter_has_size = |filter: &s3s::dto::LifecycleRuleFilter| {
filter.object_size_greater_than.is_some()
|| filter.object_size_less_than.is_some()
|| filter
.and
.as_ref()
.is_some_and(|and| and.object_size_greater_than.is_some() || and.object_size_less_than.is_some())
};
lifecycle
.rules
.iter()
.find(|rule| {
if rule_id.is_empty() {
rule.id.as_deref().is_none_or(str::is_empty)
} else {
rule.id.as_deref() == Some(rule_id)
}
})
.and_then(|rule| rule.filter.as_ref())
.is_some_and(filter_has_size)
}
fn lifecycle_event_allowed(resolution: &SizeResolution, event: &Event, lifecycle: &BucketLifecycleConfiguration) -> bool {
match resolution {
// Missing or invalid logical size only defers actions whose selected
// rule actually depends on that size. Time/version-only actions retain
// their existing semantics, including intrinsic events without a rule ID.
SizeResolution::Unknown { .. } | SizeResolution::Corrupt { .. } => {
!lifecycle_rule_has_size_filter(lifecycle, &event.rule_id)
}
SizeResolution::Known { .. } => true,
}
}
/// A successful newer-noncurrent batch consumes both known and unresolved
/// versions from the retained-version alert count. The two accounting paths
/// are separate because only known sizes can contribute byte totals.
fn remaining_versions_after_queued_noncurrent(remaining_versions: usize, known_count: usize, unknown_count: usize) -> usize {
remaining_versions.saturating_sub(known_count.saturating_add(unknown_count))
}
/// How the corrupt-metadata branch records the repair after attempting an
/// MRF intent (backlog#1894 axis A).
#[derive(Debug, PartialEq, Eq)]
@@ -319,34 +577,48 @@ impl ScannerItem {
"Scanner lifecycle evaluation started"
);
let resolved_sizes = resolve_sizes(&object_infos);
if let Some(first) = object_infos.first() {
size_summary.record_reconciliation_scope(
&bounded_reconciliation_field(&first.bucket),
&bounded_reconciliation_field(&first.name),
);
}
for (oi, resolution) in object_infos.iter().zip(resolved_sizes.iter()) {
record_size_resolution(size_summary, oi, resolution);
}
let has_corrupt_size = resolved_sizes
.iter()
.any(|resolution| matches!(resolution, SizeResolution::Corrupt { .. }));
// `versioning_config` is resolved once per object by the caller
// (`get_size`) and handed in; only `prefix_enabled` is consulted here.
let Some(lifecycle) = self.lifecycle.as_ref() else {
let mut cumulative_size = 0;
for oi in object_infos.iter() {
let actual_size = match oi.get_actual_size() {
Ok(size) => size,
Err(_) => {
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_LIFECYCLE_ACTION,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_LIFECYCLE,
bucket = %self.bucket,
object = %oi.name,
state = "size_lookup_failed",
"Scanner lifecycle action used fallback size"
);
let Some(lifecycle) = self.lifecycle.clone() else {
let mut cumulative_size: i64 = 0;
for (oi, resolved_size) in object_infos.iter().zip(resolved_sizes.iter()) {
let accounting_size = match resolved_size {
SizeResolution::Known { logical, .. } => *logical,
// A valid compressed legacy sentinel has no logical size,
// but heal and replication still need to run. The
// physical size is only an input to those operations; it
// is not folded into the logical total below.
SizeResolution::Unknown { physical, .. } => {
self.heal_actions(oi, *physical, size_summary).await;
size_summary.actions_accounting_unknown(oi);
continue;
}
SizeResolution::Corrupt { .. } => {
size_summary.actions_accounting_unknown(oi);
continue;
}
};
let size = self.heal_actions(oi, actual_size, size_summary).await;
let size = self.heal_actions(oi, accounting_size, size_summary).await;
size_summary.actions_accounting(oi, size, actual_size);
size_summary.actions_accounting(oi, size, accounting_size);
cumulative_size += size;
cumulative_size = cumulative_size.saturating_add(size);
}
self.alert_excessive_versions(object_infos.len(), cumulative_size);
@@ -400,25 +672,108 @@ impl ScannerItem {
let mut to_delete_objs: Vec<ObjectToDelete> = Vec::new();
let mut noncurrent_events: Vec<Event> = Vec::new();
let mut noncurrent_accounting: Vec<PendingScannerAccounting<'_>> = Vec::new();
let mut noncurrent_unknown: Vec<&ObjectInfo> = Vec::new();
let mut cumulative_size = 0;
let mut remaining_versions = object_infos.len();
'eventLoop: {
for (i, event) in events.iter().enumerate() {
let oi = &object_infos[i];
let actual_size = match oi.get_actual_size() {
Ok(size) => size,
Err(_) => {
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_LIFECYCLE_ACTION,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_LIFECYCLE,
bucket = %self.bucket,
object = %oi.name,
state = "size_lookup_failed",
"Scanner lifecycle action used fallback size"
);
0
let known_size = resolved_sizes[i].known_size();
if has_corrupt_size
&& matches!(
event.action,
IlmAction::DeleteAllVersionsAction | IlmAction::DelMarkerDeleteAllVersionsAction
)
{
// An all-version delete would also remove a corrupt
// sibling that could not be reconciled safely.
continue;
}
if !lifecycle_event_allowed(&resolved_sizes[i], event, &lifecycle) {
// An unknown logical size must not make an otherwise
// non-destructive scan disappear from heal/physical-tier
// accounting. Size-filtered or deferred events remain
// pending, so retain the version-only physical counters.
if let SizeResolution::Unknown { physical, .. } = &resolved_sizes[i] {
self.heal_actions(oi, *physical, size_summary).await;
size_summary.actions_accounting_unknown(oi);
}
continue;
}
let actual_size = match known_size {
Some(size) => size,
None => {
match event.action {
IlmAction::DeleteAction
| IlmAction::DeleteRestoredAction
| IlmAction::DeleteRestoredVersionAction
| IlmAction::DeleteAllVersionsAction
| IlmAction::DelMarkerDeleteAllVersionsAction => {
let done_ilm = Metrics::time_ilm(event.action);
let trace_started_at = trace_start_instant();
let queued = apply_expiry_rule(event, &LcEventSrc::Scanner, oi).await;
emit_scanner_ilm_action_trace(&self.bucket, &oi.name, event.action, 1, queued, trace_started_at);
if record_scanner_ilm_action_if_queued(global_metrics(), event.action, 1, queued) {
done_ilm(1)();
if event.action == IlmAction::DeleteAllVersionsAction
|| event.action == IlmAction::DelMarkerDeleteAllVersionsAction
{
remaining_versions = 0;
}
} else if matches!(
event.action,
IlmAction::DeleteAction
| IlmAction::DeleteRestoredAction
| IlmAction::DeleteRestoredVersionAction
) {
size_summary.actions_accounting_unknown(oi);
} else {
size_summary.actions_accounting_unknown(oi);
for (j, retained) in object_infos.iter().enumerate().skip(i + 1) {
match &resolved_sizes[j] {
SizeResolution::Known { logical, .. } => PendingScannerAccounting {
object: retained,
retained_size: *logical,
expired_size: 0,
}
.apply(size_summary, &mut cumulative_size, false),
SizeResolution::Unknown { .. } => {
size_summary.actions_accounting_unknown(retained);
}
SizeResolution::Corrupt { .. } => {}
}
}
}
}
IlmAction::DeleteVersionAction => {
if let Some(opt) = object_opts.get(i) {
to_delete_objs.push(ObjectToDelete {
object_name: opt.name.clone(),
version_id: opt.version_id,
..Default::default()
});
noncurrent_events.push(event.clone());
noncurrent_unknown.push(oi);
}
}
IlmAction::TransitionAction | IlmAction::TransitionVersionAction => {
let trace_started_at = trace_start_instant();
let queued = apply_transition_rule(event, &LcEventSrc::Scanner, oi).await;
emit_scanner_ilm_action_trace(&self.bucket, &oi.name, event.action, 1, queued, trace_started_at);
if record_scanner_ilm_action_if_queued(global_metrics(), event.action, 1, queued) {
let done_ilm = Metrics::time_ilm(event.action);
done_ilm(1)();
}
size_summary.actions_accounting_unknown(oi);
}
IlmAction::NoneAction | IlmAction::ActionCount => {
if let SizeResolution::Unknown { physical, .. } = &resolved_sizes[i] {
self.heal_actions(oi, *physical, size_summary).await;
}
size_summary.actions_accounting_unknown(oi);
}
}
continue;
}
};
@@ -446,36 +801,24 @@ impl ScannerItem {
done_ilm(1)();
remaining_versions = 0;
} else {
PendingScannerAccounting {
object: oi,
retained_size: actual_size,
expired_size: 0,
}
.apply(size_summary, &mut cumulative_size, false);
for retained in object_infos.iter().skip(i + 1) {
let retained_size = match retained.get_actual_size() {
Ok(size) => size,
Err(_) => {
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_LIFECYCLE_ACTION,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_LIFECYCLE,
bucket = %self.bucket,
object = %retained.name,
state = "size_lookup_failed",
"Scanner lifecycle action used fallback size"
);
0
}
};
if let Some(actual_size) = known_size {
PendingScannerAccounting {
object: retained,
retained_size,
object: oi,
retained_size: actual_size,
expired_size: 0,
}
.apply(size_summary, &mut cumulative_size, false);
}
for (j, retained) in object_infos.iter().enumerate().skip(i + 1) {
if let Some(retained_size) = resolved_sizes[j].known_size() {
PendingScannerAccounting {
object: retained,
retained_size,
expired_size: 0,
}
.apply(size_summary, &mut cumulative_size, false);
}
}
}
break 'eventLoop;
}
@@ -511,11 +854,13 @@ impl ScannerItem {
version_id: opt.version_id,
..Default::default()
});
noncurrent_accounting.push(PendingScannerAccounting {
object: oi,
retained_size: actual_size,
expired_size: 0,
});
if let Some(actual_size) = known_size {
noncurrent_accounting.push(PendingScannerAccounting {
object: oi,
retained_size: actual_size,
expired_size: 0,
});
}
account_now = false;
}
noncurrent_events.push(event.clone());
@@ -548,7 +893,7 @@ impl ScannerItem {
if account_now {
size_summary.actions_accounting(oi, size, actual_size);
cumulative_size += size;
cumulative_size = cumulative_size.saturating_add(size);
}
}
}
@@ -576,11 +921,20 @@ impl ScannerItem {
}
if record_scanner_ilm_action_if_queued(global_metrics(), action, count, queued) {
done_ilm(count)();
remaining_versions = remaining_versions.saturating_sub(noncurrent_accounting.len());
remaining_versions = remaining_versions_after_queued_noncurrent(
remaining_versions,
noncurrent_accounting.len(),
noncurrent_unknown.len(),
);
}
for pending in noncurrent_accounting {
pending.apply(size_summary, &mut cumulative_size, queued);
}
if !queued {
for object in noncurrent_unknown {
size_summary.actions_accounting_unknown(object);
}
}
}
self.alert_excessive_versions(remaining_versions, cumulative_size);
}
@@ -929,4 +1283,394 @@ mod tests {
assert_eq!(item.object_name, "object");
assert_eq!(item.object_path(), "object");
}
#[test]
fn size_resolution_rejects_negative_overflow_and_unknown_compression() {
let compressed = |actual_size: i64, declared: Option<&str>| {
let mut user_defined = HashMap::new();
rustfs_utils::http::insert_str(&mut user_defined, rustfs_utils::http::SUFFIX_COMPRESSION, "zstd".to_string());
if let Some(declared) = declared {
rustfs_utils::http::insert_str(&mut user_defined, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, declared.to_string());
}
ObjectInfo {
size: 12,
actual_size,
user_defined: Arc::new(user_defined),
..Default::default()
}
};
let normal = ObjectInfo {
size: 12,
actual_size: 10,
..Default::default()
};
assert_eq!(
resolve_size(&normal),
SizeResolution::Known {
logical: 10,
physical: 12
}
);
let stale_declared_metadata = ObjectInfo {
size: 12,
actual_size: 10,
user_defined: Arc::new(HashMap::from([("x-rustfs-internal-actual-size".to_string(), "not-a-size".to_string())])),
parts: Arc::new(vec![rustfs_filemeta::ObjectPartInfo {
actual_size: -2,
..Default::default()
}]),
..Default::default()
};
assert_eq!(
resolve_size(&stale_declared_metadata),
SizeResolution::Known {
logical: 10,
physical: 12
}
);
assert_eq!(
resolve_size(&compressed(0, Some("9"))),
SizeResolution::Known {
logical: 9,
physical: 12
}
);
assert_eq!(
resolve_size(&compressed(-1, None)),
SizeResolution::Unknown {
physical: 12,
reason: SizeResolutionReason::CompressedSizeUnknown,
}
);
assert!(matches!(
resolve_size(&compressed(0, Some("not-a-size"))),
SizeResolution::Corrupt {
reason: SizeResolutionReason::InvalidDeclaredSize,
..
}
));
assert!(matches!(
resolve_size(&ObjectInfo {
size: 12,
actual_size: -2,
..Default::default()
}),
SizeResolution::Corrupt { .. }
));
assert!(matches!(resolve_size(&compressed(0, Some("-1"))), SizeResolution::Corrupt { .. }));
assert!(matches!(resolve_size(&compressed(0, Some(""))), SizeResolution::Corrupt { .. }));
let unsupported = {
let mut object = compressed(0, None);
let mut metadata = (*object.user_defined).clone();
rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_COMPRESSION, "unsupported".to_string());
object.user_defined = Arc::new(metadata);
object
};
assert!(matches!(resolve_size(&unsupported), SizeResolution::Corrupt { .. }));
let invalid_part = {
let mut object = compressed(0, None);
object.parts = Arc::new(vec![rustfs_filemeta::ObjectPartInfo {
size: 12,
actual_size: -2,
..Default::default()
}]);
object
};
assert!(matches!(resolve_size(&invalid_part), SizeResolution::Corrupt { .. }));
let overflow = {
let mut object = compressed(0, None);
object.parts = Arc::new(vec![
rustfs_filemeta::ObjectPartInfo {
size: 1,
actual_size: i64::MAX,
..Default::default()
},
rustfs_filemeta::ObjectPartInfo {
size: 1,
actual_size: 1,
..Default::default()
},
]);
object
};
assert!(matches!(resolve_size(&overflow), SizeResolution::Corrupt { .. }));
let mismatch = compressed(0, None);
assert!(matches!(resolve_size(&mismatch), SizeResolution::Corrupt { .. }));
assert_eq!(
resolve_size(&ObjectInfo {
size: 0,
actual_size: 0,
..Default::default()
}),
SizeResolution::Known { logical: 0, physical: 0 }
);
}
#[test]
fn size_resolution_records_and_replays_one_identity() {
let version_id = uuid::Uuid::new_v4();
let generation = uuid::Uuid::new_v4();
let mut metadata = HashMap::new();
rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_COMPRESSION, "zstd".to_string());
rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, "not-a-number".to_string());
let corrupt = ObjectInfo {
bucket: "bucket".to_string(),
name: "object".to_string(),
size: 12,
version_id: Some(version_id),
data_dir: Some(generation),
user_defined: Arc::new(metadata),
..Default::default()
};
let mut summary = SizeSummary::default();
let resolution = resolve_size(&corrupt);
record_size_resolution(&mut summary, &corrupt, &resolution);
record_size_resolution(&mut summary, &corrupt, &resolution);
assert_eq!(summary.size_reconciliation.len(), 1);
assert_eq!(summary.size_reconciliation[0].reason, "invalid_declared_size");
assert_eq!(summary.size_reconciliation[0].physical_size, Some(12));
let known = ObjectInfo {
actual_size: 12,
user_defined: Arc::new(HashMap::new()),
..corrupt.clone()
};
record_size_resolution(&mut summary, &known, &resolve_size(&known));
summary.record_reconciliation_scope(&known.bucket, &known.name);
assert_eq!(summary.reconciliation_scopes.len(), 1);
assert_eq!(summary.reconciliation_scopes[0].bucket, "bucket");
}
#[test]
fn malformed_size_has_same_ilm_accounting() {
let mut metadata = HashMap::new();
rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_COMPRESSION, "zstd".to_string());
rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, "invalid".to_string());
let object = ObjectInfo {
bucket: "bucket".to_string(),
name: "object".to_string(),
size: 12,
user_defined: Arc::new(metadata),
..Default::default()
};
let resolution = resolve_size(&object);
let mut without_ilm = SizeSummary::default();
let mut with_ilm = SizeSummary::default();
record_size_resolution(&mut without_ilm, &object, &resolution);
record_size_resolution(&mut with_ilm, &object, &resolution);
assert_eq!(without_ilm.size_reconciliation, with_ilm.size_reconciliation);
assert_eq!(without_ilm.total_size, 0);
assert_eq!(with_ilm.total_size, 0);
assert!(without_ilm.tier_stats.is_empty());
assert!(with_ilm.tier_stats.is_empty());
}
#[test]
fn size_resolution_parses_once_per_version() {
let objects = vec![
ObjectInfo {
bucket: "bucket".to_string(),
name: "one".to_string(),
size: 1,
actual_size: 1,
..Default::default()
},
ObjectInfo {
bucket: "bucket".to_string(),
name: "two".to_string(),
size: 2,
actual_size: -2,
..Default::default()
},
];
let resolutions = resolve_sizes(&objects);
assert_eq!(resolutions.len(), objects.len());
assert!(matches!(resolutions[0], SizeResolution::Known { logical: 1, .. }));
assert!(matches!(resolutions[1], SizeResolution::Corrupt { .. }));
}
#[test]
fn queued_unknown_noncurrent_versions_are_removed_from_alert_count() {
assert_eq!(remaining_versions_after_queued_noncurrent(3, 1, 2), 0);
assert_eq!(remaining_versions_after_queued_noncurrent(7, 2, 1), 4);
assert_eq!(remaining_versions_after_queued_noncurrent(usize::MAX, usize::MAX, usize::MAX), 0);
}
#[test]
fn malformed_size_blocks_size_dependent_transition_but_allows_time_only_expiry() {
let size_filtered = BucketLifecycleConfiguration {
rules: vec![s3s::dto::LifecycleRule {
status: s3s::dto::ExpirationStatus::from_static(s3s::dto::ExpirationStatus::ENABLED),
expiration: None,
abort_incomplete_multipart_upload: None,
del_marker_expiration: None,
id: Some("size".to_string()),
filter: Some(s3s::dto::LifecycleRuleFilter {
object_size_greater_than: Some(1),
..Default::default()
}),
noncurrent_version_expiration: None,
noncurrent_version_transitions: None,
prefix: None,
transitions: None,
}],
..Default::default()
};
let unknown = SizeResolution::Unknown {
physical: 12,
reason: SizeResolutionReason::CompressedSizeUnknown,
};
let size_event = Event {
action: IlmAction::DeleteAction,
rule_id: "size".to_string(),
..Default::default()
};
assert!(!lifecycle_event_allowed(&unknown, &size_event, &size_filtered));
assert!(!lifecycle_event_allowed(
&unknown,
&Event {
action: IlmAction::TransitionAction,
rule_id: "size".to_string(),
..Default::default()
},
&size_filtered
));
let mixed_filters = BucketLifecycleConfiguration {
rules: vec![
size_filtered.rules[0].clone(),
s3s::dto::LifecycleRule {
status: s3s::dto::ExpirationStatus::from_static(s3s::dto::ExpirationStatus::ENABLED),
expiration: None,
abort_incomplete_multipart_upload: None,
del_marker_expiration: None,
id: Some("time".to_string()),
filter: None,
noncurrent_version_expiration: None,
noncurrent_version_transitions: None,
prefix: None,
transitions: None,
},
],
..Default::default()
};
assert!(lifecycle_event_allowed(
&unknown,
&Event {
action: IlmAction::DeleteAction,
rule_id: "time".to_string(),
..Default::default()
},
&mixed_filters
));
assert!(lifecycle_event_allowed(
&unknown,
&Event {
action: IlmAction::TransitionAction,
..Default::default()
},
&BucketLifecycleConfiguration::default()
));
assert!(lifecycle_event_allowed(
&SizeResolution::Corrupt {
physical: 12,
reason: SizeResolutionReason::InvalidDeclaredSize,
},
&Event {
action: IlmAction::DeleteAction,
..Default::default()
},
&BucketLifecycleConfiguration::default()
));
assert!(!lifecycle_event_allowed(
&SizeResolution::Corrupt {
physical: 12,
reason: SizeResolutionReason::InvalidDeclaredSize,
},
&Event {
action: IlmAction::DeleteAction,
rule_id: "size".to_string(),
..Default::default()
},
&size_filtered
));
assert!(lifecycle_rule_has_size_filter(
&BucketLifecycleConfiguration {
rules: vec![s3s::dto::LifecycleRule {
status: s3s::dto::ExpirationStatus::from_static(s3s::dto::ExpirationStatus::ENABLED),
expiration: None,
abort_incomplete_multipart_upload: None,
del_marker_expiration: None,
id: None,
filter: Some(s3s::dto::LifecycleRuleFilter {
object_size_greater_than: Some(1),
..Default::default()
}),
noncurrent_version_expiration: None,
noncurrent_version_transitions: None,
prefix: None,
transitions: None,
}],
..Default::default()
},
""
));
assert!(lifecycle_event_allowed(
&SizeResolution::Known {
logical: 10,
physical: 12,
},
&Event {
action: IlmAction::DeleteAllVersionsAction,
..Default::default()
},
&BucketLifecycleConfiguration::default()
));
assert!(lifecycle_event_allowed(
&unknown,
&Event {
action: IlmAction::DeleteAction,
rule_id: "time-only".to_string(),
..Default::default()
},
&BucketLifecycleConfiguration::default()
));
}
#[tokio::test]
async fn long_object_size_reconciliation_scope_uses_bounded_identity() {
let object_name = "o".repeat(600);
let mut item = scanner_item_with_prefix("");
item.object_name = object_name.clone();
let mut metadata = HashMap::new();
rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_COMPRESSION, "zstd".to_string());
let object = ObjectInfo {
bucket: item.bucket.clone(),
name: object_name.clone(),
size: 12,
actual_size: -1,
version_id: Some(uuid::Uuid::new_v4()),
user_defined: Arc::new(metadata),
..Default::default()
};
let mut summary = SizeSummary::default();
item.apply_actions(vec![object], None, VersioningConfiguration::default(), &mut summary)
.await;
let bounded_bucket = bounded_reconciliation_field(&item.bucket);
let bounded_object = bounded_reconciliation_field(&object_name);
assert_eq!(summary.reconciliation_scopes[0].bucket, bounded_bucket);
assert_eq!(summary.reconciliation_scopes[0].object, bounded_object);
assert_eq!(summary.size_reconciliation[0].object, bounded_object);
assert_eq!(summary.versions, 1);
assert_eq!(summary.total_size, 0);
}
}
@@ -326,6 +326,9 @@ async fn build_test_scanner() -> (FolderScanner, std::path::PathBuf) {
skip_heal: Arc::new(AtomicBool::new(false)),
local_disk: disk,
pending_heals_changed: false,
pending_size_reconciliation_keys: HashSet::new(),
pending_size_reconciliation_scopes: HashSet::new(),
pending_size_reconciliation_truncated: false,
list_path_raw_options_observer: None,
};
@@ -388,6 +391,66 @@ async fn test_record_failed_ttl_zero_noop() {
assert!(!scanner.should_skip_failed("path2"));
}
#[tokio::test]
async fn malformed_size_reconciliation_replays_after_restart() {
let (mut scanner, temp_dir) = build_test_scanner().await;
let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir);
let entry = SizeReconciliationEntry {
key: "1:b|6:object|0:|0:".to_string(),
bucket: "b".to_string(),
object: "object".to_string(),
reason: "invalid_declared_size".to_string(),
physical_size: Some(12),
..Default::default()
};
let mut summary = SizeSummary::default();
summary.record_size_reconciliation(entry.clone());
summary.record_reconciliation_scope("b", "object");
scanner.apply_size_reconciliation(&summary);
scanner.apply_size_reconciliation(&summary);
assert_eq!(scanner.new_cache.info.size_reconciliation.len(), 1);
assert_eq!(scanner.update_cache.info.size_reconciliation.len(), 1);
assert_eq!(scanner.new_cache.info.size_reconciliation[&entry.key].attempts, 2);
let encoded = rmp_serde::to_vec_named(&scanner.new_cache.info).expect("size ledger should encode");
let decoded: crate::data_usage_define::DataUsageCacheInfo =
rmp_serde::from_slice(&encoded).expect("size ledger should decode");
assert_eq!(decoded.size_reconciliation.len(), 1);
assert_eq!(decoded.size_reconciliation[&entry.key].reason, "invalid_declared_size");
let mut resolved = SizeSummary::default();
resolved.record_reconciliation_scope("b", "object");
scanner.apply_size_reconciliation(&resolved);
assert!(scanner.new_cache.info.size_reconciliation.is_empty());
assert!(scanner.update_cache.info.size_reconciliation.is_empty());
}
#[tokio::test]
async fn malformed_size_reconciliation_clears_bounded_long_object_scope() {
let (mut scanner, temp_dir) = build_test_scanner().await;
let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir);
let long_object = "o".repeat(600);
let bounded_object = item_actions::bounded_reconciliation_field(&long_object);
let entry = SizeReconciliationEntry {
key: "long-object-key".to_string(),
bucket: "b".to_string(),
object: bounded_object,
reason: "invalid_declared_size".to_string(),
..Default::default()
};
let mut summary = SizeSummary::default();
summary.record_size_reconciliation(entry);
scanner.apply_size_reconciliation(&summary);
assert_eq!(scanner.new_cache.info.size_reconciliation.len(), 1);
let mut resolved = SizeSummary::default();
resolved.record_reconciliation_scope("b", &long_object);
scanner.apply_size_reconciliation(&resolved);
assert!(scanner.new_cache.info.size_reconciliation.is_empty());
}
#[test]
fn test_classify_get_size_failure_marks_metadata_heal_object_path() {
let temp_dir = std::env::temp_dir();