gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-06 03:59:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main' into houseme/chore/scanner-heal-v2-b3-base

Browse Source
This commit is contained in:
houseme
2026-09-05 16:43:37 +08:00
parent ef7e7afd8c cf9688898d
commit cee35f7e54
14 changed files with 1152 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/heal/src/heal/manager.rs
+60 -12
View File
@@ -45,6 +45,11 @@ use tracing::{debug, error, info, warn};
use super::{DiskError, Endpoint, HealDiskExt as _, local_disk_map_read};
const KEEP_HEAL_TASK_STATUS_DURATION: Duration = Duration::from_secs(10 * 60);
// Each cache includes alias tokens in its count and byte budget. Eviction
// removes every token sharing a snapshot; neither cache retains repair state.
const MAX_COMPLETED_HEAL_TOKENS: usize = 1024;
const MAX_COMPLETED_HEAL_BYTES: usize = 64 * 1024 * 1024;
const MAX_COMPLETED_HEAL_RESULT_BYTES: usize = 1024 * 1024;
const DISPLACED_HEAL_REASON: &str = "reason=displaced; retry_hint=submit_again";
const LOG_COMPONENT_HEAL: &str = "heal";
const LOG_SUBSYSTEM_DISK_SCANNER: &str = "disk_scanner";
@@ -180,6 +185,8 @@ fn record_displaced_terminal(
request: &HealRequest,
) -> Arc<CompletedHealStatus> {
let terminal = Arc::new(CompletedHealStatus {
progress: None,
retained_bytes: std::sync::OnceLock::new(),
heal_type: request.heal_type.clone(),
status: HealTaskStatus::Failed {
error: format!("heal task displaced by a higher-priority request ({DISPLACED_HEAL_REASON})"),
@@ -193,6 +200,7 @@ fn record_displaced_terminal(
let mut terminals = lock_displaced_terminals(registry);
prune_completed_heal_statuses(&mut terminals);
terminals.insert(request.id.clone(), Arc::clone(&terminal));
prune_completed_heal_statuses(&mut terminals);
terminal
}
@@ -209,9 +217,15 @@ async fn remove_displaced_task_aliases(
.collect::<Vec<_>>();
let mut displaced_terminals = lock_displaced_terminals(terminals);
prune_completed_heal_statuses(&mut displaced_terminals);
for alias_id in alias_ids {
displaced_terminals.insert(alias_id, Arc::clone(terminal));
if displaced_terminals
.get(task_id)
.is_some_and(|current| Arc::ptr_eq(current, terminal))
{
for alias_id in alias_ids {
displaced_terminals.insert(alias_id, Arc::clone(terminal));
}
}
prune_completed_heal_statuses(&mut displaced_terminals);
aliases.retain(|alias_id, alias| alias_id != task_id && alias.task_id != task_id);
}
@@ -222,6 +236,36 @@ async fn remove_task_aliases_for_task(registry: &Arc<Mutex<HashMap<String, HealT
.retain(|alias_id, alias| alias_id != task_id && alias.task_id != task_id);
}
// Callers hold active ownership until publication. Lock order is active ->
// retrying (when needed) -> aliases -> completed; queries release aliases
// before looking up active state. Publishing aliases before removing their
// mapping keeps both an already-resolved token and a new lookup valid.
async fn publish_completed_heal(
completed_heals: &Mutex<HashMap<String, Arc<CompletedHealStatus>>>,
task_aliases: &Mutex<HashMap<String, HealTaskAlias>>,
task_id: &str,
completed: CompletedHealStatus,
terminal: bool,
) {
let completed = Arc::new(completed);
completed.retained_bytes();
let mut aliases = task_aliases.lock().await;
let mut retained = completed_heals.lock().await;
if let Some(previous) = retained.get(task_id).cloned() {
for entry in retained.values_mut().filter(|entry| Arc::ptr_eq(entry, &previous)) {
*entry = Arc::clone(&completed);
}
}
retained.insert(task_id.to_owned(), Arc::clone(&completed));
if terminal {
for (alias_id, _) in aliases.iter().filter(|(_, alias)| alias.task_id == task_id) {
retained.insert(alias_id.clone(), Arc::clone(&completed));
}
aliases.retain(|alias_id, alias| alias_id != task_id && alias.task_id != task_id);
}
prune_completed_heal_statuses(&mut retained);
}
#[derive(Debug, Clone)]
pub struct HealTaskReport {
pub status: HealTaskStatus,
@@ -268,7 +312,7 @@ fn completed_task_report(completed: &CompletedHealStatus, since: Option<u64>) ->
let result_items = match since {
None => completed.seqed_items.iter().map(|(_, item)| item.clone()).collect(),
Some(cursor) => {
if cursor + 1 < completed.min_seq {
if cursor.saturating_add(1) < completed.min_seq {
lagged = true;
}
completed
@@ -283,7 +327,7 @@ fn completed_task_report(completed: &CompletedHealStatus, since: Option<u64>) ->
status: completed.status.clone(),
result_items,
result_items_truncated: completed.result_items_truncated || lagged,
progress: None,
progress: completed.progress.clone(),
next_seq: completed.next_seq,
min_seq: completed.min_seq,
}
@@ -1847,14 +1891,14 @@ impl HealManager {
pub async fn get_task_progress(&self, task_id: &str) -> Result<HealProgress> {
let canonical_task_id = self.canonical_task_id(task_id).await;
let active_heals = self.active_heals.lock().await;
if let Some(task) = active_heals.get(&canonical_task_id) {
Ok(task.get_progress().await)
} else {
Err(Error::TaskNotFound {
task_id: task_id.to_string(),
})
}
let progress = match self.lookup_task_state(&canonical_task_id, None).await {
TaskStateLookup::Active(task) => Some(task.get_progress().await),
TaskStateLookup::Completed(completed) => completed.progress.clone(),
_ => None,
};
progress.ok_or_else(|| Error::TaskNotFound {
task_id: task_id.to_string(),
})
}
/// Cancel task
@@ -1864,6 +1908,8 @@ impl HealManager {
let mut active_heals = self.active_heals.lock().await;
if let Some(task) = active_heals.get(&canonical_task_id) {
task.cancel().await?;
let completed = CompletedHealStatus::snapshot(task, HealTaskStatus::Cancelled).await;
publish_completed_heal(&self.completed_heals, &self.task_aliases, &canonical_task_id, completed, true).await;
active_heals.remove(&canonical_task_id);
publish_active_heal_count(&active_heals);
info!(
@@ -1940,6 +1986,8 @@ impl HealManager {
for task_id in &task_ids {
if let Some(task) = active_heals.get(task_id) {
task.cancel().await?;
let completed = CompletedHealStatus::snapshot(task, HealTaskStatus::Cancelled).await;
publish_completed_heal(&self.completed_heals, &self.task_aliases, task_id, completed, true).await;
}
active_heals.remove(task_id);
cancelled += 1;
crates/heal/src/heal/manager/queue.rs
+129
View File
@@ -82,6 +82,8 @@ pub(super) enum QueuePushOutcome {
pub(super) struct CompletedHealStatus {
pub(super) heal_type: HealType,
pub(super) status: HealTaskStatus,
pub(super) progress: Option<HealProgress>,
pub(super) retained_bytes: std::sync::OnceLock<usize>,
pub(super) result_items_truncated: bool,
pub(super) completed_at: SystemTime,
/// Sequence-stamped retained window, archived with the completion so
@@ -92,6 +94,133 @@ pub(super) struct CompletedHealStatus {
pub(super) min_seq: u64,
}
impl CompletedHealStatus {
// Account for owned capacities, including nested drive arrays. Aliases
// conservatively charge the shared allocation again, keeping both token
// count and retained payload bounded without a second ownership index.
pub(super) fn retained_bytes(&self) -> usize {
*self.retained_bytes.get_or_init(|| self.measure_retained_bytes())
}
fn measure_retained_bytes(&self) -> usize {
let mut bytes = size_of::<Self>();
let mut add = |amount: usize| bytes = bytes.saturating_add(amount);
match &self.heal_type {
HealType::Cluster => {}
HealType::Bucket { bucket } => add(bucket.capacity()),
HealType::Object {
bucket,
object,
version_id,
}
| HealType::ECDecode {
bucket,
object,
version_id,
} => {
add(bucket.capacity());
add(object.capacity());
add(version_id.as_ref().map_or(0, String::capacity));
}
HealType::Prefix { bucket, prefix } => {
add(bucket.capacity());
add(prefix.capacity());
}
HealType::Metadata { bucket, object } => {
add(bucket.capacity());
add(object.capacity());
}
HealType::ErasureSet { buckets, set_disk_id } => {
add(buckets.capacity().saturating_mul(size_of::<String>()));
for bucket in buckets {
add(bucket.capacity());
}
add(set_disk_id.capacity());
}
}
if let HealTaskStatus::Failed { error } | HealTaskStatus::Retrying { error, .. } = &self.status {
add(error.capacity());
}
add(self
.progress
.as_ref()
.and_then(|progress| progress.current_object.as_ref())
.map_or(0, String::capacity));
add(self.seqed_items.capacity().saturating_mul(size_of::<(u64, HealResultItem)>()));
for (_, item) in &self.seqed_items {
add(Self::result_item_heap_bytes(item));
}
bytes
}
fn result_item_heap_bytes(item: &HealResultItem) -> usize {
let mut bytes = 0usize;
let mut add = |amount: usize| bytes = bytes.saturating_add(amount);
for value in [
&item.heal_item_type,
&item.bucket,
&item.object,
&item.version_id,
&item.detail,
] {
add(value.capacity());
}
for infos in [&item.before, &item.after] {
add(infos
.drives
.capacity()
.saturating_mul(size_of::<rustfs_madmin::heal_commands::HealDriveInfo>()));
for drive in &infos.drives {
add(drive.uuid.capacity());
add(drive.endpoint.capacity());
add(drive.state.capacity());
}
}
bytes
}
pub(super) fn bound_result_window(&mut self) {
let mut bytes = 0usize;
let retained = self
.seqed_items
.iter()
.rev()
.take_while(|(_, item)| {
bytes = bytes
.saturating_add(size_of::<(u64, HealResultItem)>())
.saturating_add(Self::result_item_heap_bytes(item));
bytes <= MAX_COMPLETED_HEAL_RESULT_BYTES
})
.count();
let truncated = retained < self.seqed_items.len();
if truncated {
self.seqed_items.drain(..self.seqed_items.len() - retained);
self.seqed_items.shrink_to_fit();
self.min_seq = self.seqed_items.first().map_or(self.next_seq, |(seq, _)| *seq);
self.result_items_truncated = true;
self.retained_bytes.take();
}
}
pub(super) async fn snapshot(task: &HealTask, status: HealTaskStatus) -> Self {
let seqed_items = task.get_seqed_result_items().await;
let (next_seq, min_seq) = task.result_seq_cursors();
let mut snapshot = Self {
heal_type: task.heal_type.clone(),
status,
progress: Some(task.get_progress().await),
retained_bytes: std::sync::OnceLock::new(),
result_items_truncated: task.result_items_truncated(),
completed_at: SystemTime::now(),
seqed_items,
next_seq,
min_seq,
};
snapshot.bound_result_window();
snapshot
}
}
#[derive(Debug, Clone)]
pub(super) struct HealTaskAlias {
pub(super) task_id: String,
crates/heal/src/heal/manager/scheduler.rs
+75 -39
View File
@@ -264,7 +264,7 @@ impl HealManager {
error: error.clone(),
retry_attempt: request.retry_attempts,
});
let retry_request_for_queue = retry_request;
let mut retry_request_for_queue = retry_request;
let retry_cancel_token = retry_request_for_queue.as_ref().map(|_| CancellationToken::new());
if retry_request_for_queue.is_none() {
replacement_recovery_anchors_clone
@@ -272,7 +272,35 @@ impl HealManager {
.unwrap_or_else(|poisoned| poisoned.into_inner())
.remove(&task_id);
}
let mut completed_status = match retry_request_for_status {
Some(status) => status,
None => task.get_status().await,
};
let mut completed_status_entry = CompletedHealStatus::snapshot(&task, completed_status.clone()).await;
let completed_progress = task.get_progress().await;
#[cfg(test)]
tests::pause_completed_retention_before_publish(&task_id, &completed_status).await;
let mut active_heals_guard = active_heals_clone.lock().await;
let owns_completion = active_heals_guard.contains_key(&task_id);
let cancelled_completion = if owns_completion {
false
} else {
// Cancellation can win while a finished worker waits
// for active ownership. It must not resurrect a retry
// or replace an acknowledged cancellation with success.
retry_request_for_queue = None;
completed_heals_clone
.lock()
.await
.get(&task_id)
.is_some_and(|completed| completed.status == HealTaskStatus::Cancelled)
};
if cancelled_completion {
completed_status = HealTaskStatus::Cancelled;
completed_status_entry.status = HealTaskStatus::Cancelled;
}
let terminal_completion = !matches!(completed_status, HealTaskStatus::Retrying { .. });
let successful_completion = matches!(completed_status, HealTaskStatus::Completed);
// Keep retry ownership continuous: status snapshots acquire
// these locks in the same active -> retrying order.
let mut retrying_heals_guard = if let (Some((request, _, error)), Some(cancel_token)) =
@@ -295,6 +323,16 @@ impl HealManager {
} else {
None
};
if owns_completion || cancelled_completion {
publish_completed_heal(
&completed_heals_clone,
&task_aliases_clone,
&task_id,
completed_status_entry,
terminal_completion,
)
.await;
}
let completed_task = active_heals_guard.remove(&task_id);
if let Some(completed_task) = completed_task.as_ref() {
publish_active_heal_count(&active_heals_guard);
@@ -304,33 +342,10 @@ impl HealManager {
drop(retrying_heals_guard.take());
drop(active_heals_guard);
if let Some(completed_task) = completed_task {
let completed_status = if let Some(status) = retry_request_for_status {
status
} else {
completed_task.get_status().await
};
let terminal_completion = !matches!(completed_status, HealTaskStatus::Retrying { .. });
let successful_completion = matches!(completed_status, HealTaskStatus::Completed);
let completed_progress = completed_task.get_progress().await;
// Single snapshot of the retained window: the task is
// finished and already off the active map, so there is
// no concurrent writer to race with.
let seqed_items = completed_task.get_seqed_result_items().await;
let (next_seq, min_seq) = completed_task.result_seq_cursors();
let completed_status_entry = CompletedHealStatus {
heal_type: completed_task.heal_type.clone(),
status: completed_status.clone(),
result_items_truncated: completed_task.result_items_truncated(),
completed_at: SystemTime::now(),
seqed_items,
next_seq,
min_seq,
};
let mut completed_heals_guard = completed_heals_clone.lock().await;
prune_completed_heal_statuses(&mut completed_heals_guard);
completed_heals_guard.insert(task_id.clone(), Arc::new(completed_status_entry));
drop(completed_heals_guard);
#[cfg(test)]
tests::pause_completed_retention_handoff(&task_id).await;
if completed_task.is_some() {
// update statistics
let mut stats = statistics_clone.write().await;
match completed_status {
@@ -352,10 +367,6 @@ impl HealManager {
} else {
release_mrf_repair_notice_targets(notice_targets);
}
task_aliases_clone
.lock()
.await
.retain(|alias_id, alias| alias_id != &task_id && alias.task_id != task_id);
}
}
@@ -718,17 +729,42 @@ pub(super) fn heal_request_set_key_for_task(task: &HealTask) -> Option<String> {
}
pub(super) fn prune_completed_heal_statuses(completed_heals: &mut HashMap<String, Arc<CompletedHealStatus>>) {
let Ok(now) = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH) else {
return;
};
prune_completed_heal_statuses_at(completed_heals, SystemTime::now());
}
pub(super) fn prune_completed_heal_statuses_at(completed_heals: &mut HashMap<String, Arc<CompletedHealStatus>>, now: SystemTime) {
completed_heals.retain(|_, completed| {
completed
.completed_at
.duration_since(SystemTime::UNIX_EPOCH)
.map(|completed_at| now.saturating_sub(completed_at) <= KEEP_HEAL_TASK_STATUS_DURATION)
now.duration_since(completed.completed_at)
.map(|age| age <= KEEP_HEAL_TASK_STATUS_DURATION)
.unwrap_or(false)
});
let entry_bytes = |key: &String, value: &Arc<CompletedHealStatus>| {
key.capacity()
.saturating_add(size_of::<(String, Arc<CompletedHealStatus>)>())
.saturating_add(value.retained_bytes())
};
let mut bytes = completed_heals
.iter()
.fold(0usize, |total, (key, value)| total.saturating_add(entry_bytes(key, value)));
while completed_heals.len() > MAX_COMPLETED_HEAL_TOKENS || bytes > MAX_COMPLETED_HEAL_BYTES {
let Some(oldest) = completed_heals
.iter()
.min_by(|(left_id, left), (right_id, right)| {
left.completed_at.cmp(&right.completed_at).then_with(|| left_id.cmp(right_id))
})
.map(|(_, value)| Arc::clone(value))
else {
break;
};
completed_heals.retain(|key, value| {
if Arc::ptr_eq(value, &oldest) {
bytes = bytes.saturating_sub(entry_bytes(key, value));
false
} else {
true
}
});
}
}
pub(super) fn can_schedule_request(
crates/heal/src/heal/manager/tests.rs
+348 -3
View File
@@ -101,6 +101,326 @@ async fn process_manager_queue_once(manager: &HealManager) {
struct MockStorage;
fn completed_retention_fixture(completed_at: SystemTime) -> CompletedHealStatus {
CompletedHealStatus {
heal_type: HealType::Cluster,
status: HealTaskStatus::Completed,
progress: Some(HealProgress {
objects_scanned: 9,
objects_healed: 8,
objects_failed: 1,
..Default::default()
}),
retained_bytes: std::sync::OnceLock::new(),
result_items_truncated: false,
completed_at,
seqed_items: vec![(3, HealResultItem::default()), (4, HealResultItem::default())],
next_seq: 5,
min_seq: 3,
}
}
#[test]
fn completed_retention_cursor_boundaries_preserve_progress() {
let completed = completed_retention_fixture(SystemTime::now());
for (cursor, count, lagged) in [
(0, 2, true),
(1, 2, true),
(2, 2, false),
(3, 1, false),
(4, 0, false),
(5, 0, false),
(u64::MAX, 0, false),
] {
let report = completed_task_report(&completed, Some(cursor));
assert_eq!(report.result_items.len(), count, "cursor={cursor}");
assert_eq!(report.result_items_truncated, lagged, "cursor={cursor}");
assert_eq!(report.progress, completed.progress);
assert_eq!((report.next_seq, report.min_seq), (5, 3));
}
assert_eq!(completed_task_report(&completed, None).result_items.len(), 2);
}
#[tokio::test]
async fn completed_retention_displaced_alias_does_not_resurrect_evicted_snapshot() {
let manager = HealManager::new(Arc::new(MockStorage), None);
let request = HealRequest::bucket("bucket".to_string());
manager.insert_task_alias("alias", &request.id).await;
let terminal = record_displaced_terminal(&manager.displaced_terminals, &request);
lock_displaced_terminals(&manager.displaced_terminals).remove(&request.id);
remove_displaced_task_aliases(&manager.task_aliases, &manager.displaced_terminals, &request.id, &terminal).await;
for token in [&request.id, &"alias".to_string()] {
assert!(matches!(manager.get_task_report(token).await, Err(Error::TaskNotFound { .. })));
}
assert!(manager.task_aliases.lock().await.is_empty());
assert!(lock_displaced_terminals(&manager.displaced_terminals).is_empty());
}
#[test]
fn completed_retention_count_ttl_and_alias_eviction_are_bounded() {
let now = SystemTime::now();
let mut entries = HashMap::new();
let oldest = Arc::new(completed_retention_fixture(now - KEEP_HEAL_TASK_STATUS_DURATION));
entries.insert("oldest".to_string(), Arc::clone(&oldest));
entries.insert("oldest-alias".to_string(), Arc::clone(&oldest));
for index in 2..MAX_COMPLETED_HEAL_TOKENS {
entries.insert(format!("task-{index}"), Arc::new(completed_retention_fixture(now)));
}
prune_completed_heal_statuses_at(&mut entries, now);
assert_eq!(entries.len(), MAX_COMPLETED_HEAL_TOKENS);
entries.insert("cap-plus-one".to_string(), Arc::new(completed_retention_fixture(now)));
prune_completed_heal_statuses_at(&mut entries, now);
assert_eq!(entries.len(), MAX_COMPLETED_HEAL_TOKENS - 1);
assert!(!entries.contains_key("oldest"));
assert!(!entries.contains_key("oldest-alias"));
entries.clear();
entries.insert("ttl-boundary".to_string(), oldest);
entries.insert(
"expired".to_string(),
Arc::new(completed_retention_fixture(
now - KEEP_HEAL_TASK_STATUS_DURATION - Duration::from_nanos(1),
)),
);
entries.insert("future".to_string(), Arc::new(completed_retention_fixture(now + Duration::from_nanos(1))));
prune_completed_heal_statuses_at(&mut entries, now);
assert_eq!(entries.len(), 1);
assert!(entries.contains_key("ttl-boundary"));
prune_completed_heal_statuses_at(&mut entries, now + Duration::from_nanos(1));
assert!(entries.is_empty());
}
#[test]
fn completed_retention_total_byte_cap_and_cap_plus_one() {
let now = SystemTime::now();
let key = "large".to_string();
let mut entry = completed_retention_fixture(now);
let base_bytes = entry.retained_bytes() + key.capacity() + size_of::<(String, Arc<CompletedHealStatus>)>();
entry.retained_bytes.take();
entry.status = HealTaskStatus::Failed {
error: "x".repeat(MAX_COMPLETED_HEAL_BYTES - base_bytes),
};
assert_eq!(
entry.retained_bytes() + key.capacity() + size_of::<(String, Arc<CompletedHealStatus>)>(),
MAX_COMPLETED_HEAL_BYTES
);
let mut entries = HashMap::from([(key, Arc::new(entry))]);
prune_completed_heal_statuses_at(&mut entries, now);
assert_eq!(entries.len(), 1, "exact byte cap remains retained");
let mut over = Arc::try_unwrap(entries.remove("large").expect("entry retained")).expect("entry not shared");
over.retained_bytes.take();
if let HealTaskStatus::Failed { error } = &mut over.status {
*error = "x".repeat(error.len() + 1);
}
entries.insert("large".to_string(), Arc::new(over));
prune_completed_heal_statuses_at(&mut entries, now);
assert!(entries.is_empty(), "oversized metadata cannot escape total byte bound");
}
#[tokio::test]
async fn completed_retention_large_window_keeps_cursors_and_progress() {
let task = HealTask::from_request(HealRequest::bucket("bucket".to_string()), Arc::new(MockStorage));
let mut snapshot = completed_retention_fixture(SystemTime::now());
snapshot.seqed_items[0].1.detail = "x".repeat(MAX_COMPLETED_HEAL_RESULT_BYTES);
snapshot.bound_result_window();
assert_eq!(snapshot.seqed_items.len(), 1);
assert_eq!((snapshot.min_seq, snapshot.next_seq), (4, 5));
assert!(snapshot.result_items_truncated);
assert!(snapshot.retained_bytes() < MAX_COMPLETED_HEAL_RESULT_BYTES);
let report = completed_task_report(&snapshot, Some(0));
assert_eq!(report.progress.expect("progress retained").objects_scanned, 9);
assert!(report.result_items_truncated);
let active_max = task.get_result_items_since(Some(u64::MAX)).await;
assert!(active_max.items.is_empty());
assert!(!active_max.lagged);
}
#[test]
fn completed_retention_result_byte_cap_and_cap_plus_one() {
for extra in [0, 1] {
let mut snapshot = completed_retention_fixture(SystemTime::now());
snapshot.seqed_items = vec![(
4,
HealResultItem {
detail: "x".repeat(MAX_COMPLETED_HEAL_RESULT_BYTES - size_of::<(u64, HealResultItem)>() + extra),
..Default::default()
},
)];
snapshot.min_seq = 4;
snapshot.bound_result_window();
assert_eq!(snapshot.seqed_items.len(), 1 - extra);
assert_eq!(snapshot.result_items_truncated, extra == 1);
assert_eq!(snapshot.min_seq, if extra == 0 { 4 } else { 5 });
assert_eq!(snapshot.next_seq, 5);
assert_eq!(snapshot.progress.as_ref().expect("progress retained").objects_scanned, 9);
}
}
#[derive(Default)]
struct CompletedRetentionHook {
started: Notify,
execute: Notify,
handoff: Notify,
finish: Notify,
pause_before_publish: bool,
before_publish: Notify,
publish: Notify,
prepared_status: Mutex<Option<HealTaskStatus>>,
}
static COMPLETED_RETENTION_HOOKS: LazyLock<Mutex<HashMap<String, Arc<CompletedRetentionHook>>>> =
LazyLock::new(|| Mutex::new(HashMap::new()));
pub(super) async fn pause_completed_retention_handoff(task_id: &str) {
let hook = COMPLETED_RETENTION_HOOKS.lock().await.get(task_id).cloned();
if let Some(hook) = hook {
hook.handoff.notify_one();
hook.finish.notified().await;
}
}
pub(super) async fn pause_completed_retention_before_publish(task_id: &str, status: &HealTaskStatus) {
let hook = COMPLETED_RETENTION_HOOKS.lock().await.get(task_id).cloned();
if let Some(hook) = hook.filter(|hook| hook.pause_before_publish) {
*hook.prepared_status.lock().await = Some(status.clone());
hook.before_publish.notify_one();
hook.publish.notified().await;
}
}
#[tokio::test]
async fn completed_retention_cancel_wins_over_a_prepared_retry_snapshot() {
let bucket = "completed-retention-retry-cancel";
let manager = HealManager::new(Arc::new(MockStorage), None);
let request = HealRequest::object(bucket.to_string(), "object".to_string(), None);
let task_id = request.id.clone();
let duplicate = HealRequest::object(bucket.to_string(), "object".to_string(), None);
let alias = duplicate.id.clone();
let hook = Arc::new(CompletedRetentionHook {
pause_before_publish: true,
..Default::default()
});
{
let mut hooks = COMPLETED_RETENTION_HOOKS.lock().await;
hooks.insert(bucket.to_string(), Arc::clone(&hook));
hooks.insert(task_id.clone(), Arc::clone(&hook));
}
manager.submit_heal_request(request).await.expect("admit original");
manager.submit_heal_request(duplicate).await.expect("admit alias");
process_manager_queue_once(&manager).await;
tokio::time::timeout(Duration::from_secs(5), hook.started.notified())
.await
.expect("scheduler starts");
let task = manager.active_heals.lock().await.get(&task_id).cloned().expect("active task");
task.progress.write().await.update_object_progress(1, 1, 0, 0, 4096);
hook.execute.notify_one();
tokio::time::timeout(Duration::from_secs(5), hook.before_publish.notified())
.await
.expect("retry snapshot prepared");
manager.cancel_task(&alias).await.expect("cancel wins active ownership");
assert!(matches!(*hook.prepared_status.lock().await, Some(HealTaskStatus::Retrying { .. })));
hook.publish.notify_one();
tokio::time::timeout(Duration::from_secs(5), hook.handoff.notified())
.await
.expect("scheduler finishes handoff");
for token in [&task_id, &alias] {
let report = manager.get_task_report(token).await.expect("cancelled token retained");
assert_eq!(report.status, HealTaskStatus::Cancelled);
assert_eq!(report.progress.expect("frozen progress").objects_scanned, 1);
}
assert!(!manager.retrying_heals.lock().await.contains_key(&task_id));
assert!(!manager.heal_queue.lock().await.contains_request_id(&task_id));
hook.finish.notify_one();
COMPLETED_RETENTION_HOOKS
.lock()
.await
.retain(|key, _| key != bucket && key != &task_id);
}
#[tokio::test]
async fn completed_retention_scheduler_preserves_progress_aliases_and_atomic_handoff() {
for outcome in ["success", "failed", "cancelled"] {
let bucket = format!("completed-retention-{outcome}");
let hook = Arc::new(CompletedRetentionHook::default());
let manager = Arc::new(HealManager::new(Arc::new(MockStorage), None));
let request = HealRequest::object(bucket.clone(), "object".to_string(), None);
let task_id = request.id.clone();
let duplicate = HealRequest::object(bucket.clone(), "object".to_string(), None);
let alias = duplicate.id.clone();
{
let mut hooks = COMPLETED_RETENTION_HOOKS.lock().await;
hooks.insert(bucket.clone(), Arc::clone(&hook));
hooks.insert(task_id.clone(), Arc::clone(&hook));
}
manager.submit_heal_request(request).await.expect("admit original");
manager.submit_heal_request(duplicate).await.expect("admit alias");
process_manager_queue_once(&manager).await;
tokio::time::timeout(Duration::from_secs(5), hook.started.notified())
.await
.expect("scheduler reaches storage");
let task = manager
.active_heals
.lock()
.await
.get(&task_id)
.cloned()
.expect("task is active");
task.progress.write().await.update_object_progress(1, 1, 0, 0, 4096);
let before = manager.get_task_report(&alias).await.expect("alias resolves active progress");
assert_eq!(before.progress.as_ref().expect("active progress").objects_scanned, 1);
let poll_manager = Arc::clone(&manager);
let poll_alias = alias.clone();
let stop = CancellationToken::new();
let poll_stop = stop.clone();
let polling = tokio::spawn(async move {
while !poll_stop.is_cancelled() {
let report = poll_manager
.get_task_report(&poll_alias)
.await
.expect("handoff must never return NotFound");
assert!(report.progress.expect("progress never disappears").objects_scanned >= 1);
tokio::task::yield_now().await;
}
});
if outcome == "cancelled" {
manager.cancel_task(&alias).await.expect("cancel active task by alias");
} else {
hook.execute.notify_one();
}
tokio::time::timeout(Duration::from_secs(5), hook.handoff.notified())
.await
.expect("scheduler archives terminal");
assert!(!manager.active_heals.lock().await.contains_key(&task_id));
let expected = task.get_progress().await;
for token in [&task_id, &alias] {
assert_eq!(manager.get_task_progress(token).await.expect("terminal progress query"), expected);
let report = manager
.get_task_report_for_path_since(&format!("{bucket}/object"), token, Some(u64::MAX))
.await
.expect("terminal token remains queryable at handoff");
assert_eq!(report.progress.as_ref(), Some(&expected));
assert!(report.result_items.is_empty());
match outcome {
"success" => assert_eq!(report.status, HealTaskStatus::Completed),
"failed" => assert!(matches!(report.status, HealTaskStatus::Failed { .. })),
_ => assert_eq!(report.status, HealTaskStatus::Cancelled),
}
}
let retained = manager.completed_heals.lock().await;
assert!(Arc::ptr_eq(&retained[&task_id], &retained[&alias]));
drop(retained);
stop.cancel();
polling.await.expect("concurrent polling succeeds");
// Archived progress must not alias a mutable live progress object.
task.progress.write().await.objects_scanned = 999;
assert_eq!(manager.get_task_report(&alias).await.expect("frozen report").progress, Some(expected));
hook.finish.notify_one();
COMPLETED_RETENTION_HOOKS
.lock()
.await
.retain(|key, _| key != &bucket && key != &task_id);
}
}
#[async_trait::async_trait]
impl HealStorageAPI for MockStorage {
async fn get_object_meta(&self, _bucket: &str, _object: &str) -> Result<Option<HealObjectInfo>> {
@@ -123,6 +443,12 @@ impl HealStorageAPI for MockStorage {
}
async fn object_exists(&self, bucket: &str, _object: &str) -> Result<bool> {
let hook = COMPLETED_RETENTION_HOOKS.lock().await.get(bucket).cloned();
if let Some(hook) = hook {
hook.started.notify_one();
hook.execute.notified().await;
return Ok(true);
}
Ok(bucket == "retry-transition")
}
@@ -133,13 +459,18 @@ impl HealStorageAPI for MockStorage {
_version_id: Option<&str>,
_opts: &HealOpts,
) -> Result<(HealResultItem, Option<Error>)> {
if bucket == "completed-retention-failed" {
return Err(Error::TaskExecutionFailed {
message: "retention fixture failure".to_string(),
});
}
if let Some(hook) = manager_recovery_test_hook() {
*hook
.heal_object_calls
.lock()
.expect("manager recovery object call lock should not poison") += 1;
}
if bucket == "retry-transition" {
if matches!(bucket, "retry-transition" | "completed-retention-retry-cancel") {
return Ok((
HealResultItem::default(),
Some(Error::Storage(EcstoreError::InsufficientReadQuorum(
@@ -1145,7 +1476,13 @@ async fn test_active_duplicate_token_can_query_and_cancel_original_task() {
.expect("duplicate token should cancel merged active task");
assert!(manager.active_heals.lock().await.get(&active_task_id).is_none());
assert!(matches!(manager.get_task_status(&active_task_id).await, Err(Error::TaskNotFound { .. })));
assert_eq!(
manager
.get_task_status(&active_task_id)
.await
.expect("cancelled task remains queryable"),
HealTaskStatus::Cancelled
);
}
#[tokio::test]
@@ -1638,6 +1975,8 @@ async fn insert_retrying_request(manager: &HealManager, request: HealRequest) ->
manager.completed_heals.lock().await.insert(
task_id,
Arc::new(CompletedHealStatus {
progress: None,
retained_bytes: std::sync::OnceLock::new(),
heal_type: request.heal_type,
status: HealTaskStatus::Retrying {
error: "Lock acquisition timeout".to_string(),
@@ -2053,7 +2392,7 @@ async fn admin_force_start_cancels_overlapping_active_task_first() {
"the overlapping admin task must be cancelled (removed from the active table) before the new one starts"
);
assert!(
matches!(manager.get_task_status(&old_id).await, Err(Error::TaskNotFound { .. })),
matches!(manager.get_task_status(&old_id).await, Ok(HealTaskStatus::Cancelled)),
"a cancelled task must no longer resolve as an active heal"
);
}
@@ -2360,6 +2699,8 @@ async fn test_retrying_completion_outranks_the_queue_for_the_same_id() {
manager.completed_heals.lock().await.insert(
task_id.clone(),
Arc::new(CompletedHealStatus {
progress: None,
retained_bytes: std::sync::OnceLock::new(),
heal_type: request.heal_type.clone(),
status: HealTaskStatus::Retrying {
error: "transient disk failure".to_string(),
@@ -2395,6 +2736,8 @@ async fn test_get_task_status_reads_recent_completed_status() {
manager.completed_heals.lock().await.insert(
"completed-token".to_string(),
Arc::new(CompletedHealStatus {
progress: None,
retained_bytes: std::sync::OnceLock::new(),
heal_type: HealType::Bucket {
bucket: "bucket".to_string(),
},
@@ -2424,6 +2767,8 @@ async fn test_get_task_report_for_path_reads_completed_items() {
manager.completed_heals.lock().await.insert(
"completed-token".to_string(),
Arc::new(CompletedHealStatus {
progress: None,
retained_bytes: std::sync::OnceLock::new(),
heal_type: HealType::Object {
bucket: "bucket".to_string(),
object: "object".to_string(),
crates/heal/src/heal/task.rs
+1 -1
View File
@@ -999,7 +999,7 @@ impl HealTask {
let items = match since {
None => result_items.iter().map(|(_, item)| item.clone()).collect::<Vec<_>>(),
Some(cursor) => {
if cursor + 1 < min_seq {
if cursor.saturating_add(1) < min_seq {
lagged = true;
}
result_items
crates/scanner/src/scanner.rs
+15 -8
View File
@@ -1703,14 +1703,18 @@ where
let (sender, receiver) = mpsc::channel::<DataUsageInfo>(1);
let done_cycle = Metrics::time(Metric::ScanCycle);
let scan_result = crate::scanner_io::nsscanner_with_storage_status(
let scan_result = crate::scanner_io::nsscanner_with_storage_status_scoped(
storeapi.as_ref(),
cycle_budget.token(),
cycle_budget.clone(),
sender,
cycle_info.current,
leader_epoch,
scan_mode,
crate::scanner_io::ScannerCycleRequest {
ctx: cycle_budget.token(),
budget: cycle_budget.clone(),
updates: sender,
want_cycle: cycle_info.current,
leader_epoch,
scan_mode,
scan_scope: crate::scanner_io::ScannerBucketScanScope::default(),
persisted_usage_baseline: usage_persist_baseline.data.clone(),
},
)
.await;
let publication_defer_reason = match &scan_result {
@@ -3424,10 +3428,13 @@ use cycle_state::*;
use leadership::*;
use usage_store::*;
#[cfg(test)]
pub(crate) use activity::scanner_activity_snapshot_digest;
pub use activity::scanner_topology_digest;
pub(crate) use activity::{
ScannerActivitySnapshot, ScannerDirtyUsageAcknowledgement, probe_scanner_activity, scanner_activity_allows_usage_publication,
scanner_activity_publication_lease_targets, scanner_activity_snapshot_digest, scanner_dirty_usage_acknowledgements,
scanner_activity_dirty_usage_state_for_host, scanner_activity_publication_lease_targets, scanner_activity_structural_digest,
scanner_dirty_usage_acknowledgements,
};
pub(crate) use activity::{ScannerCycleOutcome, scanner_cycle_outcome_with_pending_maintenance};
pub use backlog::{
crates/scanner/src/scanner/activity.rs
+41
View File
@@ -902,6 +902,7 @@ where
observation
}
#[cfg(test)]
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());
@@ -925,6 +926,30 @@ pub(crate) fn scanner_activity_snapshot_digest(snapshot: &ScannerActivitySnapsho
hasher.finalize().into()
}
/// Hash the activity inputs that make an existing scanner cache unsafe to
/// reuse. Regular namespace writes and dirty-usage generations are omitted:
/// their affected buckets are tracked separately and may be refreshed from a
/// complete authoritative cache baseline.
pub(crate) fn scanner_activity_structural_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.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.movement_generation.to_be_bytes());
hasher.update([u8::from(activity.publication_blocked)]);
}
hasher.finalize().into()
}
pub(crate) fn scanner_activity_allows_usage_publication(snapshot: &ScannerActivitySnapshot) -> bool {
!snapshot.is_empty()
&& snapshot.values().all(|activity| {
@@ -955,6 +980,22 @@ pub(crate) fn scanner_dirty_usage_acknowledgements(snapshot: &ScannerActivitySna
.collect()
}
pub(crate) fn scanner_activity_dirty_usage_state_for_host<'a>(
snapshot: &'a ScannerActivitySnapshot,
host: &str,
) -> Option<(&'a str, u64, bool)> {
snapshot
.get(host)
.filter(|_| host != LOCAL_SCANNER_ACTIVITY_NODE)
.map(|activity| {
(
activity.instance_id.as_str(),
activity.dirty_usage_generation,
activity.dirty_usage_pending,
)
})
}
pub fn scanner_topology_digest(storeapi: &ECStore) -> [u8; 32] {
let endpoint_pools = storeapi.endpoints();
let mut hasher = Sha256::new();
crates/scanner/src/scanner/tests.rs
+38
View File
@@ -8169,6 +8169,44 @@ fn scanner_activity_snapshot_digest_fences_dirty_usage_state() {
assert_ne!(scanner_activity_snapshot_digest(&clean), scanner_activity_snapshot_digest(&pending));
}
#[test]
fn scanner_activity_structural_digest_ignores_regular_bucket_writes() {
let baseline = BTreeMap::from([("node-2".to_string(), scanner_node_activity("epoch-a", 7, 3))]);
let mut written = baseline.clone();
let activity = written.get_mut("node-2").expect("node should exist");
activity.namespace_generation = 8;
activity.dirty_usage_generation = 6;
activity.dirty_usage_pending = true;
assert_ne!(scanner_activity_snapshot_digest(&baseline), scanner_activity_snapshot_digest(&written));
assert_eq!(
scanner_activity_structural_digest(&baseline),
scanner_activity_structural_digest(&written),
"bucket writes are refreshed through the dirty-bucket scope rather than invalidating every cache"
);
}
#[test]
fn scanner_activity_structural_digest_fences_restart_and_maintenance() {
let baseline = BTreeMap::from([("node-2".to_string(), scanner_node_activity("epoch-a", 7, 3))]);
let mut restarted = baseline.clone();
restarted.get_mut("node-2").expect("node should exist").instance_id = "epoch-b".to_string();
let mut maintained = baseline.clone();
maintained
.get_mut("node-2")
.expect("node should exist")
.maintenance_generation = 4;
assert_ne!(
scanner_activity_structural_digest(&baseline),
scanner_activity_structural_digest(&restarted)
);
assert_ne!(
scanner_activity_structural_digest(&baseline),
scanner_activity_structural_digest(&maintained)
);
}
#[test]
fn scanner_dirty_usage_acknowledgements_exclude_local_and_clean_nodes() {
let snapshot = BTreeMap::from([
crates/scanner/src/scanner_io.rs
+118 -1
View File
@@ -21,6 +21,7 @@ use crate::{
DataUsageCacheSource, DataUsageEntry, DataUsageEntryInfo, DataUsageInfo, DataUsageScanPlanDigest, DataUsageSnapshotSetState,
ScannerError, SizeSummary, TierStats,
};
use bytes::Bytes;
use futures::future::join_all;
use metrics::counter;
use rand::seq::SliceRandom as _;
@@ -54,6 +55,7 @@ use tokio_util::task::AbortOnDropHandle;
use tracing::{debug, error, warn};
use crate::ScannerObjectInfo as ObjectInfo;
use crate::storage_api::EcstoreScannerPeerDirtyUsageSnapshot;
use crate::storage_api::ScannerStorage;
use crate::storage_api::scan::NamespaceLocking as _;
use crate::storage_api::scanner_io::{BucketInfo, BucketOptions};
@@ -111,6 +113,121 @@ pub(crate) struct ScannerBucketScanScope {
baseline_scan_plan_digest: Option<DataUsageScanPlanDigest>,
}
impl ScannerBucketScanScope {
fn is_default(&self) -> bool {
self.selected_buckets.is_none() && self.baseline_scan_plan_digest.is_none()
}
fn from_dirty_buckets(selected_buckets: HashSet<String>, baseline_scan_plan_digest: DataUsageScanPlanDigest) -> Self {
Self {
selected_buckets: Some(Arc::new(selected_buckets)),
baseline_scan_plan_digest: Some(baseline_scan_plan_digest),
}
}
}
#[derive(Clone, Copy)]
pub(super) struct ScannerCacheBaselineProof<'a> {
pub(super) data: Option<&'a Bytes>,
pub(super) expected_sources: &'a HashSet<DataUsageCacheSource>,
pub(super) leader_epoch: u64,
pub(super) want_cycle: u64,
pub(super) scan_plan_digest: DataUsageScanPlanDigest,
}
#[derive(Clone, Debug, PartialEq, Eq)]
struct ScannerPeerDirtyUsageExpectation {
instance_id: String,
generation: u64,
pending: bool,
}
fn verified_remote_dirty_usage_buckets(
expected_peers: &HashMap<String, ScannerPeerDirtyUsageExpectation>,
peer_snapshots: Vec<(String, EcstoreScannerPeerDirtyUsageSnapshot)>,
) -> Option<HashSet<String>> {
if expected_peers.is_empty() || peer_snapshots.len() != expected_peers.len() {
return None;
}
let mut received_peers = HashSet::with_capacity(peer_snapshots.len());
let mut dirty_buckets = HashSet::new();
for (host, snapshot) in peer_snapshots {
let expected = expected_peers.get(&host)?;
if !received_peers.insert(host)
|| snapshot.instance_id != expected.instance_id
|| snapshot.generation != expected.generation
|| snapshot.generation == u64::MAX
|| snapshot.protocol_version != crate::SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION
|| !snapshot.complete
|| snapshot.pending_bucket_count != u64::try_from(snapshot.buckets.len()).unwrap_or(u64::MAX)
|| (expected.pending && snapshot.pending_bucket_count == 0)
{
return None;
}
dirty_buckets.extend(snapshot.buckets.into_keys());
}
(received_peers.len() == expected_peers.len()).then_some(dirty_buckets)
}
fn complete_scanner_cache_baseline_plan_digest(proof: ScannerCacheBaselineProof<'_>) -> Option<DataUsageScanPlanDigest> {
let data = proof.data?;
let baseline = serde_json::from_slice::<DataUsageInfo>(data).ok()?;
if !baseline.is_complete_bucket_usage_snapshot()
|| baseline.usage_snapshot_partial
|| baseline.usage_snapshot_converged != Some(true)
|| baseline.scanner_epoch != Some(proof.leader_epoch)
|| baseline.usage_snapshot_set_states.len() != proof.expected_sources.len()
{
return None;
}
let mut states = HashSet::with_capacity(baseline.usage_snapshot_set_states.len());
for state in &baseline.usage_snapshot_set_states {
let source = DataUsageCacheSource::new(usize::try_from(state.pool_index).ok()?, usize::try_from(state.set_index).ok()?);
if !proof.expected_sources.contains(&source)
|| !states.insert(source)
|| !state.complete
|| state.tombstone
|| state.scanner_epoch != Some(proof.leader_epoch)
|| state.scanner_cycle.is_none_or(|cycle| cycle > proof.want_cycle)
|| state.scan_plan_digest != Some(proof.scan_plan_digest.0)
{
return None;
}
}
(states == *proof.expected_sources).then_some(proof.scan_plan_digest)
}
fn scoped_scan_scope_from_dirty_buckets(
requested_scope: ScannerBucketScanScope,
dirty_buckets: HashSet<String>,
dirty_snapshot_complete: bool,
all_buckets: &[BucketInfo],
baseline_proof: ScannerCacheBaselineProof<'_>,
) -> ScannerBucketScanScope {
if !requested_scope.is_default() || !dirty_snapshot_complete {
return requested_scope;
}
let current_buckets = all_buckets.iter().map(|bucket| bucket.name.as_str()).collect::<HashSet<_>>();
let selected_buckets = dirty_buckets
.into_iter()
.filter(|bucket| current_buckets.contains(bucket.as_str()))
.collect::<HashSet<_>>();
if selected_buckets.is_empty() {
return requested_scope;
}
let Some(baseline_scan_plan_digest) = complete_scanner_cache_baseline_plan_digest(baseline_proof) else {
return requested_scope;
};
ScannerBucketScanScope::from_dirty_buckets(selected_buckets, baseline_scan_plan_digest)
}
pub(crate) fn is_scanner_metadata_corrupt_error(err: &StorageError) -> bool {
matches!(err, StorageError::Io(io) if io.to_string().starts_with(SCANNER_METADATA_CORRUPT_ERROR))
}
@@ -749,7 +866,7 @@ mod io_cache;
mod io_cycle;
#[cfg(test)]
use io_cache::{ScannerSetCacheGeneration, prepare_scoped_set_scan};
pub(crate) use io_cycle::nsscanner_with_storage_status;
pub(crate) use io_cycle::{ScannerCycleRequest, nsscanner_with_storage_status_scoped};
mod io_disk;
#[cfg(test)]
mod publish_gate_tests;
crates/scanner/src/scanner_io/cache.rs
+18
View File
@@ -282,9 +282,26 @@ pub(super) fn completed_data_usage_info(
.iter()
.map(|(bucket, usage)| (bucket.clone(), usage.size))
.collect();
let mut usage_snapshot_set_states = results
.iter()
.map(|result| {
let source = result.info.source?;
Some(DataUsageSnapshotSetState {
pool_index: u64::try_from(source.pool_index).ok()?,
set_index: u64::try_from(source.set_index).ok()?,
scanner_cycle: Some(result.info.next_cycle),
scanner_epoch: Some(result.info.leader_epoch),
scan_plan_digest: Some(result.info.scan_plan_digest?.0),
complete: true,
tombstone: false,
})
})
.collect::<Option<Vec<_>>>()?;
usage_snapshot_set_states.sort_by_key(|state| (state.pool_index, state.set_index));
let data_usage_info = DataUsageInfo {
last_update: Some(merged_last_update),
scanner_cycle: Some(results.first()?.info.next_cycle),
scanner_epoch: Some(results.first()?.info.leader_epoch),
objects_total_count: u64::try_from(total.objects).ok()?,
versions_total_count: u64::try_from(total.versions).ok()?,
delete_markers_total_count: u64::try_from(total.delete_markers).ok()?,
@@ -295,6 +312,7 @@ pub(super) fn completed_data_usage_info(
bucket_sizes,
buckets_usage,
usage_snapshot_complete: true,
usage_snapshot_set_states,
..Default::default()
};
Some((data_usage_info, merged_last_update))
crates/scanner/src/scanner_io/io_cycle.rs
+94 -1
View File
@@ -71,6 +71,7 @@ where
leader_epoch,
scan_mode,
scan_scope: ScannerBucketScanScope::default(),
persisted_usage_baseline: None,
};
nsscanner_with_storage_status_scoped(store, request).await
}
@@ -83,6 +84,79 @@ pub(crate) struct ScannerCycleRequest {
pub(crate) leader_epoch: u64,
pub(crate) scan_mode: HealScanMode,
pub(crate) scan_scope: ScannerBucketScanScope,
pub(crate) persisted_usage_baseline: Option<Bytes>,
}
struct ScannerBucketScopeResolution<'a> {
requested_scope: ScannerBucketScanScope,
baseline_proof: ScannerCacheBaselineProof<'a>,
activity_before: &'a crate::scanner::ScannerActivitySnapshot,
dirty_usage_snapshot: &'a DirtyUsageSnapshot,
all_buckets: &'a [BucketInfo],
}
async fn resolve_scanner_bucket_scan_scope<S>(
store: &S,
distributed: bool,
resolution: ScannerBucketScopeResolution<'_>,
) -> ScannerBucketScanScope
where
S: ScannerStorage,
{
if !resolution.requested_scope.is_default()
|| !resolution.dirty_usage_snapshot.covers_all_pending
|| resolution.dirty_usage_snapshot.generation == u64::MAX
|| resolution.dirty_usage_snapshot.buckets.len() > crate::SCANNER_DIRTY_USAGE_SNAPSHOT_MAX_ENTRIES
{
return resolution.requested_scope;
}
let mut dirty_buckets = resolution
.dirty_usage_snapshot
.buckets
.keys()
.cloned()
.collect::<HashSet<_>>();
if distributed {
let Some(notification_system) = store.scanner_notification_system() else {
return resolution.requested_scope;
};
let Ok(peer_snapshots) = notification_system.scanner_dirty_usage_snapshots().await else {
return resolution.requested_scope;
};
let mut expected_peers = HashMap::new();
for (host, lease_instance_id, _) in crate::scanner::scanner_activity_publication_lease_targets(resolution.activity_before)
{
let Some((activity_instance_id, generation, pending)) =
crate::scanner::scanner_activity_dirty_usage_state_for_host(resolution.activity_before, &host)
else {
return resolution.requested_scope;
};
if activity_instance_id != lease_instance_id || expected_peers.contains_key(&host) {
return resolution.requested_scope;
}
expected_peers.insert(
host,
ScannerPeerDirtyUsageExpectation {
instance_id: activity_instance_id.to_string(),
generation,
pending,
},
);
}
let Some(remote_dirty_buckets) = verified_remote_dirty_usage_buckets(&expected_peers, peer_snapshots) else {
return resolution.requested_scope;
};
dirty_buckets.extend(remote_dirty_buckets);
}
scoped_scan_scope_from_dirty_buckets(
resolution.requested_scope,
dirty_buckets,
true,
resolution.all_buckets,
resolution.baseline_proof,
)
}
pub(crate) async fn nsscanner_with_storage_status_scoped<S>(store: &S, request: ScannerCycleRequest) -> Result<ScannerCycleResult>
@@ -97,6 +171,7 @@ where
leader_epoch,
scan_mode,
scan_scope,
persisted_usage_baseline,
} = request;
let child_token = ctx.child_token();
let _tier_cycle_guard = begin_tier_registry_cycle(want_cycle, leader_epoch);
@@ -186,8 +261,26 @@ where
}
bucket_plan_complete &= buckets_by_source.keys().copied().collect::<HashSet<_>>() == *expected_sources;
let scan_plan_digest =
scanner_bucket_plan_digest(&all_buckets, crate::scanner::scanner_activity_snapshot_digest(&activity_before));
scanner_bucket_plan_digest(&all_buckets, crate::scanner::scanner_activity_structural_digest(&activity_before));
let dirty_usage_snapshot = Arc::new(snapshot_dirty_usage_buckets(&all_buckets, dirty_generation_before_bucket_list));
let scan_scope = resolve_scanner_bucket_scan_scope(
store,
distributed,
ScannerBucketScopeResolution {
requested_scope: scan_scope,
baseline_proof: ScannerCacheBaselineProof {
data: persisted_usage_baseline.as_ref(),
expected_sources: &expected_sources,
leader_epoch,
want_cycle,
scan_plan_digest,
},
activity_before: &activity_before,
dirty_usage_snapshot: &dirty_usage_snapshot,
all_buckets: &all_buckets,
},
)
.await;
let cache_cycle_floor = Arc::new(AtomicU64::new(want_cycle));
let tier_registry = runtime_tier_registry_for_cycle(want_cycle, leader_epoch).await;
let tier_registry_generation = tier_registry.generation;
crates/scanner/src/scanner_io/publish_gate_tests.rs
+22
View File
@@ -655,9 +655,31 @@ fn completed_data_usage_info_requires_every_set_before_publish() {
.expect("all completed sets should produce a publishable data usage snapshot");
assert_eq!(last_update, SystemTime::UNIX_EPOCH + Duration::from_secs(20));
assert_eq!(data_usage_info.scanner_cycle, Some(0));
assert_eq!(data_usage_info.scanner_epoch, Some(0));
assert_eq!(data_usage_info.objects_total_count, 3);
assert_eq!(data_usage_info.buckets_usage.len(), 3);
assert!(data_usage_info.usage_snapshot_complete);
assert_eq!(
data_usage_info
.usage_snapshot_set_states
.iter()
.map(|state| {
(
state.pool_index,
state.set_index,
state.scanner_cycle,
state.scanner_epoch,
state.scan_plan_digest,
state.complete,
state.tombstone,
)
})
.collect::<Vec<_>>(),
vec![
(0, 0, Some(0), Some(0), Some(TEST_PLAN_DIGEST.0), true, false),
(1, 0, Some(0), Some(0), Some(TEST_PLAN_DIGEST.0), true, false),
]
);
assert_eq!(
data_usage_info
.buckets_usage
crates/scanner/src/scanner_io/tests.rs
+190
View File
@@ -17,6 +17,7 @@ use super::io_disk::tier_stats_template;
use super::*;
use crate::scanner_budget::ScannerCycleBudgetConfig;
use crate::scanner_folder::ScannerItem;
use crate::storage_api::EcstoreScannerPeerDirtyUsageSnapshot;
use crate::storage_api::owner::{
EcstorePoolDecommissionInfo, EcstoreRebalStatus, EcstoreRebalanceInfo, EcstoreRebalanceMeta, EcstoreRebalanceStats,
};
@@ -796,6 +797,195 @@ fn complete_set_usage_cache(buckets: &[(&str, usize)], scan_plan_digest: DataUsa
cache
}
fn complete_usage_baseline(
source: DataUsageCacheSource,
scan_plan_digest: DataUsageScanPlanDigest,
scanner_cycle: u64,
scanner_epoch: u64,
) -> bytes::Bytes {
let baseline = DataUsageInfo {
last_update: Some(SystemTime::UNIX_EPOCH + Duration::from_secs(10)),
scanner_cycle: Some(scanner_cycle),
scanner_epoch: Some(scanner_epoch),
buckets_count: 1,
buckets_usage: HashMap::from([("photos".to_string(), Default::default())]),
usage_snapshot_complete: true,
usage_snapshot_converged: Some(true),
usage_snapshot_set_states: vec![DataUsageSnapshotSetState {
pool_index: u64::try_from(source.pool_index).expect("test pool index should fit"),
set_index: u64::try_from(source.set_index).expect("test set index should fit"),
scanner_cycle: Some(scanner_cycle),
scanner_epoch: Some(scanner_epoch),
scan_plan_digest: Some(scan_plan_digest.0),
complete: true,
tombstone: false,
}],
..Default::default()
};
bytes::Bytes::from(serde_json::to_vec(&baseline).expect("test baseline should encode"))
}
#[test]
fn scoped_scan_requires_a_converged_complete_baseline_with_exact_set_provenance() {
let source = DataUsageCacheSource::new(1, 2);
let expected_sources = HashSet::from([source]);
let scan_plan_digest = DataUsageScanPlanDigest([9; 32]);
let baseline = complete_usage_baseline(source, scan_plan_digest, 7, 11);
assert_eq!(
complete_scanner_cache_baseline_plan_digest(ScannerCacheBaselineProof {
data: Some(&baseline),
expected_sources: &expected_sources,
leader_epoch: 11,
want_cycle: 8,
scan_plan_digest,
}),
Some(scan_plan_digest)
);
let mut incomplete = serde_json::from_slice::<DataUsageInfo>(&baseline).expect("test baseline should decode");
incomplete.usage_snapshot_converged = Some(false);
let incomplete = bytes::Bytes::from(serde_json::to_vec(&incomplete).expect("test baseline should encode"));
assert_eq!(
complete_scanner_cache_baseline_plan_digest(ScannerCacheBaselineProof {
data: Some(&incomplete),
expected_sources: &expected_sources,
leader_epoch: 11,
want_cycle: 8,
scan_plan_digest,
}),
None
);
let mut wrong_provenance = serde_json::from_slice::<DataUsageInfo>(&baseline).expect("test baseline should decode");
wrong_provenance.usage_snapshot_set_states[0].scan_plan_digest = Some([8; 32]);
let wrong_provenance = bytes::Bytes::from(serde_json::to_vec(&wrong_provenance).expect("test baseline should encode"));
assert_eq!(
complete_scanner_cache_baseline_plan_digest(ScannerCacheBaselineProof {
data: Some(&wrong_provenance),
expected_sources: &expected_sources,
leader_epoch: 11,
want_cycle: 8,
scan_plan_digest,
}),
None
);
}
#[test]
fn scoped_scan_selects_only_current_dirty_buckets_after_baseline_validation() {
let source = DataUsageCacheSource::new(1, 2);
let expected_sources = HashSet::from([source]);
let baseline_scan_plan_digest = DataUsageScanPlanDigest([4; 32]);
let current_scan_plan_digest = DataUsageScanPlanDigest([5; 32]);
let baseline = complete_usage_baseline(source, current_scan_plan_digest, 7, 11);
let scope = scoped_scan_scope_from_dirty_buckets(
ScannerBucketScanScope::default(),
HashSet::from(["photos".to_string(), "deleted".to_string()]),
true,
&[bucket_info("photos")],
ScannerCacheBaselineProof {
data: Some(&baseline),
expected_sources: &expected_sources,
leader_epoch: 11,
want_cycle: 8,
scan_plan_digest: current_scan_plan_digest,
},
);
assert_eq!(scope.baseline_scan_plan_digest, Some(current_scan_plan_digest));
assert_eq!(
scope
.selected_buckets
.as_deref()
.expect("validated scope should select a bucket"),
&HashSet::from(["photos".to_string()])
);
assert_ne!(scope.baseline_scan_plan_digest, Some(baseline_scan_plan_digest));
}
fn peer_dirty_usage_snapshot(
instance_id: &str,
generation: u64,
complete: bool,
buckets: &[(&str, u64)],
) -> EcstoreScannerPeerDirtyUsageSnapshot {
EcstoreScannerPeerDirtyUsageSnapshot {
instance_id: instance_id.to_string(),
generation,
pending_bucket_count: u64::try_from(buckets.len()).expect("test bucket count should fit"),
protocol_version: crate::SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION,
complete,
buckets: buckets
.iter()
.map(|(bucket, generation)| ((*bucket).to_string(), *generation))
.collect(),
}
}
#[test]
fn verified_remote_dirty_usage_buckets_merges_only_complete_current_snapshots() {
let expected_peers = HashMap::from([
(
"node-a:9000".to_string(),
ScannerPeerDirtyUsageExpectation {
instance_id: "instance-a".to_string(),
generation: 7,
pending: true,
},
),
(
"node-b:9000".to_string(),
ScannerPeerDirtyUsageExpectation {
instance_id: "instance-b".to_string(),
generation: 3,
pending: false,
},
),
]);
assert_eq!(
verified_remote_dirty_usage_buckets(
&expected_peers,
vec![
(
"node-a:9000".to_string(),
peer_dirty_usage_snapshot("instance-a", 7, true, &[("photos", 7)]),
),
(
"node-b:9000".to_string(),
peer_dirty_usage_snapshot("instance-b", 3, true, &[("archive", 3)]),
),
],
),
Some(HashSet::from(["photos".to_string(), "archive".to_string()]))
);
}
#[test]
fn verified_remote_dirty_usage_buckets_rejects_incomplete_or_stale_peer_state() {
let expected_peers = HashMap::from([(
"node-a:9000".to_string(),
ScannerPeerDirtyUsageExpectation {
instance_id: "instance-a".to_string(),
generation: 7,
pending: true,
},
)]);
for snapshot in [
peer_dirty_usage_snapshot("instance-a", 7, false, &[("photos", 7)]),
peer_dirty_usage_snapshot("instance-a", 6, true, &[("photos", 6)]),
peer_dirty_usage_snapshot("instance-b", 7, true, &[("photos", 7)]),
peer_dirty_usage_snapshot("instance-a", 7, true, &[]),
] {
assert!(
verified_remote_dirty_usage_buckets(&expected_peers, vec![("node-a:9000".to_string(), snapshot)]).is_none(),
"incomplete, stale, mismatched, or empty pending peer state must fall back to a full scan"
);
}
}
#[test]
fn scoped_set_scan_preserves_unselected_usage_and_drops_deleted_buckets() {
let baseline_digest = DataUsageScanPlanDigest([1; 32]);
crates/scanner/src/storage_api.rs
+3 -1
View File
@@ -103,7 +103,9 @@ pub(crate) use rustfs_ecstore::api::rebalance::{
RebalStatus as EcstoreRebalStatus, RebalanceInfo as EcstoreRebalanceInfo, RebalanceMeta as EcstoreRebalanceMeta,
RebalanceStats as EcstoreRebalanceStats,
};
pub(crate) use rustfs_ecstore::api::rpc::ScannerBucketListing as EcstoreScannerBucketListing;
pub(crate) use rustfs_ecstore::api::rpc::{
ScannerBucketListing as EcstoreScannerBucketListing, ScannerPeerDirtyUsageSnapshot as EcstoreScannerPeerDirtyUsageSnapshot,
};
#[cfg(test)]
pub(crate) use rustfs_ecstore::api::runtime::InstanceContext as EcstoreInstanceContext;
pub(crate) use rustfs_ecstore::api::runtime::{