mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-23 04:39:04 +00:00
Compare commits
6 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 16fc3ff0e7 | |||
| 6dbff1c036 | |||
| 854b6dfb0c | |||
| 893a4a11c1 | |||
| 2086ade967 | |||
| 23b85b792e |
@@ -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.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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(),
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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?;
|
||||
|
||||
@@ -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!";
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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?;
|
||||
|
||||
|
||||
@@ -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?;
|
||||
|
||||
|
||||
@@ -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]
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
@@ -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();
|
||||
|
||||
Reference in New Issue
Block a user