gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-21 03:46:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(scanner): split scanner.rs cycle/leadership/persist children (#6305)

Browse Source 
Split the 8178-line scanner.rs (48% inline tests) into a canonical
scanner.rs + scanner/ module tree with zero behavior change:

- scanner.rs (~2140): cycle constants, schedule status, budget/config
  helpers, startup, maintenance features, the two run loops, and
  cycle-result finalization
- scanner/activity.rs (~770): wake/backoff policy and scanner activity
  observation (probing, generations, topology digest)
- scanner/heal_info.rs (~110): the background-heal info object
- scanner/cycle_state.rs (~500): cycle-state codec, persisted usage
  floors, and cycle-state persistence
- scanner/leadership.rs (~360): leader-lock claiming, usage-epoch
  fencing, and lock-loss handling
- scanner/usage_store.rs (~480): the CAS data-usage store pipeline and
  observed-snapshot cleanup
- scanner/tests.rs (~3920): the inline test module as a child module

All crate paths are unchanged: scanner::BackgroundHealInfo,
scanner::read_background_heal_info, scanner::store_data_usage_in_backend,
and scanner_topology_digest resolve through root re-exports with their
original visibilities, and the pub(crate) surface used by scanner_io and
remote_scanner re-exports at pub(crate). Cross-module items gain
pub(super), whose scope equals the old single-module privacy domain.
Code is moved verbatim apart from those markers, per-module import
headers, and rustfmt re-wraps.

Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
2026-08-20 19:12:27 +08:00
committed by GitHub
parent 2cf0ad0f85
commit 095bf34086
7 changed files with 6167 additions and 6059 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/scanner/src/scanner.rs
+21 -6059
View File
File diff suppressed because it is too large Load Diff
crates/scanner/src/scanner/activity.rs
+773
View File
@@ -0,0 +1,773 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
/// Cycle wake/backoff policy and scanner activity observation (probing, generations, topology digest).
use super::*;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum ScannerCycleWakeReason {
Timer,
DirtyUsage,
ClusterActivity,
ClusterMaintenance,
ClusterActivityUnavailable,
RuntimeConfig,
MaintenanceConfig,
LeaderLockLost,
Cancelled,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum ScannerCycleOutcome {
Completed,
CompletedWithPendingMaintenance,
Partial,
Superseded,
Deferred(ScannerCycleDeferReason),
Failed,
}
pub(crate) fn scanner_cycle_outcome_with_pending_maintenance(
outcome: ScannerCycleOutcome,
pending_maintenance_work: bool,
) -> ScannerCycleOutcome {
if outcome == ScannerCycleOutcome::Completed && pending_maintenance_work {
ScannerCycleOutcome::CompletedWithPendingMaintenance
} else {
outcome
}
}
pub(super) async fn remote_dirty_usage_acknowledgement_pending<F, E>(
cycle: u64,
acknowledgement_count: usize,
acknowledgement: F,
) -> bool
where
F: Future<Output = Result<bool, E>>,
E: std::fmt::Display,
{
match acknowledgement.await {
Ok(dirty_usage_pending) => dirty_usage_pending,
Err(err) => {
warn!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
cycle,
acknowledgement_count,
error = %err,
state = "remote_dirty_usage_acknowledgement_pending",
"Scanner cycle left remote dirty usage acknowledgements pending"
);
true
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) struct ScannerCleanIdleBackoff {
pub(super) interval_multiplier: u32,
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub(super) struct ScannerRetryBackoff {
pub(super) consecutive_cycles: u32,
}
impl ScannerRetryBackoff {
pub(super) fn record_retryable_cycle(&mut self, retryable: bool) {
if retryable {
self.consecutive_cycles = self.consecutive_cycles.saturating_add(1);
} else {
self.consecutive_cycles = 0;
}
}
pub(super) fn retry_interval(self, configured_interval: Duration) -> Option<Duration> {
let exponent = self.consecutive_cycles.checked_sub(1)?.min(31);
let multiplier = 1u32.checked_shl(exponent).unwrap_or(u32::MAX);
let base_interval = configured_interval
.max(Duration::from_secs(1))
.min(SCANNER_RETRY_BASE_INTERVAL);
let cap = SCANNER_RETRY_MAX_INTERVAL.max(configured_interval.max(Duration::from_secs(1)));
Some(base_interval.saturating_mul(multiplier).min(cap))
}
}
impl Default for ScannerCleanIdleBackoff {
fn default() -> Self {
Self { interval_multiplier: 1 }
}
}
impl ScannerCleanIdleBackoff {
pub(super) fn reset(&mut self) {
self.interval_multiplier = 1;
}
pub(super) fn effective_interval(self, base_interval: Duration, max_interval: Duration, enabled: bool) -> Duration {
let base_interval = base_interval.max(Duration::from_secs(1));
if !enabled {
return base_interval;
}
let max_interval = max_interval.max(base_interval);
base_interval.saturating_mul(self.interval_multiplier).min(max_interval)
}
pub(super) fn record_cycle(
&mut self,
base_interval: Duration,
max_interval: Duration,
enabled: bool,
wake_reason: ScannerCycleWakeReason,
outcome: ScannerCycleOutcome,
dirty_work_observed: bool,
) {
if !enabled
|| wake_reason != ScannerCycleWakeReason::Timer
|| outcome != ScannerCycleOutcome::Completed
|| dirty_work_observed
{
self.reset();
return;
}
let max_interval = max_interval.max(base_interval.max(Duration::from_secs(1)));
if self.effective_interval(base_interval, max_interval, true) < max_interval {
self.interval_multiplier = self.interval_multiplier.saturating_mul(CLEAN_IDLE_BACKOFF_FACTOR);
}
}
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub(super) struct ScannerMaintenanceInspectionRetry {
pub(super) consecutive_failures: u32,
pub(super) retry_at: Option<Instant>,
}
impl ScannerMaintenanceInspectionRetry {
pub(super) fn from_features(features: ScannerMaintenanceFeatures, now: Instant) -> Self {
let mut retry = Self::default();
retry.record_inspection(features, now);
retry
}
pub(super) fn reset(&mut self) {
self.consecutive_failures = 0;
self.retry_at = None;
}
pub(super) fn retry_interval(self) -> Option<Duration> {
if self.consecutive_failures == 0 {
return None;
}
let exponent = self.consecutive_failures.saturating_sub(1).min(31);
let multiplier = 1u32.checked_shl(exponent).unwrap_or(u32::MAX);
Some(
MAINTENANCE_FEATURE_INSPECTION_RETRY_BASE_INTERVAL
.saturating_mul(multiplier)
.min(MAINTENANCE_FEATURE_INSPECTION_RETRY_MAX_INTERVAL),
)
}
pub(super) fn record_inspection(&mut self, features: ScannerMaintenanceFeatures, now: Instant) {
if !features.inspection_failed {
self.reset();
return;
}
self.consecutive_failures = self.consecutive_failures.saturating_add(1);
self.retry_at = self.retry_interval().map(|interval| now + interval);
}
pub(super) fn retry_due(
self,
features: ScannerMaintenanceFeatures,
wake_reason: ScannerCycleWakeReason,
now: Instant,
) -> bool {
features.inspection_failed
&& wake_reason == ScannerCycleWakeReason::Timer
&& self.retry_at.is_some_and(|retry_at| now >= retry_at)
}
}
pub(super) fn scanner_cycle_observed_dirty_work(
pending_before_wait: bool,
generation_before_wait: u64,
generation_after_cycle: u64,
) -> bool {
pending_before_wait || generation_before_wait != generation_after_cycle
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) struct ScannerCycleWaitPlan {
pub(super) effective_interval: Duration,
pub(super) clean_idle_max_interval: Duration,
pub(super) delay: Duration,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) struct ScannerCycleObservedGenerations {
pub(super) dirty_usage: Option<u64>,
pub(super) runtime_config: u64,
pub(super) maintenance: u64,
pub(super) defer_cluster_activity: bool,
}
pub(super) const LOCAL_SCANNER_ACTIVITY_NODE: &str = "<local>";
#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) struct ScannerNodeActivity {
pub(super) instance_id: String,
pub(super) namespace_generation: u64,
pub(super) maintenance_generation: u64,
pub(super) protocol_version: u32,
pub(super) topology_digest: [u8; 32],
pub(super) data_movement_active: bool,
pub(super) dirty_usage_generation: u64,
pub(super) dirty_usage_pending: bool,
}
pub(crate) type ScannerActivitySnapshot = BTreeMap<String, ScannerNodeActivity>;
#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) struct ScannerDirtyUsageAcknowledgement {
pub(crate) host: String,
pub(crate) instance_id: String,
pub(crate) generation: u64,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum ScannerActivityObservation {
NotRequired,
Unchanged,
Changed,
MaintenanceChanged,
Unverified,
}
pub(super) fn scanner_cycle_wait_plan(
runtime_config: &ScannerRuntimeConfig,
clean_idle_backoff: ScannerCleanIdleBackoff,
clean_idle_backoff_enabled: bool,
jitter: impl FnOnce(Duration) -> Duration,
) -> ScannerCycleWaitPlan {
let clean_idle_max_interval = scanner_clean_idle_max_interval(runtime_config.cycle_interval, runtime_config);
let effective_interval =
clean_idle_backoff.effective_interval(runtime_config.cycle_interval, clean_idle_max_interval, clean_idle_backoff_enabled);
let delay = cap_clean_idle_cycle_delay(jitter(effective_interval), clean_idle_max_interval, clean_idle_backoff_enabled);
ScannerCycleWaitPlan {
effective_interval,
clean_idle_max_interval,
delay,
}
}
pub(super) fn record_scanner_cycle_result(
clean_idle_backoff: &mut ScannerCleanIdleBackoff,
runtime_config: &ScannerRuntimeConfig,
clean_idle_backoff_enabled: bool,
wake_reason: ScannerCycleWakeReason,
outcome: ScannerCycleOutcome,
dirty_work_observed: bool,
) {
clean_idle_backoff.record_cycle(
runtime_config.cycle_interval,
scanner_clean_idle_max_interval(runtime_config.cycle_interval, runtime_config),
clean_idle_backoff_enabled,
wake_reason,
outcome,
dirty_work_observed,
);
}
pub(super) fn scanner_clean_idle_backoff_configured(runtime_config: &ScannerRuntimeConfig) -> bool {
let bitrot_cycle_allows_backoff =
runtime_config.bitrot_cycle.is_none() || runtime_config.bitrot_cycle_source == ScannerRuntimeConfigSource::Default;
runtime_config.cycle_interval_source == ScannerRuntimeConfigSource::Default && bitrot_cycle_allows_backoff
}
pub(super) fn scanner_clean_idle_max_interval(base_interval: Duration, runtime_config: &ScannerRuntimeConfig) -> Duration {
let policy_max = CLEAN_IDLE_MAX_INTERVAL.max(base_interval);
let Some(bitrot_cycle) = runtime_config.bitrot_cycle else {
return policy_max;
};
if runtime_config.bitrot_cycle_source != ScannerRuntimeConfigSource::Default {
return policy_max;
}
let selection_window = heal_object_select_prob();
if selection_window == 0 {
return policy_max;
}
bitrot_cycle
.checked_div(selection_window)
.unwrap_or(base_interval)
.max(base_interval)
.min(policy_max)
}
pub(super) fn scanner_clean_idle_backoff_enabled(
topology_supported: bool,
cluster_activity_ready: bool,
features: ScannerMaintenanceFeatures,
runtime_config: &ScannerRuntimeConfig,
) -> bool {
topology_supported
&& cluster_activity_ready
&& !features.needs_regular_cycle()
&& scanner_clean_idle_backoff_configured(runtime_config)
}
pub(super) fn scanner_activity_probe_required(
topology_supported: bool,
backoff_blocked: bool,
features: ScannerMaintenanceFeatures,
runtime_config: &ScannerRuntimeConfig,
) -> bool {
topology_supported
&& !backoff_blocked
&& !features.needs_regular_cycle()
&& scanner_clean_idle_backoff_configured(runtime_config)
}
pub(super) fn scanner_activity_observed_work(observation: ScannerActivityObservation) -> bool {
matches!(
observation,
ScannerActivityObservation::Changed
| ScannerActivityObservation::MaintenanceChanged
| ScannerActivityObservation::Unverified
)
}
pub(super) fn scanner_activity_backoff_blocked_after_wake(currently_blocked: bool, wake_reason: ScannerCycleWakeReason) -> bool {
match wake_reason {
ScannerCycleWakeReason::ClusterMaintenance => true,
ScannerCycleWakeReason::MaintenanceConfig => false,
_ => currently_blocked,
}
}
pub(super) async fn wait_for_next_scanner_cycle<F>(
ctx: &CancellationToken,
delay: Duration,
dirty_usage_generation_seen: Option<u64>,
runtime_config_generation: u64,
maintenance_generation: u64,
is_lock_lost: F,
) -> ScannerCycleWakeReason
where
F: Fn() -> bool,
{
let sleep = tokio::time::sleep(delay);
tokio::pin!(sleep);
let lock_poll = tokio::time::sleep(SCANNER_LEADER_LOCK_POLL_INTERVAL);
tokio::pin!(lock_poll);
loop {
if is_lock_lost() {
return ScannerCycleWakeReason::LeaderLockLost;
}
if scanner_runtime_config_generation() != runtime_config_generation {
return ScannerCycleWakeReason::RuntimeConfig;
}
if scanner_maintenance_generation() != maintenance_generation {
return ScannerCycleWakeReason::MaintenanceConfig;
}
if dirty_usage_generation_seen.is_some_and(|seen| dirty_usage_buckets_pending() && dirty_usage_generation() != seen) {
return ScannerCycleWakeReason::DirtyUsage;
}
tokio::select! {
_ = ctx.cancelled() => return ScannerCycleWakeReason::Cancelled,
_ = &mut sleep => return ScannerCycleWakeReason::Timer,
_ = &mut lock_poll => {
if is_lock_lost() {
return ScannerCycleWakeReason::LeaderLockLost;
}
lock_poll.as_mut().reset(Instant::now() + SCANNER_LEADER_LOCK_POLL_INTERVAL);
}
_ = dirty_usage_bucket_notified() => {
if scanner_runtime_config_generation() != runtime_config_generation {
return ScannerCycleWakeReason::RuntimeConfig;
}
if scanner_maintenance_generation() != maintenance_generation {
return ScannerCycleWakeReason::MaintenanceConfig;
}
if dirty_usage_generation_seen
.is_some_and(|seen| dirty_usage_buckets_pending() && dirty_usage_generation() != seen)
{
return ScannerCycleWakeReason::DirtyUsage;
}
}
_ = scanner_runtime_config_changed() => {
if scanner_runtime_config_generation() != runtime_config_generation {
return ScannerCycleWakeReason::RuntimeConfig;
}
}
_ = scanner_maintenance_changed() => {
if scanner_maintenance_generation() != maintenance_generation {
return ScannerCycleWakeReason::MaintenanceConfig;
}
}
}
}
}
pub(super) async fn wait_for_next_scanner_cycle_with_activity<F, Probe, ProbeFuture>(
ctx: &CancellationToken,
delay: Duration,
activity_poll_interval: Option<Duration>,
activity_seen: &mut Option<ScannerActivitySnapshot>,
generations: ScannerCycleObservedGenerations,
is_lock_lost: F,
mut probe_activity: Probe,
) -> ScannerCycleWakeReason
where
F: Fn() -> bool,
Probe: FnMut() -> ProbeFuture,
ProbeFuture: Future<Output = Result<ScannerActivitySnapshot, String>>,
{
let deadline = Instant::now() + delay;
loop {
let remaining = deadline.saturating_duration_since(Instant::now());
if remaining.is_zero() {
return ScannerCycleWakeReason::Timer;
}
let wait_slice = activity_poll_interval
.map(|interval| interval.max(Duration::from_secs(1)).min(remaining))
.unwrap_or(remaining);
let wake_reason = wait_for_next_scanner_cycle(
ctx,
wait_slice,
generations.dirty_usage,
generations.runtime_config,
generations.maintenance,
&is_lock_lost,
)
.await;
if wake_reason != ScannerCycleWakeReason::Timer || Instant::now() >= deadline {
return wake_reason;
}
let Some(_) = activity_poll_interval else {
return ScannerCycleWakeReason::Timer;
};
if is_lock_lost() {
return ScannerCycleWakeReason::LeaderLockLost;
}
let probe = probe_activity();
tokio::pin!(probe);
let lock_lost = async {
loop {
tokio::time::sleep(SCANNER_LEADER_LOCK_POLL_INTERVAL).await;
if is_lock_lost() {
break;
}
}
};
tokio::pin!(lock_lost);
let probe_result = tokio::select! {
result = &mut probe => result,
_ = ctx.cancelled() => return ScannerCycleWakeReason::Cancelled,
_ = &mut lock_lost => return ScannerCycleWakeReason::LeaderLockLost,
};
let had_baseline = activity_seen.is_some();
let (observation, probe_error) = apply_scanner_activity_probe_result(activity_seen, probe_result);
if let Some(err) = probe_error {
log_scanner_activity_probe_error(had_baseline, &err);
}
match observation {
ScannerActivityObservation::Unchanged | ScannerActivityObservation::NotRequired => {}
ScannerActivityObservation::Changed if !generations.defer_cluster_activity => {
return ScannerCycleWakeReason::ClusterActivity;
}
ScannerActivityObservation::Changed => {}
ScannerActivityObservation::MaintenanceChanged => return ScannerCycleWakeReason::ClusterMaintenance,
ScannerActivityObservation::Unverified if !generations.defer_cluster_activity => {
return ScannerCycleWakeReason::ClusterActivityUnavailable;
}
ScannerActivityObservation::Unverified => {}
}
}
}
pub(super) fn log_scanner_activity_probe_error(had_baseline: bool, err: &str) {
if had_baseline {
warn!(
target: "rustfs::scanner",
event = EVENT_SCANNER_CYCLE_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
state = "cluster_activity_probe_failed",
error = %err,
"Scanner cluster activity probe failed; preserving the base cycle"
);
} else {
debug!(
target: "rustfs::scanner",
event = EVENT_SCANNER_CYCLE_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
state = "cluster_activity_probe_unavailable",
error = %err,
"Scanner cluster activity probe remains unavailable"
);
}
}
pub(super) fn compare_scanner_activity(
previous: &ScannerActivitySnapshot,
current: &ScannerActivitySnapshot,
) -> ScannerActivityObservation {
if previous == current {
return ScannerActivityObservation::Unchanged;
}
for (host, current_activity) in current {
let Some(previous_activity) = previous.get(host) else {
continue;
};
if host != LOCAL_SCANNER_ACTIVITY_NODE
&& previous_activity.instance_id == current_activity.instance_id
&& previous_activity.maintenance_generation != current_activity.maintenance_generation
{
return ScannerActivityObservation::MaintenanceChanged;
}
}
ScannerActivityObservation::Changed
}
pub(super) fn apply_scanner_activity_probe_result(
activity_seen: &mut Option<ScannerActivitySnapshot>,
result: Result<ScannerActivitySnapshot, String>,
) -> (ScannerActivityObservation, Option<String>) {
match result {
Ok(current) => {
let observation = match activity_seen.as_ref() {
Some(previous) => compare_scanner_activity(previous, &current),
None => ScannerActivityObservation::Unverified,
};
*activity_seen = Some(current);
(observation, None)
}
Err(err) => {
*activity_seen = None;
(ScannerActivityObservation::Unverified, Some(err))
}
}
}
pub(super) async fn observe_scanner_activity(
storeapi: &Arc<ECStore>,
distributed: bool,
activity_seen: &mut Option<ScannerActivitySnapshot>,
) -> ScannerActivityObservation {
let had_baseline = activity_seen.is_some();
let (observation, probe_error) =
apply_scanner_activity_probe_result(activity_seen, probe_scanner_activity(storeapi, distributed).await);
if let Some(err) = probe_error {
log_scanner_activity_probe_error(had_baseline, &err);
}
observation
}
pub(crate) fn scanner_activity_snapshot_digest(snapshot: &ScannerActivitySnapshot) -> [u8; 32] {
let mut hasher = Sha256::new();
hasher.update(u64::try_from(snapshot.len()).unwrap_or(u64::MAX).to_be_bytes());
for (host, activity) in snapshot {
let host = host.as_bytes();
let instance_id = activity.instance_id.as_bytes();
hasher.update(u64::try_from(host.len()).unwrap_or(u64::MAX).to_be_bytes());
hasher.update(host);
hasher.update(u64::try_from(instance_id.len()).unwrap_or(u64::MAX).to_be_bytes());
hasher.update(instance_id);
hasher.update(activity.namespace_generation.to_be_bytes());
hasher.update(activity.maintenance_generation.to_be_bytes());
hasher.update(activity.protocol_version.to_be_bytes());
hasher.update(activity.topology_digest);
hasher.update([u8::from(activity.data_movement_active)]);
hasher.update(activity.dirty_usage_generation.to_be_bytes());
hasher.update([u8::from(activity.dirty_usage_pending)]);
}
hasher.finalize().into()
}
pub(crate) fn scanner_activity_allows_usage_publication(snapshot: &ScannerActivitySnapshot) -> bool {
snapshot.values().all(|activity| !activity.data_movement_active)
}
pub(crate) fn scanner_dirty_usage_acknowledgements(snapshot: &ScannerActivitySnapshot) -> Vec<ScannerDirtyUsageAcknowledgement> {
snapshot
.iter()
.filter(|(host, activity)| host.as_str() != LOCAL_SCANNER_ACTIVITY_NODE && activity.dirty_usage_pending)
.map(|(host, activity)| ScannerDirtyUsageAcknowledgement {
host: host.clone(),
instance_id: activity.instance_id.clone(),
generation: activity.dirty_usage_generation,
})
.collect()
}
pub fn scanner_topology_digest(storeapi: &ECStore) -> [u8; 32] {
let endpoint_pools = storeapi.endpoints();
let mut hasher = Sha256::new();
hasher.update(u64::try_from(endpoint_pools.0.len()).unwrap_or(u64::MAX).to_be_bytes());
for (pool_index, pool) in endpoint_pools.0.iter().enumerate() {
hasher.update(u64::try_from(pool_index).unwrap_or(u64::MAX).to_be_bytes());
hasher.update(u64::try_from(pool.set_count).unwrap_or(u64::MAX).to_be_bytes());
hasher.update(u64::try_from(pool.drives_per_set).unwrap_or(u64::MAX).to_be_bytes());
let mut endpoints = pool.endpoints.as_ref().iter().collect::<Vec<_>>();
endpoints.sort_unstable_by(|left, right| {
(left.pool_idx, left.set_idx, left.disk_idx, left.url.as_str()).cmp(&(
right.pool_idx,
right.set_idx,
right.disk_idx,
right.url.as_str(),
))
});
hasher.update(u64::try_from(endpoints.len()).unwrap_or(u64::MAX).to_be_bytes());
for endpoint in endpoints {
hasher.update(endpoint.pool_idx.to_be_bytes());
hasher.update(endpoint.set_idx.to_be_bytes());
hasher.update(endpoint.disk_idx.to_be_bytes());
let url = endpoint.url.as_str().as_bytes();
hasher.update(u64::try_from(url.len()).unwrap_or(u64::MAX).to_be_bytes());
hasher.update(url);
}
}
hasher.finalize().into()
}
pub(super) fn record_scanner_activity_instance(
instance_hosts: &mut BTreeMap<String, String>,
host: &str,
instance_id: &str,
) -> Result<(), String> {
if let Some(existing_host) = instance_hosts.insert(instance_id.to_string(), host.to_string()) {
return Err(format!(
"scanner activity peers {existing_host} and {host} report the same process instance"
));
}
Ok(())
}
pub(crate) async fn probe_scanner_activity(storeapi: &ECStore, distributed: bool) -> Result<ScannerActivitySnapshot, String> {
let topology_digest = scanner_topology_digest(storeapi);
let data_movement_active = storeapi.scanner_data_movement_active().await;
let namespace_generation = storeapi.scanner_namespace_mutation_generation();
let maintenance_generation = scanner_maintenance_generation();
let dirty_usage = scanner_dirty_usage_state();
if namespace_generation == u64::MAX || maintenance_generation == u64::MAX || dirty_usage.generation == u64::MAX {
return Err("local scanner activity generation is exhausted".to_string());
}
let local_instance_id = crate::scanner_io::scanner_activity_epoch().to_string();
let mut instance_hosts = BTreeMap::from([(local_instance_id.clone(), LOCAL_SCANNER_ACTIVITY_NODE.to_string())]);
let mut snapshot = ScannerActivitySnapshot::from([(
LOCAL_SCANNER_ACTIVITY_NODE.to_string(),
ScannerNodeActivity {
instance_id: local_instance_id,
namespace_generation,
maintenance_generation,
protocol_version: SCANNER_ACTIVITY_PROTOCOL_VERSION,
topology_digest,
data_movement_active,
dirty_usage_generation: dirty_usage.generation,
dirty_usage_pending: dirty_usage.pending,
},
)]);
if !distributed {
return Ok(snapshot);
}
let notification_system = storeapi
.notification_system()
.ok_or_else(|| "notification system is not initialized".to_string())?;
let peers = notification_system
.scanner_activity_snapshots()
.await
.map_err(|err| err.to_string())?;
for (host, activity) in peers {
if activity.namespace_generation == u64::MAX || activity.maintenance_generation == u64::MAX {
return Err(format!("scanner activity peer {host} exhausted its activity generation"));
}
let (peer_topology_digest, peer_data_movement_active, peer_dirty_usage_generation, peer_dirty_usage_pending) =
match activity.protocol_version {
SCANNER_ACTIVITY_LEGACY_PROTOCOL_VERSION => {
return Err(format!("scanner activity peer {host} cannot verify data movement publication fencing"));
}
SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION => {
return Err(format!(
"scanner activity peer {host} cannot safely share scanner cache locks with protocol {}",
SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION
));
}
SCANNER_ACTIVITY_PROTOCOL_VERSION => (
activity
.topology_digest
.ok_or_else(|| format!("scanner activity peer {host} omitted its storage topology"))?,
activity
.data_movement_active
.ok_or_else(|| format!("scanner activity peer {host} omitted its data movement state"))?,
activity
.dirty_usage_generation
.ok_or_else(|| format!("scanner activity peer {host} omitted its dirty usage generation"))?,
activity
.dirty_usage_pending
.ok_or_else(|| format!("scanner activity peer {host} omitted its dirty usage state"))?,
),
version => {
return Err(format!(
"scanner activity peer {host} uses protocol {version}, expected {}",
SCANNER_ACTIVITY_PROTOCOL_VERSION
));
}
};
if peer_dirty_usage_generation == u64::MAX {
return Err(format!("scanner activity peer {host} exhausted its dirty usage generation"));
}
if peer_topology_digest != topology_digest {
return Err(format!("scanner activity peer {host} has a different storage topology"));
}
record_scanner_activity_instance(&mut instance_hosts, &host, &activity.instance_id)?;
if snapshot
.insert(
host.clone(),
ScannerNodeActivity {
instance_id: activity.instance_id,
namespace_generation: activity.namespace_generation,
maintenance_generation: activity.maintenance_generation,
protocol_version: activity.protocol_version,
topology_digest: peer_topology_digest,
data_movement_active: peer_data_movement_active,
dirty_usage_generation: peer_dirty_usage_generation,
dirty_usage_pending: peer_dirty_usage_pending,
},
)
.is_some()
{
return Err(format!("duplicate scanner activity peer: {host}"));
}
}
Ok(snapshot)
}
crates/scanner/src/scanner/cycle_state.rs
+498
View File
@@ -0,0 +1,498 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
/// Scanner cycle-state codec, persisted usage floors, and cycle-state persistence.
use super::*;
#[derive(Debug, thiserror::Error)]
pub(super) enum ScannerCycleStateError {
#[error("failed to encode scanner cycle state: {0}")]
Encode(#[from] rmp_serde::encode::Error),
#[error("failed to decode scanner cycle state: {0}")]
Decode(#[from] rmp_serde::decode::Error),
#[error("{0}")]
InvalidData(&'static str),
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub(super) struct PersistedUsageFloor {
pub(super) next_cycle: u64,
pub(super) leader_epoch: u64,
}
pub(super) fn encode_scanner_cycle_state(
cycle_info: &CurrentCycle,
leader_epoch: u64,
) -> Result<Vec<u8>, ScannerCycleStateError> {
if cycle_info.next == u64::MAX {
return Err(ScannerCycleStateError::InvalidData("scanner cycle counter is exhausted"));
}
let cycle_info_buf = rmp_serde::to_vec(cycle_info)?;
let mut buf = Vec::with_capacity(cycle_info_buf.len() + SCANNER_CYCLE_STATE_HEADER_LEN);
buf.extend_from_slice(&cycle_info.next.to_le_bytes());
buf.extend_from_slice(SCANNER_CYCLE_STATE_MAGIC);
buf.extend_from_slice(&leader_epoch.to_le_bytes());
buf.extend_from_slice(&cycle_info_buf);
Ok(buf)
}
pub(super) fn decode_scanner_cycle_state(buf: &[u8]) -> Result<(CurrentCycle, u64), ScannerCycleStateError> {
if buf.len() < 8 {
return Err(ScannerCycleStateError::InvalidData("scanner cycle state is truncated"));
}
let persisted_next = u64::from_le_bytes(
buf[0..8]
.try_into()
.map_err(|_| ScannerCycleStateError::InvalidData("scanner cycle counter is truncated"))?,
);
if persisted_next == u64::MAX {
return Err(ScannerCycleStateError::InvalidData("scanner cycle counter is exhausted"));
}
if buf.len() == 8 {
return Ok((
CurrentCycle {
next: persisted_next,
..Default::default()
},
0,
));
}
let (leader_epoch, payload) = if buf.len() >= 16 && &buf[8..16] == SCANNER_CYCLE_STATE_MAGIC {
if buf.len() < SCANNER_CYCLE_STATE_HEADER_LEN {
return Err(ScannerCycleStateError::InvalidData("scanner cycle fencing header is truncated"));
}
let epoch = u64::from_le_bytes(
buf[16..24]
.try_into()
.map_err(|_| ScannerCycleStateError::InvalidData("scanner leader epoch is truncated"))?,
);
if epoch == 0 {
return Err(ScannerCycleStateError::InvalidData("scanner leader epoch is zero"));
}
(epoch, &buf[SCANNER_CYCLE_STATE_HEADER_LEN..])
} else {
(0, &buf[8..])
};
let cycle_info = rmp_serde::from_slice::<CurrentCycle>(payload)?;
if cycle_info.next != persisted_next {
return Err(ScannerCycleStateError::InvalidData("scanner cycle counter disagrees with encoded state"));
}
Ok((cycle_info, leader_epoch))
}
pub(crate) fn decode_persisted_scanner_cycle_fence(buf: &[u8]) -> Result<(u64, u64), ScannerError> {
decode_scanner_cycle_state(buf)
.map(|(cycle, leader_epoch)| (cycle.next, leader_epoch))
.map_err(|err| ScannerError::Other(format!("persisted scanner cycle state is invalid: {err}")))
}
#[cfg(test)]
pub(crate) fn encode_scanner_cycle_fence_for_test(next_cycle: u64, leader_epoch: u64) -> Vec<u8> {
encode_scanner_cycle_state(
&CurrentCycle {
next: next_cycle,
..Default::default()
},
leader_epoch,
)
.expect("test scanner cycle fence should encode")
}
pub(crate) async fn current_scanner_leader_epoch() -> Result<u64, ScannerError> {
let store = crate::resolve_scanner_object_store_handle()
.ok_or_else(|| ScannerError::Other("scanner object layer is unavailable".to_string()))?;
match read_config(store, &DATA_USAGE_BLOOM_NAME_PATH).await {
Ok(buf) => {
let (_, leader_epoch) = decode_persisted_scanner_cycle_fence(&buf)?;
if leader_epoch == 0 {
return Err(ScannerError::Other("persisted scanner cycle state has no leader epoch".to_string()));
}
Ok(leader_epoch)
}
Err(err) => Err(ScannerError::Other(format!("failed to read persisted scanner leader epoch: {err}"))),
}
}
pub(super) fn decode_scanner_cycle_state_for_startup(buf: &[u8]) -> Result<(CurrentCycle, u64), ScannerCycleStateError> {
if buf.is_empty() {
Ok((CurrentCycle::default(), 0))
} else {
decode_scanner_cycle_state(buf)
}
}
pub(super) fn advance_scanner_cycle(cycle_info: &mut CurrentCycle) -> Result<(), ScannerCycleStateError> {
let next = cycle_info
.next
.checked_add(1)
.filter(|next| *next < u64::MAX)
.ok_or(ScannerCycleStateError::InvalidData("scanner cycle counter is exhausted"))?;
cycle_info.next = next;
Ok(())
}
pub(super) async fn persisted_usage_floor(storeapi: Arc<impl ScannerObjectIO>) -> Result<PersistedUsageFloor, ScannerError> {
let mut floor = PersistedUsageFloor::default();
let update_floor = |floor: &mut PersistedUsageFloor, usage: DataUsageInfo, path: &str| -> Result<(), ScannerError> {
floor.leader_epoch = floor.leader_epoch.max(usage.scanner_epoch.unwrap_or_default());
if let Some(completed_cycle) = usage.scanner_cycle {
let next_cycle = completed_cycle
.checked_add(1)
.filter(|next| *next < u64::MAX)
.ok_or_else(|| ScannerError::Other(format!("persisted scanner usage cycle is exhausted in {path}")))?;
floor.next_cycle = floor.next_cycle.max(next_cycle);
}
Ok(())
};
for primary_path in [DATA_USAGE_OBJ_NAME_PATH.as_str(), LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()] {
let backup_path = format!("{primary_path}.bkp");
let mut pair_found = false;
for path in [primary_path, backup_path.as_str()] {
let data = match read_config(storeapi.clone(), path).await {
Ok(data) => {
pair_found = true;
data
}
Err(EcstoreError::ConfigNotFound) => continue,
Err(err) => {
return Err(ScannerError::Other(format!(
"failed to read scanner usage epoch floor from {path}: {err}"
)));
}
};
let usage = serde_json::from_slice::<DataUsageInfo>(&data)
.map_err(|err| ScannerError::Other(format!("failed to decode scanner usage floor from {path}: {err}")))?;
update_floor(&mut floor, usage, path)?;
}
if pair_found {
break;
}
}
Ok(floor)
}
pub(super) fn apply_persisted_usage_floor(cycle_info: &mut CurrentCycle, leader_epoch: &mut u64, floor: PersistedUsageFloor) {
cycle_info.next = cycle_info.next.max(floor.next_cycle);
*leader_epoch = (*leader_epoch).max(floor.leader_epoch);
}
pub(super) async fn persist_scanner_cycle_state(
ctx: &CancellationToken,
storeapi: Arc<impl ScannerObjectIO>,
cycle_info: &mut CurrentCycle,
revision: &mut DataUsageCacheRevision,
leader_epoch: u64,
) -> bool {
let buf = match encode_scanner_cycle_state(cycle_info, leader_epoch) {
Ok(buf) => buf,
Err(e) => {
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %&*DATA_USAGE_BLOOM_NAME_PATH,
state = "encode_failed",
error = %e,
"Scanner state encoding failed"
);
return false;
}
};
for retry in 0..=SCANNER_PERSIST_CAS_RETRIES {
if ctx.is_cancelled() {
debug!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %&*DATA_USAGE_BLOOM_NAME_PATH,
state = "cancelled_before_save",
retry,
"Scanner state persistence cancelled by the leader fence"
);
return false;
}
#[cfg(test)]
notify_scanner_cycle_state_persist_test_hook(leader_epoch);
match save_config_with_preconditions(storeapi.clone(), &DATA_USAGE_BLOOM_NAME_PATH, buf.clone(), revision.preconditions())
.await
{
Ok(object_info) => {
let Some(etag) = object_info.etag.filter(|etag| !etag.is_empty()) else {
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %&*DATA_USAGE_BLOOM_NAME_PATH,
state = "missing_revision",
"Scanner state save returned no ETag"
);
return false;
};
*revision = DataUsageCacheRevision::Etag(etag);
if ctx.is_cancelled() {
debug!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %&*DATA_USAGE_BLOOM_NAME_PATH,
state = "cancelled_after_save",
retry,
"Scanner state save completed after the leader fence was cancelled"
);
return false;
}
debug!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %&*DATA_USAGE_BLOOM_NAME_PATH,
state = "saved",
"Scanner state saved"
);
return true;
}
Err(EcstoreError::PreconditionFailed) => {
let (persisted, persisted_revision) =
match read_config_with_revision(storeapi.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str()).await {
Ok(result) => result,
Err(e) => {
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %&*DATA_USAGE_BLOOM_NAME_PATH,
state = "conflict_reload_failed",
error = %e,
"Scanner state conflict reconciliation failed"
);
return false;
}
};
*revision = persisted_revision;
if ctx.is_cancelled() {
debug!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %&*DATA_USAGE_BLOOM_NAME_PATH,
state = "cancelled_after_conflict",
retry,
"Scanner state conflict reconciliation cancelled by the leader fence"
);
return false;
}
if let Some(persisted) = persisted {
if persisted.len() < 8 {
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %&*DATA_USAGE_BLOOM_NAME_PATH,
state = "conflict_state_invalid",
length = persisted.len(),
"Scanner state conflict winner is truncated"
);
return false;
}
let (persisted_cycle, persisted_epoch) = match decode_scanner_cycle_state(&persisted) {
Ok(state) => state,
Err(e) => {
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %&*DATA_USAGE_BLOOM_NAME_PATH,
state = "conflict_state_decode_failed",
error = %e,
"Scanner state conflict winner could not be decoded"
);
return false;
}
};
if persisted_epoch != leader_epoch {
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %&*DATA_USAGE_BLOOM_NAME_PATH,
state = "leader_epoch_fenced",
expected_epoch = leader_epoch,
persisted_epoch,
"Scanner state save rejected by a newer leadership epoch"
);
return false;
}
if persisted_cycle.next >= cycle_info.next {
*cycle_info = persisted_cycle;
debug!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %&*DATA_USAGE_BLOOM_NAME_PATH,
state = "conflict_reconciled",
retry,
"Scanner state adopted the current persisted cycle"
);
return true;
}
}
if retry < SCANNER_PERSIST_CAS_RETRIES {
debug!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %&*DATA_USAGE_BLOOM_NAME_PATH,
state = "conflict_retry",
retry = retry + 1,
"Scanner state CAS conflict will be retried"
);
continue;
}
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %&*DATA_USAGE_BLOOM_NAME_PATH,
state = "conflict_retries_exhausted",
retries = SCANNER_PERSIST_CAS_RETRIES,
"Scanner state CAS conflict retries exhausted"
);
return false;
}
Err(e) => {
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %&*DATA_USAGE_BLOOM_NAME_PATH,
state = "failed",
error = %e,
"Scanner state persistence failed"
);
return false;
}
}
}
false
}
pub(super) async fn finalize_partial_scan_cycle(
ctx: &CancellationToken,
storeapi: Arc<impl ScannerObjectIO>,
cycle_info: &mut CurrentCycle,
revision: &mut DataUsageCacheRevision,
leader_epoch: u64,
cycle_metrics_guard: &mut ScannerCycleMetricsGuard,
) -> bool {
// A budget-limited cycle is deliberate pacing, not a failure. The cycle counter
// must still advance (and persist) because per-bucket next_cycle is stamped from
// it and compacted folders are only rescanned when their hash matches
// next_cycle % DATA_USAGE_UPDATE_DIR_CYCLES; a pinned counter starves lifecycle
// expiry and usage refresh on every folder outside the stuck window.
if let Err(err) = advance_scanner_cycle(cycle_info) {
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
state = "cycle_counter_exhausted",
error = %err,
"Scanner partial cycle could not advance"
);
mark_scan_cycle_idle(cycle_info, cycle_metrics_guard).await;
return false;
}
cycle_info.current = 0;
global_metrics().clear_current_scan_mode();
let persisted = persist_scanner_cycle_state(ctx, storeapi, cycle_info, revision, leader_epoch).await;
cycle_metrics_guard.finish(cycle_info.clone()).await;
persisted
}
pub(super) async fn persist_required_scanner_cycle_floor(
ctx: &CancellationToken,
storeapi: Arc<impl ScannerObjectIO>,
cycle_info: &mut CurrentCycle,
revision: &mut DataUsageCacheRevision,
leader_epoch: u64,
required_cycle: u64,
cycle_metrics_guard: &mut ScannerCycleMetricsGuard,
) -> bool {
if required_cycle <= cycle_info.current || required_cycle == u64::MAX {
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
current_cycle = cycle_info.current,
required_cycle,
state = "invalid_cache_cycle_floor",
"Scanner cache cycle floor is invalid"
);
mark_scan_cycle_idle(cycle_info, cycle_metrics_guard).await;
return false;
}
cycle_info.next = cycle_info.next.max(required_cycle);
cycle_info.current = 0;
global_metrics().clear_current_scan_mode();
let persisted = persist_scanner_cycle_state(ctx, storeapi, cycle_info, revision, leader_epoch).await;
cycle_metrics_guard.finish(cycle_info.clone()).await;
persisted
}
pub(super) async fn await_scanner_cycle_with_lock_fence<Cycle, LockLost>(
cycle_ctx: &CancellationToken,
cycle: Cycle,
lock_lost: LockLost,
) -> Option<Cycle::Output>
where
Cycle: Future,
LockLost: Future<Output = ()>,
{
tokio::pin!(cycle);
tokio::pin!(lock_lost);
tokio::select! {
biased;
_ = &mut lock_lost => {
cycle_ctx.cancel();
tokio::time::timeout(SCANNER_LOCK_LOSS_SHUTDOWN_TIMEOUT, &mut cycle).await.ok()
}
output = &mut cycle => Some(output),
}
}
crates/scanner/src/scanner/heal_info.rs
+109
View File
@@ -0,0 +1,109 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
/// The background-heal info object persisted between scanner cycles.
use super::*;
/// Background healing information
#[derive(Clone, Debug, Default, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct BackgroundHealInfo {
/// Bitrot scan start time
pub bitrot_start_time: Option<DateTime<Utc>>,
/// Bitrot scan start cycle
pub bitrot_start_cycle: u64,
/// Current scan mode
pub current_scan_mode: HealScanMode,
}
/// Read background healing information from storage
pub async fn read_background_heal_info(storeapi: Arc<ECStore>) -> BackgroundHealInfo {
// Skip for ErasureSD setup
if scanner_is_erasure_sd().await {
return BackgroundHealInfo::default();
}
// Get last healing information
match read_config(storeapi, &BACKGROUND_HEAL_INFO_PATH).await {
Ok(buf) => serde_json::from_slice::<BackgroundHealInfo>(&buf).unwrap_or_else(|e| {
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_BACKGROUND_HEAL_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_BACKGROUND_HEAL,
path = %&*BACKGROUND_HEAL_INFO_PATH,
state = "decode_failed",
error = %e,
"Scanner background heal decode failed"
);
BackgroundHealInfo::default()
}),
Err(e) => {
// Only log if it's not a ConfigNotFound error
if e != EcstoreError::ConfigNotFound {
warn!(
target: "rustfs::scanner",
event = EVENT_SCANNER_BACKGROUND_HEAL_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_BACKGROUND_HEAL,
path = %&*BACKGROUND_HEAL_INFO_PATH,
state = "read_failed",
error = %e,
"Scanner background heal read failed"
);
}
BackgroundHealInfo::default()
}
}
}
/// Save background healing information to storage
#[instrument(skip(storeapi))]
pub async fn save_background_heal_info(storeapi: Arc<ECStore>, info: BackgroundHealInfo) {
// Skip for ErasureSD setup
if scanner_is_erasure_sd().await {
return;
}
// Serialize to JSON
let data = match serde_json::to_vec(&info) {
Ok(data) => data,
Err(e) => {
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_BACKGROUND_HEAL_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_BACKGROUND_HEAL,
path = %&*BACKGROUND_HEAL_INFO_PATH,
state = "encode_failed",
error = %e,
"Scanner background heal encode failed"
);
return;
}
};
// Save configuration
if let Err(e) = save_config(storeapi, &BACKGROUND_HEAL_INFO_PATH, data).await {
warn!(
target: "rustfs::scanner",
event = EVENT_SCANNER_BACKGROUND_HEAL_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_BACKGROUND_HEAL,
path = %&*BACKGROUND_HEAL_INFO_PATH,
state = "save_failed",
error = %e,
"Scanner background heal save failed"
);
}
}
crates/scanner/src/scanner/leadership.rs
+363
View File
@@ -0,0 +1,363 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
/// Leader-lock claiming, usage-epoch fencing, and lock-loss handling.
use super::*;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum ScannerLeadershipClaimReconcile {
Durable,
Changed,
Unchanged,
}
pub(super) async fn reconcile_scanner_leadership_claim(
storeapi: Arc<impl ScannerObjectIO>,
attempted: &[u8],
previous_revision: &DataUsageCacheRevision,
claimed_epoch: u64,
cycle_info: &mut CurrentCycle,
revision: &mut DataUsageCacheRevision,
persisted_epoch: &mut u64,
) -> Result<ScannerLeadershipClaimReconcile, ScannerError> {
let (persisted, persisted_revision) = read_config_with_revision(storeapi, DATA_USAGE_BLOOM_NAME_PATH.as_str())
.await
.map_err(|err| ScannerError::Other(format!("failed to reconcile scanner leadership claim: {err}")))?;
let revision_changed = &persisted_revision != previous_revision;
*revision = persisted_revision;
let Some(persisted) = persisted else {
*cycle_info = CurrentCycle::default();
return Ok(if revision_changed {
ScannerLeadershipClaimReconcile::Changed
} else {
ScannerLeadershipClaimReconcile::Unchanged
});
};
if persisted == attempted {
*persisted_epoch = claimed_epoch;
return Ok(ScannerLeadershipClaimReconcile::Durable);
}
let (current, epoch) = decode_scanner_cycle_state(&persisted)
.map_err(|err| ScannerError::Other(format!("scanner leadership conflict winner is invalid: {err}")))?;
*cycle_info = current;
*persisted_epoch = (*persisted_epoch).max(epoch);
Ok(if revision_changed {
ScannerLeadershipClaimReconcile::Changed
} else {
ScannerLeadershipClaimReconcile::Unchanged
})
}
pub(super) fn decode_usage_snapshot_for_epoch_fence(data: &[u8], path: &str) -> Result<DataUsageInfo, ScannerError> {
serde_json::from_slice(data)
.map_err(|err| ScannerError::Other(format!("failed to decode scanner usage epoch fence from {path}: {err}")))
}
pub(super) async fn usage_snapshot_for_epoch_fence(
storeapi: Arc<impl ScannerObjectIO>,
primary: Option<&[u8]>,
) -> Result<DataUsageInfo, ScannerError> {
if let Some(primary) = primary {
return decode_usage_snapshot_for_epoch_fence(primary, DATA_USAGE_OBJ_NAME_PATH.as_str());
}
let backup_path = format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str());
let (backup, _) = read_config_with_revision(storeapi.clone(), &backup_path)
.await
.map_err(|err| ScannerError::Other(format!("failed to read scanner usage epoch fence backup: {err}")))?;
if let Some(backup) = backup.as_deref() {
return decode_usage_snapshot_for_epoch_fence(backup, &backup_path);
}
for path in [
LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str().to_string(),
format!("{}.bkp", LEGACY_DATA_USAGE_OBJ_NAME_PATH.as_str()),
] {
let (legacy, _) = read_config_with_revision(storeapi.clone(), &path)
.await
.map_err(|err| ScannerError::Other(format!("failed to read legacy scanner usage epoch fence: {err}")))?;
if let Some(legacy) = legacy.as_deref() {
return decode_usage_snapshot_for_epoch_fence(legacy, &path);
}
}
Ok(DataUsageInfo::default())
}
pub(super) async fn fence_scanner_usage_epoch(
ctx: &CancellationToken,
storeapi: Arc<impl ScannerObjectIO>,
claimed_epoch: u64,
) -> Result<(), ScannerError> {
for retry in 0..=SCANNER_PERSIST_CAS_RETRIES {
if ctx.is_cancelled() {
return Err(ScannerError::Other("scanner leadership was cancelled before usage fencing".to_string()));
}
let (primary, revision) = read_config_with_revision(storeapi.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
.await
.map_err(|err| ScannerError::Other(format!("failed to read scanner usage epoch fence: {err}")))?;
let mut usage = usage_snapshot_for_epoch_fence(storeapi.clone(), primary.as_deref()).await?;
match usage.scanner_epoch {
Some(epoch) if epoch > claimed_epoch => {
return Err(ScannerError::Other(format!(
"scanner usage epoch fence lost to newer leader: claimed={claimed_epoch}, persisted={epoch}"
)));
}
Some(epoch) if epoch == claimed_epoch => return Ok(()),
Some(_) | None => {}
}
usage.scanner_epoch = Some(claimed_epoch);
let data = serde_json::to_vec(&usage)
.map_err(|err| ScannerError::Other(format!("failed to encode scanner usage epoch fence: {err}")))?;
let save_result =
save_config_with_preconditions(storeapi.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str(), data, revision.preconditions())
.await;
if save_result
.as_ref()
.ok()
.and_then(|object_info| object_info.etag.as_deref())
.is_some_and(|etag| !etag.is_empty())
{
return Ok(());
}
let (persisted, persisted_revision) = read_config_with_revision(storeapi.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
.await
.map_err(|err| ScannerError::Other(format!("failed to reconcile scanner usage epoch fence: {err}")))?;
if let Some(persisted) = persisted {
let persisted = decode_usage_snapshot_for_epoch_fence(&persisted, DATA_USAGE_OBJ_NAME_PATH.as_str())?;
match persisted.scanner_epoch {
Some(epoch) if epoch == claimed_epoch => return Ok(()),
Some(epoch) if epoch > claimed_epoch => {
return Err(ScannerError::Other(format!(
"scanner usage epoch fence lost to newer leader: claimed={claimed_epoch}, persisted={epoch}"
)));
}
Some(_) | None => {}
}
}
let precondition_failed = matches!(save_result, Err(EcstoreError::PreconditionFailed));
if retry < SCANNER_PERSIST_CAS_RETRIES && (precondition_failed || persisted_revision != revision) {
continue;
}
return Err(ScannerError::Other(match save_result {
Ok(_) => "scanner usage epoch fence returned no ETag and could not be confirmed".to_string(),
Err(err) => format!("scanner usage epoch fence save failed: {err}"),
}));
}
Err(ScannerError::Other("scanner usage epoch fence retries exhausted".to_string()))
}
pub(super) async fn complete_scanner_leadership_claim(
ctx: &CancellationToken,
storeapi: Arc<impl ScannerObjectIO>,
claimed_epoch: u64,
) -> bool {
if let Err(err) = fence_scanner_usage_epoch(ctx, storeapi, claimed_epoch).await {
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %DATA_USAGE_OBJ_NAME_PATH.as_str(),
state = "usage_epoch_fence_failed",
claimed_epoch,
error = %err,
"Scanner leadership usage epoch fencing failed"
);
return false;
}
!ctx.is_cancelled()
}
pub(super) async fn claim_scanner_leadership(
ctx: &CancellationToken,
storeapi: Arc<impl ScannerObjectIO>,
cycle_info: &mut CurrentCycle,
revision: &mut DataUsageCacheRevision,
persisted_epoch: &mut u64,
) -> bool {
for retry in 0..=SCANNER_PERSIST_CAS_RETRIES {
if ctx.is_cancelled() {
return false;
}
let Some(claimed_epoch) = persisted_epoch.checked_add(1) else {
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %&*DATA_USAGE_BLOOM_NAME_PATH,
state = "leader_epoch_exhausted",
"Scanner leadership epoch is exhausted"
);
return false;
};
let data = match encode_scanner_cycle_state(cycle_info, claimed_epoch) {
Ok(data) => data,
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 = "leader_claim_encode_failed",
error = %err,
"Scanner leadership claim encoding failed"
);
return false;
}
};
let previous_revision = revision.clone();
let save_result =
save_config_with_preconditions(storeapi.clone(), &DATA_USAGE_BLOOM_NAME_PATH, data.clone(), revision.preconditions())
.await;
match save_result {
Ok(object_info) => {
if let Some(etag) = object_info.etag.filter(|etag| !etag.is_empty()) {
*revision = DataUsageCacheRevision::Etag(etag);
*persisted_epoch = claimed_epoch;
return complete_scanner_leadership_claim(ctx, storeapi, claimed_epoch).await;
}
match reconcile_scanner_leadership_claim(
storeapi.clone(),
&data,
&previous_revision,
claimed_epoch,
cycle_info,
revision,
persisted_epoch,
)
.await
{
Ok(ScannerLeadershipClaimReconcile::Durable) => {
return complete_scanner_leadership_claim(ctx, storeapi, claimed_epoch).await;
}
Ok(ScannerLeadershipClaimReconcile::Changed) if retry < SCANNER_PERSIST_CAS_RETRIES => continue,
Ok(ScannerLeadershipClaimReconcile::Changed | ScannerLeadershipClaimReconcile::Unchanged) => {
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %&*DATA_USAGE_BLOOM_NAME_PATH,
state = "leader_claim_missing_revision",
"Scanner leadership claim returned no ETag and could not be confirmed"
);
return false;
}
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 = "leader_claim_reconcile_failed",
error = %err,
"Scanner leadership claim read-back failed"
);
return false;
}
}
}
Err(err) => {
let precondition_failed = matches!(err, EcstoreError::PreconditionFailed);
match reconcile_scanner_leadership_claim(
storeapi.clone(),
&data,
&previous_revision,
claimed_epoch,
cycle_info,
revision,
persisted_epoch,
)
.await
{
Ok(ScannerLeadershipClaimReconcile::Durable) => {
return complete_scanner_leadership_claim(ctx, storeapi, claimed_epoch).await;
}
Ok(ScannerLeadershipClaimReconcile::Changed)
if retry < SCANNER_PERSIST_CAS_RETRIES && !ctx.is_cancelled() =>
{
continue;
}
Ok(ScannerLeadershipClaimReconcile::Unchanged)
if precondition_failed && retry < SCANNER_PERSIST_CAS_RETRIES && !ctx.is_cancelled() =>
{
continue;
}
Ok(ScannerLeadershipClaimReconcile::Changed | ScannerLeadershipClaimReconcile::Unchanged) => {
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %&*DATA_USAGE_BLOOM_NAME_PATH,
state = if precondition_failed {
"leader_claim_conflicts_exhausted"
} else {
"leader_claim_failed"
},
error = %err,
"Scanner leadership claim failed"
);
return false;
}
Err(reconcile_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 = "leader_claim_reload_failed",
error = %reconcile_err,
save_error = %err,
"Scanner leadership claim reconciliation failed"
);
return false;
}
}
}
}
}
false
}
pub(super) async fn record_scanner_leader_lock_lost(message: &'static str) {
reset_scanner_cycle_schedule();
record_scanner_leader_lock_state("lost");
global_metrics()
.record_scanner_leader_liveness("lost", false, "leader lock refresh quorum lost")
.await;
warn!(
target: "rustfs::scanner",
event = EVENT_SCANNER_LOCK_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
lock_name = "leader.lock",
state = "lost",
reason = message,
"Scanner leader lock lost"
);
}
crates/scanner/src/scanner/tests.rs
+3920
View File
File diff suppressed because it is too large Load Diff
crates/scanner/src/scanner/usage_store.rs
+483
View File
@@ -0,0 +1,483 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
/// Data-usage snapshot persistence: CAS store pipeline, epoch baselines, and observed-snapshot cleanup.
use super::*;
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub(super) enum DataUsagePersistOutcome {
#[default]
NoUpdate,
Current,
AlreadyDurable,
PriorCycleDurable,
Saved,
Failed,
}
#[derive(Clone, Debug)]
pub(super) struct DataUsagePersistBaseline {
pub(super) data: Option<Bytes>,
pub(super) revision: DataUsageCacheRevision,
}
#[derive(Debug)]
pub(super) enum DataUsagePersistTaskResult {
Completed(DataUsagePersistOutcome),
Cancelled,
TimedOut,
JoinFailed(tokio::task::JoinError),
}
pub(super) async fn wait_for_data_usage_persist_task(
ctx: &CancellationToken,
task: &mut AbortOnDropHandle<DataUsagePersistOutcome>,
timeout: Duration,
) -> DataUsagePersistTaskResult {
tokio::select! {
biased;
result = &mut *task => match result {
Ok(outcome) => DataUsagePersistTaskResult::Completed(outcome),
Err(err) => DataUsagePersistTaskResult::JoinFailed(err),
},
_ = ctx.cancelled() => {
task.abort();
let _ = (&mut *task).await;
DataUsagePersistTaskResult::Cancelled
},
_ = tokio::time::sleep(timeout) => {
task.abort();
let _ = (&mut *task).await;
DataUsagePersistTaskResult::TimedOut
}
}
}
#[instrument(skip(ctx, storeapi))]
pub async fn store_data_usage_in_backend(
ctx: CancellationToken,
storeapi: Arc<impl ScannerObjectIO + ScannerConfigObjectDelete>,
receiver: mpsc::Receiver<DataUsageInfo>,
) {
let _ = store_data_usage_in_backend_with_outcome(ctx, storeapi, receiver).await;
}
pub(super) async fn store_data_usage_in_backend_with_outcome(
ctx: CancellationToken,
storeapi: Arc<impl ScannerObjectIO + ScannerConfigObjectDelete>,
receiver: mpsc::Receiver<DataUsageInfo>,
) -> DataUsagePersistOutcome {
store_data_usage_in_backend_with_outcome_for_epoch(ctx, storeapi, receiver, None).await
}
pub(super) async fn store_data_usage_in_backend_with_outcome_for_epoch(
ctx: CancellationToken,
storeapi: Arc<impl ScannerObjectIO + ScannerConfigObjectDelete>,
receiver: mpsc::Receiver<DataUsageInfo>,
leader_epoch: Option<u64>,
) -> DataUsagePersistOutcome {
store_data_usage_in_backend_with_outcome_for_epoch_and_baseline(ctx, storeapi, receiver, leader_epoch, None).await
}
pub(super) async fn store_data_usage_in_backend_with_outcome_for_epoch_and_baseline(
ctx: CancellationToken,
storeapi: Arc<impl ScannerObjectIO + ScannerConfigObjectDelete>,
mut receiver: mpsc::Receiver<DataUsageInfo>,
leader_epoch: Option<u64>,
initial_baseline: Option<DataUsagePersistBaseline>,
) -> DataUsagePersistOutcome {
let mut outcome = DataUsagePersistOutcome::NoUpdate;
let mut next_baseline = initial_baseline;
'updates: while let Some(mut data_usage_info) = receiver.recv().await {
let _activity_guard = ScannerActivityGuard::new();
if ctx.is_cancelled() {
break;
}
if let Some(leader_epoch) = leader_epoch {
data_usage_info.scanner_epoch = Some(leader_epoch);
}
let observational = data_usage_info.usage_snapshot_converged == Some(false);
let target_path = if observational {
DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str()
} else {
DATA_USAGE_OBJ_NAME_PATH.as_str()
};
if observational && data_usage_info.usage_snapshot_authoritative_baseline.is_none() {
let authoritative_data = match next_baseline.as_ref() {
Some(baseline) => baseline.data.clone(),
None => match read_config_with_revision(storeapi.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str()).await {
Ok((data, _)) => data.map(Bytes::from),
Err(err) => {
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %DATA_USAGE_OBJ_NAME_PATH.as_str(),
state = "observed_baseline_load_failed",
error = %err,
"Scanner could not identify the authoritative baseline for an observation"
);
outcome = DataUsagePersistOutcome::Failed;
continue;
}
},
};
let authoritative = match authoritative_data.as_deref() {
Some(data) => match serde_json::from_slice::<DataUsageInfo>(data) {
Ok(info) => info,
Err(err) => {
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %DATA_USAGE_OBJ_NAME_PATH.as_str(),
state = "observed_baseline_decode_failed",
error = %err,
"Scanner refused to publish an observation from an invalid authoritative baseline"
);
outcome = DataUsagePersistOutcome::Failed;
continue;
}
},
None => DataUsageInfo::default(),
};
data_usage_info.usage_snapshot_authoritative_baseline = Some(authoritative.snapshot_identity());
}
if !data_usage_info.is_complete_bucket_usage_snapshot() {
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %target_path,
state = "reject_incomplete_snapshot",
"Scanner refused to persist an incomplete data usage snapshot"
);
global_metrics().record_scanner_usage_save_result(ScannerUsageSaveResult::Failed);
outcome = DataUsagePersistOutcome::Failed;
continue;
}
let data = match serde_json::to_vec(&data_usage_info) {
Ok(data) => data,
Err(e) => {
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %target_path,
state = "encode_failed",
error = %e,
"Scanner data usage encode failed"
);
global_metrics().record_scanner_usage_save_result(ScannerUsageSaveResult::EncodeFailed);
outcome = DataUsagePersistOutcome::Failed;
continue;
}
};
let sha256hex = (!data.is_empty()).then(|| hex_simd::encode_to_string(Sha256::digest(&data), hex_simd::AsciiCase::Lower));
let data = Bytes::from(data);
let backup_due = !observational && data_usage_backup_due(&data_usage_info);
let mut cas_retry = 0usize;
let save_outcome = loop {
if ctx.is_cancelled() {
break 'updates;
}
let baseline = if !observational && cas_retry == 0 {
next_baseline.take()
} else {
None
};
let (existing_data, revision) = match baseline {
Some(baseline) => (baseline.data, baseline.revision),
None => match read_config_with_revision(storeapi.clone(), target_path).await {
Ok((data, revision)) => (data.map(Bytes::from), revision),
Err(e) => {
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %target_path,
state = "revision_load_failed",
error = %e,
"Scanner data usage revision load failed"
);
break DataUsagePersistOutcome::Failed;
}
},
};
let existing = existing_data
.as_deref()
.and_then(|buf| serde_json::from_slice::<DataUsageInfo>(buf).ok());
if cas_retry > 0 && data_usage_reintroduces_missing_bucket(&data_usage_info, existing.as_ref()) {
debug!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %target_path,
incoming_scanner_epoch = ?data_usage_info.scanner_epoch,
incoming_scanner_cycle = ?data_usage_info.scanner_cycle,
state = "skip_deleted_bucket_reintroduction",
"Scanner usage update skipped after a concurrent bucket removal"
);
break DataUsagePersistOutcome::Current;
}
if let Some(existing) = existing.as_ref() {
if existing == &data_usage_info {
break DataUsagePersistOutcome::AlreadyDurable;
}
if existing.scanner_epoch.is_some()
&& existing.scanner_epoch == data_usage_info.scanner_epoch
&& existing.scanner_cycle.is_some()
&& existing.scanner_cycle == data_usage_info.scanner_cycle
{
break DataUsagePersistOutcome::PriorCycleDurable;
}
if let Some(reason) = stale_data_usage_update_reason(&data_usage_info, existing, std::time::SystemTime::now()) {
debug!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %target_path,
incoming_scanner_epoch = ?data_usage_info.scanner_epoch,
existing_scanner_epoch = ?existing.scanner_epoch,
incoming_scanner_cycle = ?data_usage_info.scanner_cycle,
existing_scanner_cycle = ?existing.scanner_cycle,
incoming_last_update = ?data_usage_info.last_update,
existing_last_update = ?existing.last_update,
reason = reason,
state = "skip_stale_update",
"Scanner stale data usage update skipped"
);
break DataUsagePersistOutcome::Current;
}
}
if ctx.is_cancelled() {
break 'updates;
}
let done_save = Metrics::time(Metric::SaveUsage);
let save_result = save_config_shared_with_preconditions(
storeapi.clone(),
target_path,
data.clone(),
sha256hex.clone(),
revision.preconditions(),
)
.await;
done_save();
match save_result {
Ok(object_info) => {
if !observational {
next_baseline = object_info
.etag
.filter(|etag| !etag.is_empty())
.map(|etag| DataUsagePersistBaseline {
data: Some(data.clone()),
revision: DataUsageCacheRevision::Etag(etag),
});
}
break DataUsagePersistOutcome::Saved;
}
Err(EcstoreError::PreconditionFailed) if cas_retry < SCANNER_PERSIST_CAS_RETRIES => {
cas_retry += 1;
debug!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %target_path,
state = "conflict_retry",
retry = cas_retry,
"Scanner data usage CAS conflict will be reconciled"
);
}
Err(e) => {
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %target_path,
state = if matches!(e, EcstoreError::PreconditionFailed) {
"conflict_retries_exhausted"
} else {
"save_failed"
},
error = %e,
"Scanner data usage save failed"
);
break DataUsagePersistOutcome::Failed;
}
}
};
match save_outcome {
DataUsagePersistOutcome::Current => {
if observational {
invalidate_admin_data_usage_snapshot_cache().await;
} else {
invalidate_data_usage_snapshot_cache().await;
}
global_metrics().record_scanner_usage_save_result(ScannerUsageSaveResult::SkippedStale);
outcome = DataUsagePersistOutcome::Current;
continue;
}
DataUsagePersistOutcome::AlreadyDurable => {
if observational {
invalidate_admin_data_usage_snapshot_cache().await;
} else {
cleanup_observed_data_usage_snapshot(storeapi.clone(), &data_usage_info).await;
invalidate_data_usage_snapshot_cache().await;
replace_bucket_usage_memory_from_info(&data_usage_info).await;
}
global_metrics().record_scanner_usage_save_result(ScannerUsageSaveResult::Success);
outcome = DataUsagePersistOutcome::AlreadyDurable;
}
DataUsagePersistOutcome::PriorCycleDurable => {
if observational {
invalidate_admin_data_usage_snapshot_cache().await;
} else {
cleanup_observed_data_usage_snapshot(storeapi.clone(), &data_usage_info).await;
invalidate_data_usage_snapshot_cache().await;
}
global_metrics().record_scanner_usage_save_result(ScannerUsageSaveResult::Success);
outcome = DataUsagePersistOutcome::PriorCycleDurable;
}
DataUsagePersistOutcome::Failed | DataUsagePersistOutcome::NoUpdate => {
global_metrics().record_scanner_usage_save_result(ScannerUsageSaveResult::Failed);
outcome = DataUsagePersistOutcome::Failed;
continue;
}
DataUsagePersistOutcome::Saved => {
if observational {
invalidate_admin_data_usage_snapshot_cache().await;
} else {
cleanup_observed_data_usage_snapshot(storeapi.clone(), &data_usage_info).await;
invalidate_data_usage_snapshot_cache().await;
replace_bucket_usage_memory_from_info(&data_usage_info).await;
}
global_metrics().record_scanner_usage_save_result(ScannerUsageSaveResult::Success);
outcome = DataUsagePersistOutcome::Saved;
}
}
if backup_due {
let done_save = Metrics::time(Metric::SaveUsage);
if let Err(e) = sync_data_usage_backup_from_primary(&ctx, storeapi.clone()).await {
warn!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str()),
state = "backup_save_failed",
error = %e,
"Scanner data usage backup save failed"
);
}
done_save();
}
}
outcome
}
pub(super) async fn cleanup_observed_data_usage_snapshot(
storeapi: Arc<impl ScannerObjectIO + ScannerConfigObjectDelete>,
authoritative: &DataUsageInfo,
) {
let (observed_data, revision) =
match read_config_with_revision(storeapi.clone(), DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str()).await {
Ok((Some(data), revision)) => (data, revision),
Ok((None, _)) => return,
Err(err) => {
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str(),
state = "observed_cleanup_read_failed",
error = %err,
"Scanner could not inspect observational data usage snapshot before authoritative cleanup"
);
return;
}
};
let observed = match serde_json::from_slice::<DataUsageInfo>(&observed_data) {
Ok(observed) => observed,
Err(err) => {
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str(),
state = "observed_cleanup_decode_failed",
error = %err,
"Scanner refused to remove an invalid observational data usage snapshot after authoritative save"
);
return;
}
};
if observed_data_usage_is_newer(&observed, authoritative) {
return;
}
let result = storeapi
.delete_config_object(
RUSTFS_META_BUCKET,
DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str(),
ScannerObjectOptions {
delete_prefix: true,
delete_prefix_object: true,
http_preconditions: Some(revision.preconditions()),
..Default::default()
},
)
.await;
match result {
Ok(_)
| Err(
EcstoreError::FileNotFound
| EcstoreError::ConfigNotFound
| EcstoreError::ObjectNotFound(_, _)
| EcstoreError::PreconditionFailed,
) => {}
Err(err) => {
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %DATA_USAGE_OBSERVED_OBJ_NAME_PATH.as_str(),
state = "observed_cleanup_failed",
error = %err,
"Scanner could not remove stale observational data usage snapshot after authoritative save"
);
}
}
}