mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-22 12:26:37 +00:00
Compare commits
8 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| c2ce237d5a | |||
| 241271c256 | |||
| 3be1511214 | |||
| bc1caf0dc7 | |||
| ce4a72869d | |||
| 255943fa43 | |||
| 5583e8373c | |||
| e1c657652f |
@@ -39,10 +39,11 @@ jobs:
|
||||
env:
|
||||
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
|
||||
steps:
|
||||
- name: Checkout repository
|
||||
- name: Checkout main branch
|
||||
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
|
||||
with:
|
||||
persist-credentials: false
|
||||
ref: main
|
||||
|
||||
- name: Setup Rust environment
|
||||
uses: ./.github/actions/setup
|
||||
@@ -88,10 +89,11 @@ jobs:
|
||||
# either casing.
|
||||
NO_PROXY: 127.0.0.1,localhost
|
||||
steps:
|
||||
- name: Checkout repository
|
||||
- name: Checkout main branch
|
||||
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
|
||||
with:
|
||||
persist-credentials: false
|
||||
ref: main
|
||||
|
||||
- name: Setup Rust environment
|
||||
uses: ./.github/actions/setup
|
||||
@@ -176,10 +178,11 @@ jobs:
|
||||
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
|
||||
NO_PROXY: 127.0.0.1,localhost
|
||||
steps:
|
||||
- name: Checkout repository
|
||||
- name: Checkout main branch
|
||||
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
|
||||
with:
|
||||
persist-credentials: false
|
||||
ref: main
|
||||
|
||||
- name: Setup Rust environment
|
||||
uses: ./.github/actions/setup
|
||||
|
||||
+840
-252
File diff suppressed because it is too large
Load Diff
@@ -33,7 +33,7 @@ use crate::bucket::utils::check_put_object_part_args;
|
||||
use crate::bucket::utils::{check_valid_bucket_name, check_valid_bucket_name_strict, is_meta_bucketname};
|
||||
use crate::cluster::rpc::{RemoteClient, S3PeerSys};
|
||||
use crate::config::storageclass;
|
||||
use crate::core::pools::PoolMeta;
|
||||
use crate::core::pools::{DecommissionCanceler, PoolMeta};
|
||||
use crate::disk::endpoint::{Endpoint, EndpointType};
|
||||
use crate::disk::{DiskAPI, DiskInfo, DiskInfoOptions};
|
||||
use crate::error::{Error, Result};
|
||||
@@ -176,7 +176,7 @@ pub struct ECStore {
|
||||
// pub local_disks: Vec<DiskStore>,
|
||||
pub pool_meta: RwLock<PoolMeta>,
|
||||
pub rebalance_meta: RwLock<Option<RebalanceMeta>>,
|
||||
pub decommission_cancelers: RwLock<Vec<Option<CancellationToken>>>,
|
||||
pub decommission_cancelers: RwLock<Vec<Option<DecommissionCanceler>>>,
|
||||
/// Serializes rebalance/decommission start transitions.
|
||||
///
|
||||
/// Lock order: acquire `start_gate` before `pool_meta`, `rebalance_meta`,
|
||||
|
||||
@@ -16,14 +16,14 @@
|
||||
//!
|
||||
//! `scripts/test/vault_ha_kms_live.sh` owns the official Vault containers and
|
||||
//! kills the active node while this test continuously decrypts through a
|
||||
//! surviving standby. KV2 and Transit must recover after the bounded circuit
|
||||
//! interval, use a bounded number of attempts, and leave the circuit and
|
||||
//! in-flight gauges at zero after a new leader is elected.
|
||||
//! surviving standby. KV2 and Transit requests must remain successful, use a
|
||||
//! bounded number of attempts, and leave the circuit and in-flight gauges at
|
||||
//! zero after a new leader is elected.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::Duration;
|
||||
|
||||
use metrics_util::MetricKind;
|
||||
@@ -43,11 +43,6 @@ const OPERATION_ATTEMPTS: &str = "rustfs_kms_backend_operation_attempts";
|
||||
const IN_FLIGHT: &str = "rustfs_kms_backend_in_flight";
|
||||
const CIRCUIT_OPEN: &str = "rustfs_kms_backend_circuit_open";
|
||||
const MAX_ATTEMPTS: u32 = 10;
|
||||
const ATTEMPT_TIMEOUT: Duration = Duration::from_secs(2);
|
||||
const HEALTHY_PROGRESS_TIMEOUT: Duration = Duration::from_secs(20);
|
||||
// The circuit remains open for 30s after five failed attempts.
|
||||
const POST_FAILOVER_PROGRESS_TIMEOUT: Duration = Duration::from_secs(35);
|
||||
const FAILOVER_ERROR_POLL_INTERVAL: Duration = Duration::from_millis(100);
|
||||
|
||||
type MetricEntry = (
|
||||
metrics_util::CompositeKey,
|
||||
@@ -69,7 +64,7 @@ fn config(backend: KmsBackend, backend_config: BackendConfig) -> KmsConfig {
|
||||
backend,
|
||||
backend_config,
|
||||
allow_insecure_dev_defaults: true,
|
||||
timeout: ATTEMPT_TIMEOUT,
|
||||
timeout: Duration::from_secs(2),
|
||||
retry_attempts: MAX_ATTEMPTS,
|
||||
enable_cache: false,
|
||||
..KmsConfig::default()
|
||||
@@ -169,31 +164,14 @@ fn retryable_failures(snapshot: &[MetricEntry], operation: &str) -> u64 {
|
||||
.sum()
|
||||
}
|
||||
|
||||
async fn wait_for_count(
|
||||
counter: &AtomicU64,
|
||||
failure: &Mutex<Option<String>>,
|
||||
minimum: u64,
|
||||
description: &str,
|
||||
timeout: Duration,
|
||||
) {
|
||||
tokio::time::timeout(timeout, async {
|
||||
async fn wait_for_count(counter: &AtomicU64, minimum: u64, description: &str) {
|
||||
tokio::time::timeout(Duration::from_secs(20), async {
|
||||
while counter.load(Ordering::SeqCst) < minimum {
|
||||
if let Some(error) = failure.lock().expect("decrypt failure lock poisoned").as_ref() {
|
||||
panic!(
|
||||
"{description} worker failed after {} successful decrypts: {error}",
|
||||
counter.load(Ordering::SeqCst)
|
||||
);
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(25)).await;
|
||||
}
|
||||
})
|
||||
.await
|
||||
.unwrap_or_else(|_| {
|
||||
panic!(
|
||||
"timed out after {timeout:?} waiting for {description}: completed {}, expected {minimum}",
|
||||
counter.load(Ordering::SeqCst)
|
||||
)
|
||||
});
|
||||
.unwrap_or_else(|_| panic!("timed out waiting for {description}"));
|
||||
}
|
||||
|
||||
async fn wait_for_file(path: &Path, description: &str) {
|
||||
@@ -211,8 +189,7 @@ async fn decrypt_loop<B: KmsBackendTrait + Send + Sync + 'static>(
|
||||
request: DecryptRequest,
|
||||
expected: Vec<u8>,
|
||||
completed: Arc<AtomicU64>,
|
||||
allow_failover_errors: Arc<AtomicBool>,
|
||||
failure: Arc<Mutex<Option<String>>>,
|
||||
failed: Arc<AtomicBool>,
|
||||
stop: CancellationToken,
|
||||
) {
|
||||
while !stop.is_cancelled() {
|
||||
@@ -220,18 +197,8 @@ async fn decrypt_loop<B: KmsBackendTrait + Send + Sync + 'static>(
|
||||
Ok(response) if response.plaintext == expected => {
|
||||
completed.fetch_add(1, Ordering::SeqCst);
|
||||
}
|
||||
Ok(_) => {
|
||||
*failure.lock().expect("decrypt failure lock poisoned") =
|
||||
Some("decrypt returned unexpected plaintext".to_string());
|
||||
return;
|
||||
}
|
||||
Err(rustfs_kms::KmsError::BackendError { .. } | rustfs_kms::KmsError::OperationTimedOut { .. })
|
||||
if allow_failover_errors.load(Ordering::SeqCst) =>
|
||||
{
|
||||
tokio::time::sleep(FAILOVER_ERROR_POLL_INTERVAL).await;
|
||||
}
|
||||
Err(error) => {
|
||||
*failure.lock().expect("decrypt failure lock poisoned") = Some(error.to_string());
|
||||
Ok(_) | Err(_) => {
|
||||
failed.store(true, Ordering::SeqCst);
|
||||
return;
|
||||
}
|
||||
}
|
||||
@@ -329,9 +296,7 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
|
||||
);
|
||||
|
||||
let stop = CancellationToken::new();
|
||||
let allow_failover_errors = Arc::new(AtomicBool::new(false));
|
||||
let kv2_failure = Arc::new(Mutex::new(None));
|
||||
let transit_failure = Arc::new(Mutex::new(None));
|
||||
let failed = Arc::new(AtomicBool::new(false));
|
||||
let kv2_completed = Arc::new(AtomicU64::new(0));
|
||||
let transit_completed = Arc::new(AtomicU64::new(0));
|
||||
let kv2_worker = tokio::spawn(decrypt_loop(
|
||||
@@ -339,8 +304,7 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
|
||||
kv2_request,
|
||||
kv2_data_key.plaintext_key,
|
||||
Arc::clone(&kv2_completed),
|
||||
Arc::clone(&allow_failover_errors),
|
||||
Arc::clone(&kv2_failure),
|
||||
Arc::clone(&failed),
|
||||
stop.clone(),
|
||||
));
|
||||
let transit_worker = tokio::spawn(decrypt_loop(
|
||||
@@ -348,21 +312,12 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
|
||||
transit_request,
|
||||
transit_data_key.plaintext_key,
|
||||
Arc::clone(&transit_completed),
|
||||
Arc::clone(&allow_failover_errors),
|
||||
Arc::clone(&transit_failure),
|
||||
Arc::clone(&failed),
|
||||
stop.clone(),
|
||||
));
|
||||
|
||||
wait_for_count(&kv2_completed, &kv2_failure, 2, "two healthy KV2 decrypts", HEALTHY_PROGRESS_TIMEOUT).await;
|
||||
wait_for_count(
|
||||
&transit_completed,
|
||||
&transit_failure,
|
||||
2,
|
||||
"two healthy Transit decrypts",
|
||||
HEALTHY_PROGRESS_TIMEOUT,
|
||||
)
|
||||
.await;
|
||||
allow_failover_errors.store(true, Ordering::SeqCst);
|
||||
wait_for_count(&kv2_completed, 2, "two healthy KV2 decrypts").await;
|
||||
wait_for_count(&transit_completed, 2, "two healthy Transit decrypts").await;
|
||||
std::fs::write(&marker, b"ready").expect("publish failover readiness marker");
|
||||
|
||||
wait_for_file(&elected, "the replacement Vault leader").await;
|
||||
@@ -371,39 +326,18 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
|
||||
|
||||
let kv2_after_election = kv2_completed.load(Ordering::SeqCst) + 2;
|
||||
let transit_after_election = transit_completed.load(Ordering::SeqCst) + 2;
|
||||
wait_for_count(
|
||||
&kv2_completed,
|
||||
&kv2_failure,
|
||||
kv2_after_election,
|
||||
"post-failover KV2 decrypts",
|
||||
POST_FAILOVER_PROGRESS_TIMEOUT,
|
||||
)
|
||||
.await;
|
||||
wait_for_count(
|
||||
&transit_completed,
|
||||
&transit_failure,
|
||||
transit_after_election,
|
||||
"post-failover Transit decrypts",
|
||||
POST_FAILOVER_PROGRESS_TIMEOUT,
|
||||
)
|
||||
.await;
|
||||
wait_for_count(&kv2_completed, kv2_after_election, "post-failover KV2 decrypts").await;
|
||||
wait_for_count(&transit_completed, transit_after_election, "post-failover Transit decrypts").await;
|
||||
|
||||
stop.cancel();
|
||||
kv2_worker.await.expect("KV2 decrypt worker must join");
|
||||
transit_worker.await.expect("Transit decrypt worker must join");
|
||||
assert!(
|
||||
kv2_failure.lock().expect("KV2 failure lock poisoned").is_none(),
|
||||
"no KV2 decrypt may fail or return different plaintext"
|
||||
);
|
||||
assert!(
|
||||
transit_failure.lock().expect("Transit failure lock poisoned").is_none(),
|
||||
"no Transit decrypt may fail or return different plaintext"
|
||||
);
|
||||
assert!(!failed.load(Ordering::SeqCst), "no decrypt may fail or return different plaintext");
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[ignore = "requires a real three-node Vault Raft cluster; run scripts/test/vault_ha_kms_live.sh"]
|
||||
fn vault_raft_leader_failure_recovers_kv2_and_transit_decrypts() {
|
||||
fn vault_raft_leader_failure_preserves_kv2_and_transit_decrypts() {
|
||||
let recorder = DebuggingRecorder::new();
|
||||
let snapshotter = recorder.snapshotter();
|
||||
metrics::with_local_recorder(&recorder, || {
|
||||
@@ -415,6 +349,11 @@ fn vault_raft_leader_failure_recovers_kv2_and_transit_decrypts() {
|
||||
});
|
||||
let snapshot = snapshotter.snapshot().into_vec();
|
||||
|
||||
assert_eq!(
|
||||
counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "circuit_open")]),
|
||||
0,
|
||||
"a bounded leader election must not open the circuit"
|
||||
);
|
||||
assert_eq!(
|
||||
counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "budget_exhausted")]),
|
||||
0,
|
||||
|
||||
@@ -125,34 +125,6 @@ pub(crate) async fn read_config_with_revision<S: ScannerObjectIO>(
|
||||
}
|
||||
}
|
||||
|
||||
/// Read only the object revision without materializing its body.
|
||||
pub(crate) async fn read_config_revision<S: ScannerObjectIO>(store: Arc<S>, path: &str) -> StorageResult<DataUsageCacheRevision> {
|
||||
match store
|
||||
.get_object_reader(
|
||||
RUSTFS_META_BUCKET,
|
||||
path,
|
||||
None,
|
||||
HeaderMap::new(),
|
||||
&ObjectOptions {
|
||||
no_lock: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(reader) => reader
|
||||
.object_info
|
||||
.etag
|
||||
.filter(|etag| !etag.is_empty())
|
||||
.map(DataUsageCacheRevision::Etag)
|
||||
.ok_or_else(|| StorageError::other(format!("scanner config object {path} has no ETag"))),
|
||||
Err(Error::FileNotFound | Error::VolumeNotFound | Error::ObjectNotFound(_, _) | Error::BucketNotFound(_)) => {
|
||||
Ok(DataUsageCacheRevision::Missing)
|
||||
}
|
||||
Err(err) => Err(err),
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub(crate) struct DataUsageCacheRevisions {
|
||||
main: DataUsageCacheRevision,
|
||||
@@ -174,11 +146,6 @@ pub static LEGACY_DATA_USAGE_OBJ_NAME_PATH: LazyLock<String> =
|
||||
pub static DATA_USAGE_BLOOM_NAME_PATH: LazyLock<String> =
|
||||
LazyLock::new(|| format!("{BUCKET_META_PREFIX}{SLASH_SEPARATOR}{DATA_USAGE_BLOOM_NAME}"));
|
||||
|
||||
/// Durable companion object for a cycle-state object which cannot be decoded.
|
||||
/// The primary object is deliberately never replaced or deleted by recovery.
|
||||
pub static DATA_USAGE_BLOOM_RECOVERY_PATH: LazyLock<String> =
|
||||
LazyLock::new(|| format!("{}.recovery-required.json", DATA_USAGE_BLOOM_NAME_PATH.as_str()));
|
||||
|
||||
pub static BACKGROUND_HEAL_INFO_PATH: LazyLock<String> =
|
||||
LazyLock::new(|| format!("{BUCKET_META_PREFIX}{SLASH_SEPARATOR}.background-heal.json"));
|
||||
|
||||
|
||||
@@ -74,7 +74,7 @@ impl DataUsageCache {
|
||||
let loaded = Self::load_cache(store.clone(), name).await?;
|
||||
let backup = match loaded.backup_revision {
|
||||
Some(revision) => Some(revision),
|
||||
None => match read_config_revision(store, &backup_path).await {
|
||||
None => match Self::revision_for_path(store, &backup_path).await {
|
||||
Ok(revision) => Some(revision),
|
||||
Err(err) => {
|
||||
counter!(METRIC_CACHE_BACKUP_REVISION_FAILURE_TOTAL).increment(1);
|
||||
@@ -336,6 +336,33 @@ impl DataUsageCache {
|
||||
}
|
||||
}
|
||||
|
||||
async fn revision_for_path<S: ScannerObjectIO>(store: Arc<S>, path: &str) -> StorageResult<DataUsageCacheRevision> {
|
||||
match store
|
||||
.get_object_reader(
|
||||
RUSTFS_META_BUCKET,
|
||||
path,
|
||||
None,
|
||||
HeaderMap::new(),
|
||||
&ObjectOptions {
|
||||
no_lock: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(reader) => reader
|
||||
.object_info
|
||||
.etag
|
||||
.filter(|etag| !etag.is_empty())
|
||||
.map(DataUsageCacheRevision::Etag)
|
||||
.ok_or_else(|| StorageError::other(format!("scanner cache object {path} has no ETag"))),
|
||||
Err(Error::FileNotFound | Error::VolumeNotFound | Error::ObjectNotFound(_, _) | Error::BucketNotFound(_)) => {
|
||||
Ok(DataUsageCacheRevision::Missing)
|
||||
}
|
||||
Err(err) => Err(err),
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) fn cache_save_timeout() -> Duration {
|
||||
crate::runtime_config::scanner_cache_save_timeout()
|
||||
}
|
||||
|
||||
@@ -75,10 +75,7 @@ pub use remote_scanner::{
|
||||
};
|
||||
pub use runtime_config::{apply_scanner_runtime_config, scanner_runtime_config_status, validate_scanner_runtime_config};
|
||||
pub use rustfs_common::last_minute;
|
||||
pub use scanner::{
|
||||
ScannerCycleRecoveryMarker, ScannerCycleRecoveryStatus, ScannerCycleScheduleStatus, init_data_scanner,
|
||||
reset_scanner_cycle_recovery, scanner_cycle_recovery_status, scanner_cycle_schedule_status, scanner_topology_digest,
|
||||
};
|
||||
pub use scanner::{ScannerCycleScheduleStatus, init_data_scanner, scanner_cycle_schedule_status, scanner_topology_digest};
|
||||
pub use scanner_io::{
|
||||
ScannerDirtyUsageAckError, ScannerDirtyUsageState, acknowledge_dirty_usage_generation, clear_dirty_usage_bucket,
|
||||
record_dirty_usage_bucket, record_scanner_maintenance_change, scanner_activity_epoch, scanner_dirty_usage_state,
|
||||
|
||||
@@ -20,7 +20,7 @@ use std::sync::{Arc, LazyLock, RwLock};
|
||||
|
||||
use crate::data_usage_define::{
|
||||
BACKGROUND_HEAL_INFO_PATH, DATA_USAGE_BLOOM_NAME_PATH, DATA_USAGE_OBJ_NAME_PATH, DATA_USAGE_OBSERVED_OBJ_NAME_PATH,
|
||||
DataUsageCache, DataUsageCacheRevision, LEGACY_DATA_USAGE_OBJ_NAME_PATH, read_config_revision, read_config_with_revision,
|
||||
DataUsageCache, DataUsageCacheRevision, LEGACY_DATA_USAGE_OBJ_NAME_PATH, read_config_with_revision,
|
||||
};
|
||||
use crate::runtime_config::{
|
||||
ScannerRuntimeConfig, ScannerRuntimeConfigSource, refresh_scanner_runtime_config_from_global, scanner_bitrot_cycle,
|
||||
@@ -54,7 +54,9 @@ use rustfs_config::{ENV_SCANNER_CYCLE, ENV_SCANNER_SPEED, ENV_SCANNER_START_DELA
|
||||
use rustfs_data_usage::observed_data_usage_is_newer;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use sha2::{Digest as _, Sha256};
|
||||
use tokio::sync::{Notify, mpsc};
|
||||
#[cfg(test)]
|
||||
use tokio::sync::Notify;
|
||||
use tokio::sync::mpsc;
|
||||
use tokio::time::{Duration, Instant};
|
||||
use tokio_util::sync::CancellationToken;
|
||||
use tokio_util::task::AbortOnDropHandle;
|
||||
@@ -102,13 +104,6 @@ const CLEAN_IDLE_BACKOFF_FACTOR: u32 = 2;
|
||||
/// unavailable peer cannot drive a tight retry loop.
|
||||
const SCANNER_RETRY_BASE_INTERVAL: Duration = Duration::from_secs(5);
|
||||
const SCANNER_RETRY_MAX_INTERVAL: Duration = Duration::from_secs(30 * 60);
|
||||
/// A transient backend outage remains self-healing after the short retry
|
||||
/// budget is exhausted, but the probe is intentionally sparse until storage
|
||||
/// recovers or an operator reset wakes the scanner.
|
||||
const SCANNER_CYCLE_RECOVERY_PAUSED_INTERVAL: Duration = Duration::from_secs(5 * 60);
|
||||
/// Permanent recovery states still get a sparse status probe so a reset that
|
||||
/// races the wait registration cannot leave the scanner asleep forever.
|
||||
const SCANNER_CYCLE_RECOVERY_BLOCKED_PROBE_INTERVAL: Duration = Duration::from_secs(5 * 60);
|
||||
const SCANNER_LEADER_LOCK_POLL_INTERVAL: Duration = Duration::from_secs(1);
|
||||
#[cfg(not(test))]
|
||||
const SCANNER_LOCK_LOSS_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(30);
|
||||
@@ -130,12 +125,6 @@ type ScannerCycleStatePersistTestHook = (u64, Arc<Notify>);
|
||||
static SCANNER_CYCLE_STATE_PERSIST_TEST_HOOK: LazyLock<StdMutex<Option<ScannerCycleStatePersistTestHook>>> =
|
||||
LazyLock::new(|| StdMutex::new(None));
|
||||
|
||||
static SCANNER_CYCLE_RECOVERY_WAKE: LazyLock<Notify> = LazyLock::new(Notify::new);
|
||||
|
||||
pub(super) fn notify_scanner_cycle_recovery_wake() {
|
||||
SCANNER_CYCLE_RECOVERY_WAKE.notify_one();
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
struct ScannerCycleStatePersistTestHookGuard;
|
||||
|
||||
@@ -587,21 +576,19 @@ pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc<ECStore>) {
|
||||
tokio::time::sleep(sleep_time).await;
|
||||
}
|
||||
|
||||
let mut transient_backoff = ScannerRetryBackoff::default();
|
||||
let mut recovery_retry_count = 0_u32;
|
||||
loop {
|
||||
if ctx_clone.is_cancelled() {
|
||||
break;
|
||||
}
|
||||
|
||||
let run_result = run_data_scanner_with_maintenance_state(
|
||||
if let Err(e) = run_data_scanner_with_maintenance_state(
|
||||
ctx_clone.clone(),
|
||||
storeapi_clone.clone(),
|
||||
startup_features,
|
||||
startup_maintenance_generation,
|
||||
)
|
||||
.await;
|
||||
if let Err(e) = &run_result {
|
||||
.await
|
||||
{
|
||||
error!(
|
||||
target: "rustfs::scanner",
|
||||
event = EVENT_SCANNER_CYCLE_STATE,
|
||||
@@ -612,52 +599,11 @@ pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc<ECStore>) {
|
||||
"Scanner runtime iteration failed"
|
||||
);
|
||||
}
|
||||
let recovery_status = scanner_cycle_recovery_status();
|
||||
if recovery_status.retryable {
|
||||
recovery_retry_count = recovery_retry_count.saturating_add(1);
|
||||
let _ = record_scanner_cycle_recovery_retry(recovery_retry_count);
|
||||
} else {
|
||||
recovery_retry_count = 0;
|
||||
}
|
||||
|
||||
let recovery_status = scanner_cycle_recovery_status();
|
||||
if recovery_status.state == "paused" {
|
||||
transient_backoff.record_retryable_cycle(false);
|
||||
tokio::select! {
|
||||
_ = ctx_clone.cancelled() => break,
|
||||
_ = SCANNER_CYCLE_RECOVERY_WAKE.notified() => {},
|
||||
_ = tokio::time::sleep(SCANNER_CYCLE_RECOVERY_PAUSED_INTERVAL) => {},
|
||||
}
|
||||
recovery_retry_count = 0;
|
||||
continue;
|
||||
}
|
||||
if !recovery_status.retryable
|
||||
&& matches!(recovery_status.state.as_str(), "blocked" | "recovery-required" | "cleanup-pending")
|
||||
{
|
||||
transient_backoff.record_retryable_cycle(false);
|
||||
tokio::select! {
|
||||
_ = ctx_clone.cancelled() => break,
|
||||
_ = SCANNER_CYCLE_RECOVERY_WAKE.notified() => {},
|
||||
_ = tokio::time::sleep(SCANNER_CYCLE_RECOVERY_BLOCKED_PROBE_INTERVAL) => {},
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
let retry_delay = if recovery_status.retryable || run_result.is_err() {
|
||||
transient_backoff.record_retryable_cycle(true);
|
||||
transient_backoff
|
||||
.retry_interval(scanner_cycle_interval())
|
||||
.unwrap_or(SCANNER_RETRY_BASE_INTERVAL)
|
||||
} else {
|
||||
transient_backoff.record_retryable_cycle(false);
|
||||
randomized_cycle_delay()
|
||||
};
|
||||
// Backoff before retrying after lock contention or scanner-level failures.
|
||||
// Keep this cancellation-aware so shutdown is not delayed by backoff sleep.
|
||||
tokio::select! {
|
||||
_ = ctx_clone.cancelled() => break,
|
||||
_ = SCANNER_CYCLE_RECOVERY_WAKE.notified() => {},
|
||||
_ = tokio::time::sleep(retry_delay) => {}
|
||||
_ = tokio::time::sleep(randomized_cycle_delay()) => {}
|
||||
}
|
||||
}
|
||||
});
|
||||
@@ -1660,22 +1606,40 @@ async fn run_data_scanner_with_maintenance_state(
|
||||
observe_scanner_activity(&storeapi, distributed, &mut scanner_activity_seen).await;
|
||||
}
|
||||
|
||||
let (mut cycle_info, mut leader_epoch, mut cycle_revision) =
|
||||
match load_scanner_cycle_state_for_startup(storeapi.clone()).await {
|
||||
ScannerCycleStateStartup::Ready {
|
||||
cycle,
|
||||
leader_epoch,
|
||||
revision,
|
||||
} => (cycle, leader_epoch, revision),
|
||||
ScannerCycleStateStartup::Blocked => {
|
||||
global_metrics().set_cycle(None).await;
|
||||
return Ok(());
|
||||
}
|
||||
ScannerCycleStateStartup::Transient(err) => {
|
||||
global_metrics().set_cycle(None).await;
|
||||
return Err(err);
|
||||
}
|
||||
};
|
||||
let (buf, mut cycle_revision) = match read_config_with_revision(storeapi.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str()).await {
|
||||
Ok((buf, revision)) => (buf.unwrap_or_default(), revision),
|
||||
Err(err) => {
|
||||
error!(
|
||||
target: "rustfs::scanner",
|
||||
event = EVENT_SCANNER_PERSIST_STATE,
|
||||
component = LOG_COMPONENT_SCANNER,
|
||||
subsystem = LOG_SUBSYSTEM_RUNTIME,
|
||||
path = %&*DATA_USAGE_BLOOM_NAME_PATH,
|
||||
state = "revision_load_failed",
|
||||
error = %err,
|
||||
"Scanner cycle state revision load failed"
|
||||
);
|
||||
global_metrics().set_cycle(None).await;
|
||||
return Ok(());
|
||||
}
|
||||
};
|
||||
let (mut cycle_info, mut leader_epoch) = match decode_scanner_cycle_state_for_startup(&buf) {
|
||||
Ok(state) => state,
|
||||
Err(err) => {
|
||||
error!(
|
||||
target: "rustfs::scanner",
|
||||
event = EVENT_SCANNER_PERSIST_STATE,
|
||||
component = LOG_COMPONENT_SCANNER,
|
||||
subsystem = LOG_SUBSYSTEM_RUNTIME,
|
||||
path = %&*DATA_USAGE_BLOOM_NAME_PATH,
|
||||
state = "cycle_decode_failed",
|
||||
error = %err,
|
||||
"Scanner stopped because persisted cycle state is invalid"
|
||||
);
|
||||
global_metrics().set_cycle(None).await;
|
||||
return Ok(());
|
||||
}
|
||||
};
|
||||
let usage_floor = match persisted_usage_floor(storeapi.clone()).await {
|
||||
Ok(floor) => floor,
|
||||
Err(err) => {
|
||||
@@ -2255,12 +2219,7 @@ pub(crate) use activity::{
|
||||
pub(crate) use activity::{ScannerCycleOutcome, scanner_cycle_outcome_with_pending_maintenance};
|
||||
#[cfg(test)]
|
||||
pub(crate) use cycle_state::encode_scanner_cycle_fence_for_test;
|
||||
pub use cycle_state::{
|
||||
ScannerCycleRecoveryMarker, ScannerCycleRecoveryStatus, reset_scanner_cycle_recovery, scanner_cycle_recovery_status,
|
||||
};
|
||||
pub(crate) use cycle_state::{
|
||||
current_scanner_leader_epoch, decode_persisted_scanner_cycle_fence, load_scanner_cycle_state_for_startup,
|
||||
};
|
||||
pub(crate) use cycle_state::{current_scanner_leader_epoch, decode_persisted_scanner_cycle_fence};
|
||||
pub use heal_info::{BackgroundHealInfo, read_background_heal_info, save_background_heal_info};
|
||||
pub use usage_store::store_data_usage_in_backend;
|
||||
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -196,7 +196,7 @@ pub(super) async fn claim_scanner_leadership(
|
||||
if ctx.is_cancelled() {
|
||||
return false;
|
||||
}
|
||||
let Some(claimed_epoch) = persisted_epoch.checked_add(1).filter(|epoch| *epoch < u64::MAX) else {
|
||||
let Some(claimed_epoch) = persisted_epoch.checked_add(1) else {
|
||||
error!(
|
||||
target: "rustfs::scanner",
|
||||
event = EVENT_SCANNER_PERSIST_STATE,
|
||||
|
||||
@@ -15,12 +15,11 @@
|
||||
use super::*;
|
||||
use crate::EcstoreResult;
|
||||
use crate::{
|
||||
DATA_USAGE_BLOOM_RECOVERY_PATH, Endpoint, EndpointServerPools, Endpoints, InstanceContext, PoolEndpoints,
|
||||
ScannerGetObjectReader as GetObjectReader, ScannerObjectInfo as ObjectInfo, ScannerObjectOptions as ObjectOptions,
|
||||
ScannerPutObjReader as PutObjReader, init_bucket_metadata_sys_for_scanner_tests, init_ecstore_config_for_scanner_tests,
|
||||
init_local_disks_with_instance_ctx,
|
||||
Endpoint, EndpointServerPools, Endpoints, InstanceContext, PoolEndpoints, ScannerGetObjectReader as GetObjectReader,
|
||||
ScannerObjectInfo as ObjectInfo, ScannerObjectOptions as ObjectOptions, ScannerPutObjReader as PutObjReader,
|
||||
init_bucket_metadata_sys_for_scanner_tests, init_ecstore_config_for_scanner_tests, init_local_disks_with_instance_ctx,
|
||||
};
|
||||
use std::collections::{HashMap, HashSet};
|
||||
use std::collections::HashMap;
|
||||
use std::io::Cursor;
|
||||
use std::task::Poll;
|
||||
use temp_env::{with_var, with_var_unset};
|
||||
@@ -118,15 +117,6 @@ async fn scanner_cycle_lock_fence_bounds_uncooperative_shutdown() {
|
||||
assert!(cycle_ctx.is_cancelled());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn scanner_cycle_recovery_wake_survives_wait_registration_race() {
|
||||
notify_scanner_cycle_recovery_wake();
|
||||
|
||||
tokio::time::timeout(Duration::from_secs(1), SCANNER_CYCLE_RECOVERY_WAKE.notified())
|
||||
.await
|
||||
.expect("recovery wake should retain a permit until the waiter registers");
|
||||
}
|
||||
|
||||
struct ScannerDefaultSpeedGuard;
|
||||
|
||||
impl ScannerDefaultSpeedGuard {
|
||||
@@ -161,7 +151,6 @@ impl Drop for ScannerDefaultCycleGuard {
|
||||
struct MemoryConfigStore {
|
||||
objects: Mutex<HashMap<String, Vec<u8>>>,
|
||||
revisions: Mutex<HashMap<String, u64>>,
|
||||
non_regular_objects: Mutex<HashSet<String>>,
|
||||
fail_put_number: Mutex<HashMap<String, usize>>,
|
||||
object_not_found_put_number: Mutex<HashMap<String, usize>>,
|
||||
error_after_commit_put_number: Mutex<HashMap<String, usize>>,
|
||||
@@ -202,16 +191,12 @@ impl crate::storage_api::scanner_io::ObjectIO for MemoryConfigStore {
|
||||
.get(&key)
|
||||
.cloned()
|
||||
.ok_or(EcstoreError::FileNotFound)?;
|
||||
let data_len = i64::try_from(data.len()).expect("memory test object length should fit in i64");
|
||||
let revision = *self.revisions.lock().await.entry(key.clone()).or_insert(1);
|
||||
let is_dir = self.non_regular_objects.lock().await.contains(&key);
|
||||
let revision = *self.revisions.lock().await.entry(key).or_insert(1);
|
||||
|
||||
Ok(GetObjectReader {
|
||||
stream: Box::new(Cursor::new(data)),
|
||||
object_info: ObjectInfo {
|
||||
etag: Some(format!("memory-{revision}")),
|
||||
size: data_len,
|
||||
is_dir,
|
||||
..Default::default()
|
||||
},
|
||||
buffered_body: None,
|
||||
@@ -812,10 +797,6 @@ fn scanner_cycle_state_decodes_legacy_and_fenced_formats() {
|
||||
let (fenced_cycle, fenced_epoch) = decode_scanner_cycle_state(&fenced).expect("fenced cycle state should decode");
|
||||
assert_eq!(fenced_cycle.next, 13);
|
||||
assert_eq!(fenced_epoch, 7);
|
||||
|
||||
let mut trailing = fenced;
|
||||
trailing.push(0);
|
||||
assert!(decode_scanner_cycle_state(&trailing).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -842,840 +823,6 @@ fn scanner_startup_fails_closed_on_nonempty_corrupt_cycle_state() {
|
||||
assert!(encode_scanner_cycle_state(&exhausted, 7).is_err());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn corrupt_cycle_state_is_quarantined_once() {
|
||||
let store = Arc::new(MemoryConfigStore::default());
|
||||
let state_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str());
|
||||
store.objects.lock().await.insert(state_key.clone(), vec![1]);
|
||||
store.revisions.lock().await.insert(state_key.clone(), 7);
|
||||
|
||||
assert!(matches!(
|
||||
load_scanner_cycle_state_for_startup(store.clone()).await,
|
||||
ScannerCycleStateStartup::Blocked
|
||||
));
|
||||
let marker_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str());
|
||||
let marker_data = store
|
||||
.objects
|
||||
.lock()
|
||||
.await
|
||||
.get(&marker_key)
|
||||
.cloned()
|
||||
.expect("corrupt state must leave a durable recovery marker");
|
||||
let marker: ScannerCycleRecoveryMarker = serde_json::from_slice(&marker_data).expect("marker should be valid JSON");
|
||||
assert_eq!(marker.primary_revision, "memory-7");
|
||||
assert_eq!(marker.path, DATA_USAGE_BLOOM_NAME_PATH.as_str());
|
||||
assert_eq!(marker.quarantine_path, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str());
|
||||
assert_eq!(marker.classification, "corrupt");
|
||||
|
||||
// A second startup sees the matching marker before consuming the poison body.
|
||||
assert!(matches!(
|
||||
load_scanner_cycle_state_for_startup(store.clone()).await,
|
||||
ScannerCycleStateStartup::Blocked
|
||||
));
|
||||
|
||||
// Replacing the primary object advances its revision; the stale marker must
|
||||
// not quarantine the newer, valid state.
|
||||
let cycle = CurrentCycle {
|
||||
next: 9,
|
||||
..Default::default()
|
||||
};
|
||||
let encoded = encode_scanner_cycle_state(&cycle, 3).expect("valid state should encode");
|
||||
store.objects.lock().await.insert(state_key.clone(), encoded);
|
||||
store.revisions.lock().await.insert(state_key, 8);
|
||||
assert!(matches!(
|
||||
load_scanner_cycle_state_for_startup(store).await,
|
||||
ScannerCycleStateStartup::Ready {
|
||||
cycle: CurrentCycle { next: 9, .. },
|
||||
leader_epoch: 3,
|
||||
..
|
||||
}
|
||||
));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn empty_cycle_state_object_is_quarantined_as_corrupt() {
|
||||
let store = Arc::new(MemoryConfigStore::default());
|
||||
let state_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str());
|
||||
store.objects.lock().await.insert(state_key.clone(), Vec::new());
|
||||
store.revisions.lock().await.insert(state_key, 6);
|
||||
|
||||
assert!(matches!(
|
||||
load_scanner_cycle_state_for_startup(store).await,
|
||||
ScannerCycleStateStartup::Blocked
|
||||
));
|
||||
assert_eq!(scanner_cycle_recovery_status().classification.as_deref(), Some("corrupt"));
|
||||
assert!(
|
||||
scanner_cycle_recovery_status()
|
||||
.reason
|
||||
.as_deref()
|
||||
.is_some_and(|reason| reason.contains("empty"))
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn future_cycle_state_schema_is_recovery_required() {
|
||||
let store = Arc::new(MemoryConfigStore::default());
|
||||
let state_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str());
|
||||
let mut future = 17_u64.to_le_bytes().to_vec();
|
||||
future.extend_from_slice(b"RSCYC999");
|
||||
future.extend_from_slice(&4_u64.to_le_bytes());
|
||||
future.extend_from_slice(&[0x90]);
|
||||
store.objects.lock().await.insert(state_key.clone(), future);
|
||||
store.revisions.lock().await.insert(state_key, 13);
|
||||
|
||||
assert!(matches!(
|
||||
load_scanner_cycle_state_for_startup(store).await,
|
||||
ScannerCycleStateStartup::Blocked
|
||||
));
|
||||
assert_eq!(scanner_cycle_recovery_status().classification.as_deref(), Some("future_schema"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn concurrent_leaders_cannot_quarantine_newer_cycle_state() {
|
||||
let store = Arc::new(MemoryConfigStore::default());
|
||||
let state_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str());
|
||||
store.objects.lock().await.insert(state_key.clone(), vec![1]);
|
||||
store.revisions.lock().await.insert(state_key, 4);
|
||||
|
||||
let (first, second) = tokio::join!(
|
||||
load_scanner_cycle_state_for_startup(store.clone()),
|
||||
load_scanner_cycle_state_for_startup(store.clone()),
|
||||
);
|
||||
assert!(matches!(first, ScannerCycleStateStartup::Blocked));
|
||||
assert!(matches!(second, ScannerCycleStateStartup::Blocked));
|
||||
|
||||
let marker_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str());
|
||||
let marker_data = store
|
||||
.objects
|
||||
.lock()
|
||||
.await
|
||||
.get(&marker_key)
|
||||
.cloned()
|
||||
.expect("one contender must publish the recovery marker");
|
||||
let marker: ScannerCycleRecoveryMarker = serde_json::from_slice(&marker_data).expect("marker should decode");
|
||||
assert_eq!(marker.primary_revision, "memory-4");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn cleanup_pending_marker_blocks_a_rewritten_primary_after_restart() {
|
||||
let store = Arc::new(MemoryConfigStore::default());
|
||||
let state_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str());
|
||||
let marker_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str());
|
||||
let encoded = encode_scanner_cycle_state(
|
||||
&CurrentCycle {
|
||||
next: 12,
|
||||
..Default::default()
|
||||
},
|
||||
8,
|
||||
)
|
||||
.expect("valid state should encode");
|
||||
store.objects.lock().await.insert(state_key.clone(), encoded);
|
||||
store.revisions.lock().await.insert(state_key, 22);
|
||||
let marker = ScannerCycleRecoveryMarker {
|
||||
schema_version: 1,
|
||||
primary_revision: "memory-21".to_string(),
|
||||
generation: 11,
|
||||
leader_epoch: 7,
|
||||
classification: "corrupt".to_string(),
|
||||
first_detected_at_unix_secs: 1,
|
||||
last_attempt_at_unix_secs: 2,
|
||||
retry_count: 1,
|
||||
reason: "reset in progress".to_string(),
|
||||
path: DATA_USAGE_BLOOM_NAME_PATH.clone(),
|
||||
quarantine_path: DATA_USAGE_BLOOM_RECOVERY_PATH.clone(),
|
||||
state: "cleanup-pending".to_string(),
|
||||
};
|
||||
store
|
||||
.objects
|
||||
.lock()
|
||||
.await
|
||||
.insert(marker_key.clone(), serde_json::to_vec(&marker).expect("marker should encode"));
|
||||
store.revisions.lock().await.insert(marker_key, 3);
|
||||
|
||||
assert!(matches!(
|
||||
load_scanner_cycle_state_for_startup(store).await,
|
||||
ScannerCycleStateStartup::Blocked
|
||||
));
|
||||
assert_eq!(scanner_cycle_recovery_status().state, "cleanup-pending");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn full_rescan_reset_accepts_unknown_marker_fields_without_trusting_cursor() {
|
||||
let marker = br#"{
|
||||
"schema_version": 99,
|
||||
"primary_revision": "memory-7",
|
||||
"generation": 9000,
|
||||
"leader_epoch": 9000,
|
||||
"classification": "new-future-classification",
|
||||
"first_detected_at_unix_secs": 1,
|
||||
"last_attempt_at_unix_secs": 2,
|
||||
"retry_count": 9,
|
||||
"reason": "future marker",
|
||||
"path": "buckets/.bloomcycle.bin",
|
||||
"quarantine_path": "buckets/.bloomcycle.bin.recovery-required.json",
|
||||
"future_field": {"cursor": "untrusted"}
|
||||
}"#;
|
||||
let decoded =
|
||||
super::cycle_state::decode_recovery_marker_for_reset(marker, &DataUsageCacheRevision::Etag("memory-3".to_string()))
|
||||
.expect("full-rescan compatibility decoder should accept additive fields");
|
||||
assert_eq!(decoded.primary_revision, "memory-7");
|
||||
assert_eq!(decoded.classification, "future_schema");
|
||||
assert_eq!(decoded.generation, 0);
|
||||
assert_eq!(decoded.leader_epoch, 0);
|
||||
assert_eq!(decoded.state, "blocked");
|
||||
|
||||
let malformed =
|
||||
super::cycle_state::decode_recovery_marker_for_reset(b"{not-json", &DataUsageCacheRevision::Etag("memory-4".to_string()))
|
||||
.expect("a full-rescan reset must recover even when the marker is malformed");
|
||||
assert!(malformed.primary_revision.is_empty());
|
||||
assert_eq!(malformed.classification, "future_schema");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn full_rescan_reset_rebuilds_after_malformed_marker_without_trusting_cursor() {
|
||||
let (_temp_dir, store) = setup_scanner_cycle_store().await;
|
||||
save_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str(), vec![0xff, 0x00, 0x01])
|
||||
.await
|
||||
.expect("corrupt cycle state should be persisted");
|
||||
save_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), br#"{not-json"#.to_vec())
|
||||
.await
|
||||
.expect("malformed marker should be persisted");
|
||||
|
||||
reset_scanner_cycle_recovery(CancellationToken::new(), store.clone())
|
||||
.await
|
||||
.expect("full-rescan reset should recover malformed marker");
|
||||
|
||||
let state = read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
|
||||
.await
|
||||
.expect("rebuilt cycle state should remain durable");
|
||||
let (cycle, leader_epoch) = decode_scanner_cycle_state(&state).expect("rebuilt cycle state should decode");
|
||||
assert_eq!(cycle.next, 0, "reset must use the verified usage floor, not marker cursor");
|
||||
assert_eq!(leader_epoch, 1);
|
||||
assert!(matches!(
|
||||
read_config(store, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()).await,
|
||||
Err(EcstoreError::ConfigNotFound)
|
||||
));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn full_rescan_reset_ignores_epoch_from_malformed_future_primary() {
|
||||
let (_temp_dir, store) = setup_scanner_cycle_store().await;
|
||||
let mut future_primary = vec![0; 24];
|
||||
future_primary[8..16].copy_from_slice(b"RSCY9999");
|
||||
future_primary[16..24].copy_from_slice(&u64::MAX.to_le_bytes());
|
||||
save_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str(), future_primary)
|
||||
.await
|
||||
.expect("future cycle state should be persisted");
|
||||
save_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), br#"{not-json"#.to_vec())
|
||||
.await
|
||||
.expect("malformed marker should be persisted");
|
||||
|
||||
reset_scanner_cycle_recovery(CancellationToken::new(), store.clone())
|
||||
.await
|
||||
.expect("full-rescan reset should recover malformed future state");
|
||||
|
||||
let state = read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
|
||||
.await
|
||||
.expect("rebuilt cycle state should remain durable");
|
||||
let (_, leader_epoch) = decode_scanner_cycle_state(&state).expect("rebuilt cycle state should decode");
|
||||
assert_eq!(leader_epoch, 1, "invalid persisted bytes must not raise the recovery epoch");
|
||||
assert!(matches!(
|
||||
read_config(store, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()).await,
|
||||
Err(EcstoreError::ConfigNotFound)
|
||||
));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn ecstore_exact_recovery_marker_delete_honors_etag() {
|
||||
let (_temp_dir, store) = setup_scanner_cycle_store().await;
|
||||
save_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), b"marker-v1".to_vec())
|
||||
.await
|
||||
.expect("initial recovery marker should be persisted");
|
||||
let (_, stale_revision) = read_config_with_revision(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str())
|
||||
.await
|
||||
.expect("initial marker revision should load");
|
||||
save_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), b"marker-v2".to_vec())
|
||||
.await
|
||||
.expect("replacement recovery marker should be persisted");
|
||||
|
||||
let delete_result = store
|
||||
.delete_config_object(
|
||||
RUSTFS_META_BUCKET,
|
||||
DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(),
|
||||
ObjectOptions {
|
||||
http_preconditions: Some(stale_revision.preconditions()),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await;
|
||||
assert!(matches!(delete_result, Err(EcstoreError::PreconditionFailed)));
|
||||
assert_eq!(
|
||||
read_config(store, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str())
|
||||
.await
|
||||
.expect("replacement marker should remain durable"),
|
||||
b"marker-v2"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn full_rescan_reset_rejects_corrupt_primary_under_stale_blocked_marker() {
|
||||
let (_temp_dir, store) = setup_scanner_cycle_store().await;
|
||||
let corrupt_primary = vec![0xff, 0x00, 0x01];
|
||||
save_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str(), corrupt_primary.clone())
|
||||
.await
|
||||
.expect("corrupt cycle state should be persisted");
|
||||
let (_, primary_revision) = read_config_with_revision(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
|
||||
.await
|
||||
.expect("primary revision should load");
|
||||
let marker = ScannerCycleRecoveryMarker {
|
||||
schema_version: 1,
|
||||
primary_revision: "memory-stale".to_string(),
|
||||
generation: 1,
|
||||
leader_epoch: 1,
|
||||
classification: "corrupt".to_string(),
|
||||
first_detected_at_unix_secs: 1,
|
||||
last_attempt_at_unix_secs: 2,
|
||||
retry_count: 1,
|
||||
reason: "blocked primary changed".to_string(),
|
||||
path: DATA_USAGE_BLOOM_NAME_PATH.clone(),
|
||||
quarantine_path: DATA_USAGE_BLOOM_RECOVERY_PATH.clone(),
|
||||
state: "blocked".to_string(),
|
||||
};
|
||||
let marker_data = serde_json::to_vec(&marker).expect("blocked marker should encode");
|
||||
save_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), marker_data.clone())
|
||||
.await
|
||||
.expect("blocked marker should be persisted");
|
||||
|
||||
assert!(
|
||||
reset_scanner_cycle_recovery(CancellationToken::new(), store.clone())
|
||||
.await
|
||||
.is_err(),
|
||||
"a strict marker must fail closed when its primary revision changed"
|
||||
);
|
||||
assert_eq!(
|
||||
read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
|
||||
.await
|
||||
.expect("primary should remain readable"),
|
||||
corrupt_primary
|
||||
);
|
||||
assert_eq!(
|
||||
read_config(store, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str())
|
||||
.await
|
||||
.expect("blocked marker should remain durable"),
|
||||
marker_data
|
||||
);
|
||||
assert!(!matches!(primary_revision, DataUsageCacheRevision::Missing));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn full_rescan_reset_preserves_valid_primary_when_marker_is_malformed() {
|
||||
let (_temp_dir, store) = setup_scanner_cycle_store().await;
|
||||
let primary = CurrentCycle {
|
||||
next: 42,
|
||||
..Default::default()
|
||||
};
|
||||
let old_primary_data = encode_scanner_cycle_state(&primary, 7).expect("valid cycle state should encode");
|
||||
save_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str(), old_primary_data.clone())
|
||||
.await
|
||||
.expect("valid cycle state should be persisted");
|
||||
let (_, old_primary_revision) = read_config_with_revision(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
|
||||
.await
|
||||
.expect("primary state revision should load");
|
||||
let old_usage = DataUsageInfo {
|
||||
scanner_epoch: Some(7),
|
||||
scanner_cycle: Some(41),
|
||||
..Default::default()
|
||||
};
|
||||
let old_usage_data = serde_json::to_vec(&old_usage).expect("usage snapshot should encode");
|
||||
save_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str(), old_usage_data.clone())
|
||||
.await
|
||||
.expect("usage snapshot should be persisted");
|
||||
let (_, old_usage_revision) = read_config_with_revision(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
|
||||
.await
|
||||
.expect("usage snapshot revision should load");
|
||||
save_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), b"{not-json".to_vec())
|
||||
.await
|
||||
.expect("malformed marker should be persisted");
|
||||
|
||||
reset_scanner_cycle_recovery(CancellationToken::new(), store.clone())
|
||||
.await
|
||||
.expect("reset should clear a stale malformed marker");
|
||||
|
||||
let state = read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
|
||||
.await
|
||||
.expect("valid primary should remain durable");
|
||||
let (cycle, leader_epoch) = decode_scanner_cycle_state(&state).expect("primary cycle state should decode");
|
||||
assert_eq!(cycle.next, 42, "reset must not regress an independently fenced primary");
|
||||
assert_eq!(leader_epoch, 8, "reset must advance the preserved primary epoch");
|
||||
let stale_primary_save = save_config_with_preconditions(
|
||||
store.clone(),
|
||||
DATA_USAGE_BLOOM_NAME_PATH.as_str(),
|
||||
old_primary_data,
|
||||
old_primary_revision.preconditions(),
|
||||
)
|
||||
.await;
|
||||
assert!(matches!(stale_primary_save, Err(EcstoreError::PreconditionFailed)));
|
||||
let usage = read_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
|
||||
.await
|
||||
.expect("usage epoch fence should remain durable");
|
||||
assert_eq!(
|
||||
serde_json::from_slice::<DataUsageInfo>(&usage)
|
||||
.expect("fenced usage should decode")
|
||||
.scanner_epoch,
|
||||
Some(8)
|
||||
);
|
||||
let stale_save = save_config_with_preconditions(
|
||||
store.clone(),
|
||||
DATA_USAGE_OBJ_NAME_PATH.as_str(),
|
||||
old_usage_data,
|
||||
old_usage_revision.preconditions(),
|
||||
)
|
||||
.await;
|
||||
assert!(matches!(stale_save, Err(EcstoreError::PreconditionFailed)));
|
||||
assert!(matches!(
|
||||
read_config(store, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()).await,
|
||||
Err(EcstoreError::ConfigNotFound)
|
||||
));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn full_rescan_reset_resumes_cleanup_pending_preserved_primary() {
|
||||
let (_temp_dir, store) = setup_scanner_cycle_store().await;
|
||||
let completed_at = Utc::now();
|
||||
let primary = CurrentCycle {
|
||||
current: 3,
|
||||
next: 42,
|
||||
cycle_completed: vec![completed_at],
|
||||
started: completed_at,
|
||||
};
|
||||
save_config(
|
||||
store.clone(),
|
||||
DATA_USAGE_BLOOM_NAME_PATH.as_str(),
|
||||
encode_scanner_cycle_state(&primary, 7).expect("valid cycle state should encode"),
|
||||
)
|
||||
.await
|
||||
.expect("valid cycle state should be persisted");
|
||||
let usage = DataUsageInfo {
|
||||
scanner_epoch: Some(7),
|
||||
scanner_cycle: Some(41),
|
||||
..Default::default()
|
||||
};
|
||||
save_config(
|
||||
store.clone(),
|
||||
DATA_USAGE_OBJ_NAME_PATH.as_str(),
|
||||
serde_json::to_vec(&usage).expect("usage snapshot should encode"),
|
||||
)
|
||||
.await
|
||||
.expect("usage snapshot should be persisted");
|
||||
let marker = ScannerCycleRecoveryMarker {
|
||||
schema_version: 1,
|
||||
primary_revision: "memory-old".to_string(),
|
||||
generation: 41,
|
||||
leader_epoch: 7,
|
||||
classification: "corrupt".to_string(),
|
||||
first_detected_at_unix_secs: 1,
|
||||
last_attempt_at_unix_secs: 2,
|
||||
retry_count: 1,
|
||||
reason: "reset in progress".to_string(),
|
||||
path: DATA_USAGE_BLOOM_NAME_PATH.clone(),
|
||||
quarantine_path: DATA_USAGE_BLOOM_RECOVERY_PATH.clone(),
|
||||
state: "cleanup-pending".to_string(),
|
||||
};
|
||||
save_config(
|
||||
store.clone(),
|
||||
DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(),
|
||||
serde_json::to_vec(&marker).expect("marker should encode"),
|
||||
)
|
||||
.await
|
||||
.expect("cleanup marker should be persisted");
|
||||
|
||||
reset_scanner_cycle_recovery(CancellationToken::new(), store.clone())
|
||||
.await
|
||||
.expect("reset should resume a cleanup-pending preserved primary");
|
||||
|
||||
let state = read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
|
||||
.await
|
||||
.expect("preserved cycle state should remain durable");
|
||||
let (cycle, leader_epoch) = decode_scanner_cycle_state(&state).expect("cycle state should decode");
|
||||
assert_eq!(cycle.current, 3, "cleanup retry must preserve the in-progress cursor");
|
||||
assert_eq!(cycle.next, 42);
|
||||
assert_eq!(cycle.cycle_completed, vec![completed_at]);
|
||||
assert_eq!(cycle.started, completed_at);
|
||||
assert_eq!(leader_epoch, 8);
|
||||
let usage = read_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
|
||||
.await
|
||||
.expect("usage epoch fence should remain durable");
|
||||
assert_eq!(
|
||||
serde_json::from_slice::<DataUsageInfo>(&usage)
|
||||
.expect("usage should decode")
|
||||
.scanner_epoch,
|
||||
Some(8)
|
||||
);
|
||||
assert!(matches!(
|
||||
read_config(store, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()).await,
|
||||
Err(EcstoreError::ConfigNotFound)
|
||||
));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn full_rescan_reset_rebuilds_oversized_regular_primary_with_malformed_marker() {
|
||||
let (_temp_dir, store) = setup_scanner_cycle_store().await;
|
||||
save_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str(), vec![0; 1024 * 1024 + 1])
|
||||
.await
|
||||
.expect("oversized cycle state should be persisted");
|
||||
save_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), b"{not-json".to_vec())
|
||||
.await
|
||||
.expect("malformed marker should be persisted");
|
||||
|
||||
reset_scanner_cycle_recovery(CancellationToken::new(), store.clone())
|
||||
.await
|
||||
.expect("explicit full-rescan reset should replace an oversized regular primary");
|
||||
|
||||
let state = read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
|
||||
.await
|
||||
.expect("rebuilt cycle state should remain durable");
|
||||
let (cycle, leader_epoch) = decode_scanner_cycle_state(&state).expect("rebuilt cycle state should decode");
|
||||
assert_eq!(cycle.next, 0);
|
||||
assert_eq!(leader_epoch, 1);
|
||||
assert!(matches!(
|
||||
read_config(store, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()).await,
|
||||
Err(EcstoreError::ConfigNotFound)
|
||||
));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn full_rescan_reset_rebuilds_oversized_primary_after_cleanup_marker() {
|
||||
let (_temp_dir, store) = setup_scanner_cycle_store().await;
|
||||
save_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str(), vec![0; 1024 * 1024 + 1])
|
||||
.await
|
||||
.expect("oversized cycle state should be persisted");
|
||||
let (_, primary_revision) = read_config_with_revision(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
|
||||
.await
|
||||
.expect("primary revision should load");
|
||||
let marker = ScannerCycleRecoveryMarker {
|
||||
schema_version: 1,
|
||||
primary_revision: match primary_revision {
|
||||
DataUsageCacheRevision::Etag(etag) => etag,
|
||||
DataUsageCacheRevision::Missing => panic!("primary revision should be present"),
|
||||
},
|
||||
generation: 1,
|
||||
leader_epoch: 1,
|
||||
classification: "corrupt".to_string(),
|
||||
first_detected_at_unix_secs: 1,
|
||||
last_attempt_at_unix_secs: 2,
|
||||
retry_count: 1,
|
||||
reason: "reset in progress".to_string(),
|
||||
path: DATA_USAGE_BLOOM_NAME_PATH.clone(),
|
||||
quarantine_path: DATA_USAGE_BLOOM_RECOVERY_PATH.clone(),
|
||||
state: "cleanup-pending".to_string(),
|
||||
};
|
||||
save_config(
|
||||
store.clone(),
|
||||
DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(),
|
||||
serde_json::to_vec(&marker).expect("cleanup marker should encode"),
|
||||
)
|
||||
.await
|
||||
.expect("cleanup marker should be persisted");
|
||||
|
||||
reset_scanner_cycle_recovery(CancellationToken::new(), store.clone())
|
||||
.await
|
||||
.expect("cleanup retry should rebuild an oversized primary");
|
||||
|
||||
let state = read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
|
||||
.await
|
||||
.expect("rebuilt cycle state should remain durable");
|
||||
let (cycle, leader_epoch) = decode_scanner_cycle_state(&state).expect("rebuilt cycle state should decode");
|
||||
assert_eq!(cycle.next, 0);
|
||||
assert_eq!(leader_epoch, 1);
|
||||
assert!(matches!(
|
||||
read_config(store, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()).await,
|
||||
Err(EcstoreError::ConfigNotFound)
|
||||
));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn full_rescan_reset_rebuilds_with_oversized_marker() {
|
||||
let (_temp_dir, store) = setup_scanner_cycle_store().await;
|
||||
save_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str(), vec![0xff, 0x00, 0x01])
|
||||
.await
|
||||
.expect("corrupt cycle state should be persisted");
|
||||
save_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), vec![b'x'; 64 * 1024 + 1])
|
||||
.await
|
||||
.expect("oversized recovery marker should be persisted");
|
||||
|
||||
reset_scanner_cycle_recovery(CancellationToken::new(), store.clone())
|
||||
.await
|
||||
.expect("full-rescan reset should recover an oversized marker");
|
||||
|
||||
let state = read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
|
||||
.await
|
||||
.expect("rebuilt cycle state should remain durable");
|
||||
let (_, leader_epoch) = decode_scanner_cycle_state(&state).expect("rebuilt cycle state should decode");
|
||||
assert_eq!(leader_epoch, 1);
|
||||
assert!(matches!(
|
||||
read_config(store, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()).await,
|
||||
Err(EcstoreError::ConfigNotFound)
|
||||
));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn full_rescan_reset_rebuilds_with_empty_marker() {
|
||||
let (_temp_dir, store) = setup_scanner_cycle_store().await;
|
||||
save_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str(), vec![0xff, 0x00, 0x01])
|
||||
.await
|
||||
.expect("corrupt cycle state should be persisted");
|
||||
save_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), Vec::new())
|
||||
.await
|
||||
.expect("empty recovery marker should be persisted");
|
||||
|
||||
reset_scanner_cycle_recovery(CancellationToken::new(), store.clone())
|
||||
.await
|
||||
.expect("full-rescan reset should recover an empty marker");
|
||||
|
||||
let state = read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
|
||||
.await
|
||||
.expect("rebuilt cycle state should remain durable");
|
||||
let (_, leader_epoch) = decode_scanner_cycle_state(&state).expect("rebuilt cycle state should decode");
|
||||
assert_eq!(leader_epoch, 1);
|
||||
assert!(matches!(
|
||||
read_config(store, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()).await,
|
||||
Err(EcstoreError::ConfigNotFound)
|
||||
));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn full_rescan_reset_keeps_cleanup_marker_when_preserved_epoch_is_exhausted() {
|
||||
let (_temp_dir, store) = setup_scanner_cycle_store().await;
|
||||
let primary = CurrentCycle {
|
||||
next: 42,
|
||||
..Default::default()
|
||||
};
|
||||
save_config(
|
||||
store.clone(),
|
||||
DATA_USAGE_BLOOM_NAME_PATH.as_str(),
|
||||
encode_scanner_cycle_state(&primary, u64::MAX).expect("valid cycle state should encode"),
|
||||
)
|
||||
.await
|
||||
.expect("valid cycle state should be persisted");
|
||||
save_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), b"{not-json".to_vec())
|
||||
.await
|
||||
.expect("malformed marker should be persisted");
|
||||
|
||||
assert!(
|
||||
reset_scanner_cycle_recovery(CancellationToken::new(), store.clone())
|
||||
.await
|
||||
.is_err()
|
||||
);
|
||||
|
||||
let marker = read_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str())
|
||||
.await
|
||||
.expect("cleanup marker should remain durable");
|
||||
assert_eq!(
|
||||
serde_json::from_slice::<ScannerCycleRecoveryMarker>(&marker)
|
||||
.expect("cleanup marker should decode")
|
||||
.state,
|
||||
"cleanup-pending"
|
||||
);
|
||||
assert!(matches!(
|
||||
load_scanner_cycle_state_for_startup(store).await,
|
||||
ScannerCycleStateStartup::Blocked
|
||||
));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn full_rescan_reset_rejects_preserved_epoch_that_would_be_terminal() {
|
||||
let (_temp_dir, store) = setup_scanner_cycle_store().await;
|
||||
let primary = CurrentCycle {
|
||||
next: 42,
|
||||
..Default::default()
|
||||
};
|
||||
save_config(
|
||||
store.clone(),
|
||||
DATA_USAGE_BLOOM_NAME_PATH.as_str(),
|
||||
encode_scanner_cycle_state(&primary, u64::MAX - 1).expect("valid cycle state should encode"),
|
||||
)
|
||||
.await
|
||||
.expect("valid cycle state should be persisted");
|
||||
save_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), b"{not-json".to_vec())
|
||||
.await
|
||||
.expect("malformed marker should be persisted");
|
||||
|
||||
assert!(
|
||||
reset_scanner_cycle_recovery(CancellationToken::new(), store.clone())
|
||||
.await
|
||||
.is_err(),
|
||||
"reset must not persist the terminal leader epoch"
|
||||
);
|
||||
|
||||
let marker = read_config(store, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str())
|
||||
.await
|
||||
.expect("cleanup marker should remain durable");
|
||||
assert_eq!(
|
||||
serde_json::from_slice::<ScannerCycleRecoveryMarker>(&marker)
|
||||
.expect("cleanup marker should decode")
|
||||
.state,
|
||||
"cleanup-pending"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn full_rescan_reset_rejects_usage_floor_that_would_be_terminal() {
|
||||
let (_temp_dir, store) = setup_scanner_cycle_store().await;
|
||||
save_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str(), vec![0xff, 0x00, 0x01])
|
||||
.await
|
||||
.expect("corrupt cycle state should be persisted");
|
||||
save_config(
|
||||
store.clone(),
|
||||
DATA_USAGE_OBJ_NAME_PATH.as_str(),
|
||||
serde_json::to_vec(&DataUsageInfo {
|
||||
scanner_epoch: Some(u64::MAX - 1),
|
||||
..Default::default()
|
||||
})
|
||||
.expect("usage floor should encode"),
|
||||
)
|
||||
.await
|
||||
.expect("usage floor should be persisted");
|
||||
save_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), b"{not-json".to_vec())
|
||||
.await
|
||||
.expect("malformed marker should be persisted");
|
||||
|
||||
assert!(
|
||||
reset_scanner_cycle_recovery(CancellationToken::new(), store.clone())
|
||||
.await
|
||||
.is_err(),
|
||||
"reset must not persist the terminal leader epoch"
|
||||
);
|
||||
assert_eq!(
|
||||
read_config(store, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str())
|
||||
.await
|
||||
.expect("recovery marker should remain durable"),
|
||||
b"{not-json"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn full_rescan_reset_rebuilds_empty_primary_with_malformed_marker() {
|
||||
let (_temp_dir, store) = setup_scanner_cycle_store().await;
|
||||
save_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str(), Vec::new())
|
||||
.await
|
||||
.expect("empty cycle state should be persisted");
|
||||
save_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(), b"{not-json".to_vec())
|
||||
.await
|
||||
.expect("malformed marker should be persisted");
|
||||
|
||||
reset_scanner_cycle_recovery(CancellationToken::new(), store.clone())
|
||||
.await
|
||||
.expect("explicit full-rescan reset should replace an empty primary");
|
||||
|
||||
let state = read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
|
||||
.await
|
||||
.expect("rebuilt cycle state should remain durable");
|
||||
let (cycle, leader_epoch) = decode_scanner_cycle_state(&state).expect("rebuilt cycle state should decode");
|
||||
assert_eq!(cycle.next, 0);
|
||||
assert_eq!(leader_epoch, 1);
|
||||
assert!(matches!(
|
||||
read_config(store, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()).await,
|
||||
Err(EcstoreError::ConfigNotFound)
|
||||
));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn full_rescan_reset_rebuilds_when_primary_cycle_state_is_missing() {
|
||||
let (_temp_dir, store) = setup_scanner_cycle_store().await;
|
||||
let marker = ScannerCycleRecoveryMarker {
|
||||
schema_version: 1,
|
||||
primary_revision: "memory-missing".to_string(),
|
||||
generation: u64::MAX,
|
||||
leader_epoch: u64::MAX,
|
||||
classification: "corrupt".to_string(),
|
||||
first_detected_at_unix_secs: 1,
|
||||
last_attempt_at_unix_secs: 2,
|
||||
retry_count: 0,
|
||||
reason: "missing primary".to_string(),
|
||||
path: DATA_USAGE_BLOOM_NAME_PATH.clone(),
|
||||
quarantine_path: DATA_USAGE_BLOOM_RECOVERY_PATH.clone(),
|
||||
state: "blocked".to_string(),
|
||||
};
|
||||
save_config(
|
||||
store.clone(),
|
||||
DATA_USAGE_BLOOM_RECOVERY_PATH.as_str(),
|
||||
serde_json::to_vec(&marker).expect("marker should encode"),
|
||||
)
|
||||
.await
|
||||
.expect("marker should be persisted");
|
||||
|
||||
reset_scanner_cycle_recovery(CancellationToken::new(), store.clone())
|
||||
.await
|
||||
.expect("full-rescan reset should recreate missing primary");
|
||||
let state = read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
|
||||
.await
|
||||
.expect("missing primary should be rebuilt");
|
||||
let (cycle, leader_epoch) = decode_scanner_cycle_state(&state).expect("rebuilt cycle state should decode");
|
||||
assert_eq!(cycle.next, 0);
|
||||
assert_eq!(leader_epoch, 1);
|
||||
assert!(matches!(
|
||||
read_config(store, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()).await,
|
||||
Err(EcstoreError::ConfigNotFound)
|
||||
));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn corrupt_cycle_state_rename_or_marker_failure_stays_recovery_required() {
|
||||
let store = Arc::new(MemoryConfigStore::default());
|
||||
let state_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str());
|
||||
let marker_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str());
|
||||
store.objects.lock().await.insert(state_key.clone(), vec![1]);
|
||||
store.revisions.lock().await.insert(state_key, 9);
|
||||
store.fail_put_number.lock().await.insert(marker_key, 1);
|
||||
|
||||
assert!(matches!(
|
||||
load_scanner_cycle_state_for_startup(store.clone()).await,
|
||||
ScannerCycleStateStartup::Transient(_)
|
||||
));
|
||||
let status = scanner_cycle_recovery_status();
|
||||
assert_eq!(status.state, "recovery-required");
|
||||
assert!(status.retryable);
|
||||
assert!(
|
||||
store
|
||||
.objects
|
||||
.lock()
|
||||
.await
|
||||
.contains_key(&memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str()))
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn oversized_or_symlinked_cycle_state_is_rejected() {
|
||||
let store = Arc::new(MemoryConfigStore::default());
|
||||
let key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str());
|
||||
store.objects.lock().await.insert(key.clone(), vec![0; 1024 * 1024 + 1]);
|
||||
store.revisions.lock().await.insert(key.clone(), 11);
|
||||
|
||||
assert!(matches!(
|
||||
load_scanner_cycle_state_for_startup(store.clone()).await,
|
||||
ScannerCycleStateStartup::Blocked
|
||||
));
|
||||
assert_eq!(scanner_cycle_recovery_status().classification.as_deref(), Some("corrupt"));
|
||||
assert!(
|
||||
scanner_cycle_recovery_status()
|
||||
.reason
|
||||
.as_deref()
|
||||
.is_some_and(|reason| reason.contains("oversized"))
|
||||
);
|
||||
|
||||
let marker_key = memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_RECOVERY_PATH.as_str());
|
||||
store.objects.lock().await.remove(&marker_key);
|
||||
store.objects.lock().await.insert(key.clone(), vec![1]);
|
||||
store.revisions.lock().await.insert(key.clone(), 12);
|
||||
store.non_regular_objects.lock().await.insert(key);
|
||||
// The object contract exposes a non-regular object as `is_dir`; local
|
||||
// backends reject symlink/reparse entries before they become an object.
|
||||
assert!(matches!(
|
||||
load_scanner_cycle_state_for_startup(store).await,
|
||||
ScannerCycleStateStartup::Blocked
|
||||
));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn scanner_startup_uses_primary_and_backup_usage_floor() {
|
||||
let store = Arc::new(MemoryConfigStore::default());
|
||||
@@ -1708,31 +855,6 @@ async fn scanner_startup_uses_primary_and_backup_usage_floor() {
|
||||
assert_eq!(epoch, 11);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn scanner_usage_floor_ignores_older_backup_after_primary_epoch_fence() {
|
||||
let store = Arc::new(MemoryConfigStore::default());
|
||||
let backup_path = format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str());
|
||||
for (path, epoch, cycle) in [(DATA_USAGE_OBJ_NAME_PATH.as_str(), 8, 100), (backup_path.as_str(), 7, 10_000)] {
|
||||
store.objects.lock().await.insert(
|
||||
memory_config_key(RUSTFS_META_BUCKET, path),
|
||||
serde_json::to_vec(&DataUsageInfo {
|
||||
scanner_epoch: Some(epoch),
|
||||
scanner_cycle: Some(cycle),
|
||||
..Default::default()
|
||||
})
|
||||
.expect("usage snapshot should encode"),
|
||||
);
|
||||
}
|
||||
|
||||
assert_eq!(
|
||||
persisted_usage_floor(store).await.expect("usage floor should load"),
|
||||
PersistedUsageFloor {
|
||||
next_cycle: 101,
|
||||
leader_epoch: 8,
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scanner_startup_treats_incomplete_usage_snapshot_as_cold() {
|
||||
let mut legacy = complete_usage_with_bucket_count(Some(std::time::SystemTime::now()), 1);
|
||||
@@ -1865,15 +987,6 @@ async fn scanner_usage_floor_fails_closed_on_corrupt_or_exhausted_usage_state()
|
||||
|
||||
assert!(persisted_usage_floor(store.clone()).await.is_err());
|
||||
|
||||
store.objects.lock().await.insert(
|
||||
memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str()),
|
||||
br#"{}"#.to_vec(),
|
||||
);
|
||||
assert!(
|
||||
persisted_usage_floor(store.clone()).await.is_err(),
|
||||
"a structurally incomplete usage snapshot must not be treated as an empty floor"
|
||||
);
|
||||
|
||||
store.objects.lock().await.insert(
|
||||
memory_config_key(RUSTFS_META_BUCKET, DATA_USAGE_OBJ_NAME_PATH.as_str()),
|
||||
serde_json::to_vec(&DataUsageInfo {
|
||||
@@ -2131,22 +1244,6 @@ async fn test_leadership_claim_preserves_usage_epoch_floor_across_old_epoch_conf
|
||||
assert_eq!(store.put_counts.lock().await.get(&key), Some(&3));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_leadership_claim_rejects_terminal_epoch() {
|
||||
let store = Arc::new(MemoryConfigStore::default());
|
||||
let ctx = CancellationToken::new();
|
||||
let mut revision = DataUsageCacheRevision::Missing;
|
||||
let mut cycle = CurrentCycle {
|
||||
next: 12,
|
||||
..Default::default()
|
||||
};
|
||||
let mut persisted_epoch = u64::MAX - 1;
|
||||
|
||||
assert!(!claim_scanner_leadership(&ctx, store.clone(), &mut cycle, &mut revision, &mut persisted_epoch).await);
|
||||
assert_eq!(persisted_epoch, u64::MAX - 1);
|
||||
assert!(read_config(store, &DATA_USAGE_BLOOM_NAME_PATH).await.is_err());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_leadership_claim_confirms_commit_after_returned_error() {
|
||||
let store = Arc::new(MemoryConfigStore::default());
|
||||
@@ -3878,24 +2975,6 @@ fn superseded_retry_backoff_grows_from_the_default_cycle() {
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn corrupt_cycle_state_backoff_uses_virtual_clock() {
|
||||
let mut backoff = ScannerRetryBackoff::default();
|
||||
backoff.record_retryable_cycle(true);
|
||||
let first_delay = backoff
|
||||
.retry_interval(Duration::from_secs(60))
|
||||
.expect("the first recovery retry should be scheduled");
|
||||
assert_eq!(first_delay, Duration::from_secs(5));
|
||||
|
||||
let deadline = Instant::now() + first_delay;
|
||||
assert!(Instant::now() < deadline);
|
||||
tokio::time::advance(first_delay).await;
|
||||
assert!(Instant::now() >= deadline);
|
||||
|
||||
backoff.record_retryable_cycle(true);
|
||||
assert_eq!(backoff.retry_interval(Duration::from_secs(60)), Some(Duration::from_secs(10)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scanner_cycle_wait_plan_drives_growth_resets_and_bitrot_cap() {
|
||||
let runtime_config = ScannerRuntimeConfig {
|
||||
|
||||
@@ -126,7 +126,6 @@ mod tests {
|
||||
let _list_remote_target_handler = replication::ListRemoteTargetHandler {};
|
||||
let _remove_remote_target_handler = replication::RemoveRemoteTargetHandler {};
|
||||
let _scanner_status_handler = scanner::ScannerStatusHandler {};
|
||||
let _scanner_cycle_state_reset_handler = scanner::ScannerCycleStateResetHandler {};
|
||||
let _ilm_expiry_status_handler = scanner::IlmExpiryStatusHandler {};
|
||||
let _manual_transition_handler = ilm_transition::ManualTransitionRunHandler {};
|
||||
let _manual_transition_status_handler = ilm_transition::ManualTransitionJobStatusHandler {};
|
||||
|
||||
@@ -13,11 +13,8 @@
|
||||
// limitations under the License.
|
||||
|
||||
use crate::admin::auth::authorize_admin_request;
|
||||
use crate::admin::handlers::supervise_admin_mutation;
|
||||
use crate::admin::router::{AdminOperation, Operation, S3Router};
|
||||
use crate::admin::runtime_sources::{
|
||||
app_context_from_req, current_object_store_handle_for_context, current_scanner_metrics_report,
|
||||
};
|
||||
use crate::admin::runtime_sources::current_scanner_metrics_report;
|
||||
use crate::module_switches::{ENV_SCANNER_ENABLED, scanner_enabled_from_env};
|
||||
use crate::server::ADMIN_PREFIX;
|
||||
use chrono::Utc;
|
||||
@@ -25,13 +22,11 @@ use http::{HeaderMap, HeaderValue};
|
||||
use hyper::{Method, StatusCode};
|
||||
use matchit::Params;
|
||||
use rustfs_common::metrics::{ScannerLifecycleExpirySnapshot, ScannerMaintenanceControlSnapshot, ScannerMetricsReport};
|
||||
use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE;
|
||||
use rustfs_credentials::Credentials;
|
||||
use rustfs_policy::policy::action::{Action, AdminAction};
|
||||
use s3s::header::CONTENT_TYPE;
|
||||
use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use tokio_util::sync::CancellationToken;
|
||||
use serde::Serialize;
|
||||
|
||||
const JSON_CONTENT_TYPE: &str = "application/json";
|
||||
|
||||
@@ -43,13 +38,6 @@ struct ScannerStatusResponse {
|
||||
metrics: ScannerMetricsReport,
|
||||
cycle_schedule: rustfs_scanner::ScannerCycleScheduleStatus,
|
||||
runtime_config: rustfs_scanner::runtime_config::ScannerRuntimeConfigStatus,
|
||||
cycle_recovery: rustfs_scanner::ScannerCycleRecoveryStatus,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
struct ScannerCycleResetRequest {
|
||||
mode: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize)]
|
||||
@@ -129,7 +117,6 @@ fn scanner_status_response(
|
||||
metrics,
|
||||
cycle_schedule,
|
||||
runtime_config,
|
||||
cycle_recovery: rustfs_scanner::scanner::scanner_cycle_recovery_status(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -157,11 +144,6 @@ pub fn register_scanner_route(r: &mut S3Router<AdminOperation>) -> std::io::Resu
|
||||
format!("{ADMIN_PREFIX}/v3/scanner/status").as_str(),
|
||||
AdminOperation(&ScannerStatusHandler {}),
|
||||
)?;
|
||||
r.insert(
|
||||
Method::POST,
|
||||
format!("{ADMIN_PREFIX}/v3/scanner/cycle-state/reset").as_str(),
|
||||
AdminOperation(&ScannerCycleStateResetHandler {}),
|
||||
)?;
|
||||
r.insert(
|
||||
Method::GET,
|
||||
format!("{ADMIN_PREFIX}/v3/ilm/expiry/status").as_str(),
|
||||
@@ -181,13 +163,6 @@ async fn validate_scanner_status_request(req: &S3Request<Body>) -> S3Result<Cred
|
||||
authorize_admin_request(req, vec![Action::AdminAction(AdminAction::ServerInfoAdminAction)]).await
|
||||
}
|
||||
|
||||
async fn validate_scanner_reset_request(req: &S3Request<Body>) -> S3Result<Credentials> {
|
||||
if req.credentials.is_none() {
|
||||
return Err(s3_error!(InvalidRequest, "missing credentials"));
|
||||
}
|
||||
authorize_admin_request(req, vec![Action::AdminAction(AdminAction::ConfigUpdateAdminAction)]).await
|
||||
}
|
||||
|
||||
fn json_response(body: Vec<u8>) -> S3Result<S3Response<(StatusCode, Body)>> {
|
||||
let mut headers = HeaderMap::new();
|
||||
let content_type = HeaderValue::from_str(JSON_CONTENT_TYPE)
|
||||
@@ -217,37 +192,6 @@ impl Operation for ScannerStatusHandler {
|
||||
|
||||
pub struct IlmExpiryStatusHandler {}
|
||||
|
||||
pub struct ScannerCycleStateResetHandler {}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl Operation for ScannerCycleStateResetHandler {
|
||||
async fn call(&self, mut req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
|
||||
let _cred = validate_scanner_reset_request(&req).await?;
|
||||
let body = req
|
||||
.input
|
||||
.store_all_limited(MAX_ADMIN_REQUEST_BODY_SIZE)
|
||||
.await
|
||||
.map_err(|err| S3Error::with_message(S3ErrorCode::InvalidRequest, format!("invalid reset request body: {err}")))?;
|
||||
let reset = serde_json::from_slice::<ScannerCycleResetRequest>(&body)
|
||||
.map_err(|err| S3Error::with_message(S3ErrorCode::InvalidRequest, format!("invalid reset request body: {err}")))?;
|
||||
if reset.mode != "full-rescan" {
|
||||
return Err(S3Error::with_message(S3ErrorCode::InvalidRequest, "reset mode must be full-rescan"));
|
||||
}
|
||||
let context = app_context_from_req(&req)
|
||||
.ok_or_else(|| S3Error::with_message(S3ErrorCode::InternalError, "storage layer not initialized"))?;
|
||||
let store = current_object_store_handle_for_context(Some(context.as_ref()))
|
||||
.ok_or_else(|| S3Error::with_message(S3ErrorCode::InternalError, "storage layer not initialized"))?;
|
||||
supervise_admin_mutation("scanner cycle state reset", async move {
|
||||
rustfs_scanner::scanner::reset_scanner_cycle_recovery(CancellationToken::new(), store)
|
||||
.await
|
||||
.map_err(|err| S3Error::with_message(S3ErrorCode::InternalError, err.to_string()))?;
|
||||
Ok::<_, S3Error>(())
|
||||
})
|
||||
.await?;
|
||||
json_response(br#"{"status":"reset","mode":"full-rescan"}"#.to_vec())
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl Operation for IlmExpiryStatusHandler {
|
||||
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
|
||||
@@ -293,38 +237,6 @@ mod tests {
|
||||
assert_eq!(err.message(), Some("missing credentials"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn scanner_reset_gate_rejects_missing_credentials() {
|
||||
let req = S3Request {
|
||||
input: Body::from(String::new()),
|
||||
method: Method::POST,
|
||||
uri: http::Uri::from_static("/rustfs/admin/v3/scanner/cycle-state/reset"),
|
||||
headers: HeaderMap::new(),
|
||||
extensions: http::Extensions::new(),
|
||||
credentials: None,
|
||||
region: None,
|
||||
service: None,
|
||||
trailing_headers: None,
|
||||
};
|
||||
|
||||
let err = validate_scanner_reset_request(&req)
|
||||
.await
|
||||
.expect_err("a reset request without credentials must be rejected");
|
||||
assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
|
||||
assert_eq!(err.message(), Some("missing credentials"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn admin_reset_requires_full_rescan_or_verified_cursor() {
|
||||
let full_rescan: ScannerCycleResetRequest =
|
||||
serde_json::from_str(r#"{"mode":"full-rescan"}"#).expect("full rescan must be accepted");
|
||||
assert_eq!(full_rescan.mode, "full-rescan");
|
||||
let cursor: ScannerCycleResetRequest =
|
||||
serde_json::from_str(r#"{"mode":"cursor"}"#).expect("mode validation belongs to the handler");
|
||||
assert_ne!(cursor.mode, "full-rescan");
|
||||
assert!(serde_json::from_str::<ScannerCycleResetRequest>(r#"{"mode":"full-rescan","cursor":"untrusted"}"#).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scanner_disabled_reason_reports_startup_env_key() {
|
||||
assert_eq!(scanner_disabled_reason(true), None);
|
||||
@@ -392,11 +304,6 @@ mod tests {
|
||||
assert_eq!(encoded["cycle_schedule"]["effective_interval_seconds"], 0);
|
||||
assert_eq!(encoded["cycle_schedule"]["clean_idle_backoff_enabled"], false);
|
||||
assert_eq!(encoded["cycle_schedule"]["clean_idle_backoff_multiplier"], 1);
|
||||
assert_eq!(encoded["cycle_recovery"]["state"], "healthy");
|
||||
assert_eq!(
|
||||
encoded["cycle_recovery"]["quarantine_path"],
|
||||
rustfs_scanner::DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -428,12 +428,6 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
|
||||
admin(HttpMethod::Get, "/rustfs/admin/v3/config", CONFIG_UPDATE, RouteRiskLevel::High),
|
||||
admin(HttpMethod::Put, "/rustfs/admin/v3/config", CONFIG_UPDATE, RouteRiskLevel::High),
|
||||
admin(HttpMethod::Get, "/rustfs/admin/v3/scanner/status", SERVER_INFO, RouteRiskLevel::Sensitive),
|
||||
admin(
|
||||
HttpMethod::Post,
|
||||
"/rustfs/admin/v3/scanner/cycle-state/reset",
|
||||
CONFIG_UPDATE,
|
||||
RouteRiskLevel::High,
|
||||
),
|
||||
admin(
|
||||
HttpMethod::Get,
|
||||
"/rustfs/admin/v3/ilm/expiry/status",
|
||||
@@ -2026,12 +2020,6 @@ mod tests {
|
||||
assert_not_action(HttpMethod::Get, "/rustfs/admin/v3/ilm/expiry/status", SET_TIER);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn route_policy_requires_config_update_for_scanner_cycle_reset() {
|
||||
assert_action(HttpMethod::Post, "/rustfs/admin/v3/scanner/cycle-state/reset", CONFIG_UPDATE);
|
||||
assert_not_action(HttpMethod::Post, "/rustfs/admin/v3/scanner/cycle-state/reset", SERVER_INFO);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn route_policy_uses_tier_actions_for_transition_routes() {
|
||||
assert_action(HttpMethod::Post, "/rustfs/admin/v3/ilm/transition/run", SET_TIER);
|
||||
|
||||
@@ -243,7 +243,6 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
|
||||
admin_route(Method::GET, "/v3/config"),
|
||||
admin_route(Method::PUT, "/v3/config"),
|
||||
admin_route(Method::GET, "/v3/scanner/status"),
|
||||
admin_route(Method::POST, "/v3/scanner/cycle-state/reset"),
|
||||
admin_route(Method::GET, "/v3/audit/target/list"),
|
||||
admin_route_sample(
|
||||
Method::PUT,
|
||||
@@ -880,7 +879,6 @@ fn test_register_routes_cover_representative_admin_paths() {
|
||||
assert_route(&router, Method::GET, &admin_path("/v3/config"));
|
||||
assert_route(&router, Method::PUT, &admin_path("/v3/config"));
|
||||
assert_route(&router, Method::GET, &admin_path("/v3/scanner/status"));
|
||||
assert_route(&router, Method::POST, &admin_path("/v3/scanner/cycle-state/reset"));
|
||||
assert_route(&router, Method::GET, &admin_path("/v3/ilm/expiry/status"));
|
||||
assert_route(&router, Method::POST, &admin_path("/v3/ilm/transition/run"));
|
||||
assert_route(
|
||||
@@ -1369,7 +1367,6 @@ fn test_admin_alias_paths_match_existing_admin_routes() {
|
||||
(Method::GET, compat_admin_alias_path("/v3/config")),
|
||||
(Method::PUT, compat_admin_alias_path("/v3/config")),
|
||||
(Method::GET, compat_admin_alias_path("/v3/scanner/status")),
|
||||
(Method::POST, compat_admin_alias_path("/v3/scanner/cycle-state/reset")),
|
||||
(Method::GET, compat_admin_alias_path("/v3/ilm/expiry/status")),
|
||||
] {
|
||||
assert!(
|
||||
|
||||
@@ -241,7 +241,7 @@ env \
|
||||
RUSTFS_TEST_VAULT_FAILOVER_MARKER="$MARKER" \
|
||||
RUSTFS_TEST_VAULT_OLD_LEADER="$OLD_LEADER" \
|
||||
cargo test -p rustfs-kms --test vault_ha_failover_live \
|
||||
vault_raft_leader_failure_recovers_kv2_and_transit_decrypts -- \
|
||||
vault_raft_leader_failure_preserves_kv2_and_transit_decrypts -- \
|
||||
--ignored --nocapture --test-threads=1 &
|
||||
TEST_PID=$!
|
||||
|
||||
|
||||
Reference in New Issue
Block a user