gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-09 13:46:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(heal): persist scoped admin admission handoff (#7569)

Browse Source 
* fix(heal): persist root admission handoff

Persist administrator root heal ownership before queue publication so crash restart can replay an admitted control-plane start without relying on process-local replay cache state.

Protect queued durable root owners from priority displacement; callers must use forceStart cancellation semantics to replace them.

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(heal): persist admin heal handoff types

Extend durable admin heal admission handoff beyond cluster roots while preserving legacy root recovery records. Persist the heal type in schema v2, replay typed admin requests after restart, keep non-admin requests out of the durable control-plane handoff, and allow path cancellation to clear durable-only non-root records.

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(heal): preserve disjoint durable admin owners

Limit admin forceStart durable-owner cleanup to overlapping heal requests after the admin recovery journal begins storing non-root work. Add fail-closed and displacement tests so durable bucket owners cannot be silently dropped by unrelated replacement or priority pressure.

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(heal): persist admin terminal receipts

Persist terminal receipts for durable administrator heal starts before retiring their pending intent so restart can answer completed or cancelled scoped tasks without replaying them.

Retain pending work when terminal publication fails and keep timeout owners on checkpoint-only replay semantics.

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

---------

Co-authored-by: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
houseme
2026-09-09 16:22:15 +08:00
committed by GitHub
parent a6e5bdcbd2
commit e1fee0569a
5 changed files with 1273 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/heal/src/heal/manager.rs
+137 -37
View File
@@ -944,6 +944,10 @@ impl HealManager {
matches!(request.source, HealRequestSource::Admin | HealRequestSource::Internal)
}
fn queued_request_can_be_displaced(request: &HealRequest) -> bool {
!root_recovery::is_admin_heal_recovery(&request.heal_type, request.source)
}
fn request_bypasses_mainline_throttle(request: &HealRequest) -> bool {
request.force_start
|| matches!(request.source, HealRequestSource::Admin | HealRequestSource::Internal)
@@ -1080,11 +1084,15 @@ impl HealManager {
let per_object_request = request.heal_type.is_per_object();
if queue_len >= queue_capacity && !request.force_start {
if Self::can_displace_queued_work(&request) && queue.can_displace_lower_priority(request.priority) {
if Self::can_displace_queued_work(&request)
&& queue.can_displace_lower_priority_where(request.priority, Self::queued_request_can_be_displaced)
{
let request_id = request.id.clone();
let priority = request.priority;
let source = request.source;
if let Some(displaced) = queue.push_displacing_lower_priority(request) {
if let Some(displaced) =
queue.push_displacing_lower_priority_where(request, Self::queued_request_can_be_displaced)
{
publish_heal_queue_length(queue);
Self::record_admission_metric(source, HealAdmissionResult::Accepted, context);
demote_to_debug_when!(per_object_request, warn, target: "rustfs::heal::manager", {
@@ -1484,13 +1492,13 @@ impl HealManager {
);
// Keep scheduler, cancellation, and retry ownership stable until every
// unfinished root traversal has a durable successor. A failed write
// must leave the manager running and the shutdown marker unclean.
// unfinished admin control-plane heal has a durable successor. A failed
// write must leave the manager running and the shutdown marker unclean.
let mut active_heals = self.active_heals.lock().await;
let queue = self.heal_queue.lock().await;
let retrying = self.retrying_heals.lock().await;
for task in active_heals.values() {
if root_recovery::is_root_heal(&task.heal_type, task.source) {
if root_recovery::is_admin_heal_recovery(&task.heal_type, task.source) {
if task.get_status().await == HealTaskStatus::Completed {
self.root_recovery.remove(&task.id, &task.heal_type, task.source).await?;
} else {
@@ -1676,7 +1684,9 @@ impl HealManager {
.requests()
.chain(retrying.values().map(|retrying| &retrying.request))
.filter(|pending| {
root_recovery::is_root_heal(&pending.heal_type, pending.source) && pending.id != request.id
root_recovery::is_admin_heal_recovery(&pending.heal_type, pending.source)
&& heal_types_overlap(&request.heal_type, &pending.heal_type) != OverlapVerdict::Disjoint
&& pending.id != request.id
})
.map(|pending| pending.id.clone()),
);
@@ -1698,9 +1708,11 @@ impl HealManager {
}
}
// A failed or timed-out replay may have only its durable owner
// left. Root responsibility overlaps every administrator path.
// left. Cancel only records that overlap this forced start.
for pending in self.root_recovery.pending().await? {
if pending.id != request.id {
if pending.id != request.id
&& heal_types_overlap(&request.heal_type, &pending.heal_type) != OverlapVerdict::Disjoint
{
self.cancel_task(&pending.id).await?;
}
}
@@ -1930,9 +1942,29 @@ impl HealManager {
}
}
let durable_handoff = root_recovery::is_admin_heal_recovery(&request.heal_type, request.source);
let durable_handoff_required = durable_handoff
&& (request.force_start
|| queue.len() < config.queue_size
|| (Self::can_displace_queued_work(&request)
&& queue.can_displace_lower_priority_where(request.priority, Self::queued_request_can_be_displaced)));
// An admin receipt is a control-plane responsibility. Persist it before
// queue publication so a crash after admission can replay it.
if durable_handoff_required {
self.root_recovery.persist(&request).await?;
}
let mut task_id = request.id.clone();
let request_id = request.id.clone();
let request_heal_type = request.heal_type.clone();
let request_source = request.source;
let admission_decision = Self::admit_request_to_queue(&mut queue, request, &config, "submit");
let admission = admission_decision.result;
if durable_handoff_required && !admission.is_admitted() {
self.root_recovery
.remove(&request_id, &request_heal_type, request_source)
.await?;
}
if admission == HealAdmissionResult::Merged
&& let Some(queued_id) = queue.queued_request_id_for_dedup_key(&dedup_key)
{
@@ -2003,7 +2035,7 @@ impl HealManager {
/// Retrying, never Pending — then the queue, and finally a terminal
/// completed entry. `heal_path` additionally constrains the map matches
/// the way the `*_for_path` variants always have.
async fn lookup_task_state(&self, canonical_task_id: &str, heal_path: Option<&str>) -> TaskStateLookup {
async fn lookup_task_state(&self, canonical_task_id: &str, heal_path: Option<&str>) -> Result<TaskStateLookup> {
let matches_path = |heal_type: &HealType| heal_path.is_none_or(|path| heal_type_matches_path(heal_type, path));
{
@@ -2012,7 +2044,7 @@ impl HealManager {
.get(canonical_task_id)
.filter(|task| matches_path(&task.heal_type))
{
return TaskStateLookup::Active(Arc::clone(task));
return Ok(TaskStateLookup::Active(Arc::clone(task)));
}
}
@@ -2022,7 +2054,7 @@ impl HealManager {
.get(canonical_task_id)
.filter(|retrying| matches_path(&retrying.request.heal_type))
{
return TaskStateLookup::Retrying(retrying.status());
return Ok(TaskStateLookup::Retrying(retrying.status()));
}
}
@@ -2035,7 +2067,7 @@ impl HealManager {
prune_completed_heal_statuses(&mut completed_heals);
if let Some(completed) = completed_heals.get(canonical_task_id).filter(|c| matches_path(&c.heal_type)) {
if completed_status_is_retrying(&completed.status) {
return TaskStateLookup::Completed(Arc::clone(completed));
return Ok(TaskStateLookup::Completed(Arc::clone(completed)));
}
terminal_completed = Some(Arc::clone(completed));
}
@@ -2048,7 +2080,7 @@ impl HealManager {
None => queue.contains_request_id(canonical_task_id),
};
if queued {
return TaskStateLookup::Queued;
return Ok(TaskStateLookup::Queued);
}
}
@@ -2061,15 +2093,55 @@ impl HealManager {
.cloned();
}
match terminal_completed {
if terminal_completed.is_none()
&& let Some(completed) = self.root_recovery.completed(canonical_task_id).await?
&& matches_path(&completed.heal_type)
{
terminal_completed = Some(Arc::new(completed));
}
Ok(match terminal_completed {
Some(completed) => TaskStateLookup::Completed(completed),
None => TaskStateLookup::NotFound,
}
})
}
async fn publish_admin_terminal(
&self,
task_id: &str,
heal_type: &HealType,
source: HealRequestSource,
completed: &CompletedHealStatus,
) -> Result<bool> {
self.root_recovery
.persist_terminal(task_id, heal_type, source, completed)
.await
}
async fn publish_admin_cancelled_terminal(
&self,
task_id: &str,
heal_type: &HealType,
source: HealRequestSource,
) -> Result<bool> {
let completed = CompletedHealStatus {
outcome: None,
progress: None,
retained_bytes: std::sync::OnceLock::new(),
heal_type: heal_type.clone(),
status: HealTaskStatus::Cancelled,
result_items_truncated: false,
completed_at: SystemTime::now(),
seqed_items: Vec::new(),
next_seq: 0,
min_seq: 0,
};
self.publish_admin_terminal(task_id, heal_type, source, &completed).await
}
pub async fn get_task_status(&self, task_id: &str) -> Result<HealTaskStatus> {
let canonical_task_id = self.canonical_task_id(task_id).await;
match self.lookup_task_state(&canonical_task_id, None).await {
match self.lookup_task_state(&canonical_task_id, None).await? {
TaskStateLookup::Active(task) => Ok(task.get_status().await),
TaskStateLookup::Retrying(status) => Ok(status),
TaskStateLookup::Completed(completed) => Ok(completed.status.clone()),
@@ -2089,7 +2161,7 @@ impl HealManager {
/// full-snapshot semantics.
pub async fn get_task_report_since(&self, task_id: &str, since: Option<u64>) -> Result<HealTaskReport> {
let canonical_task_id = self.canonical_task_id(task_id).await;
match self.lookup_task_state(&canonical_task_id, None).await {
match self.lookup_task_state(&canonical_task_id, None).await? {
TaskStateLookup::Active(task) => Ok(active_task_report(&task, since).await),
TaskStateLookup::Retrying(status) => Ok(empty_task_report(status)),
TaskStateLookup::Completed(completed) => Ok(completed_task_report(&completed, since)),
@@ -2112,7 +2184,7 @@ impl HealManager {
since: Option<u64>,
) -> Result<HealTaskReport> {
let canonical_task_id = self.canonical_task_id(task_id).await;
match self.lookup_task_state(&canonical_task_id, Some(heal_path)).await {
match self.lookup_task_state(&canonical_task_id, Some(heal_path)).await? {
TaskStateLookup::Active(task) => Ok(active_task_report(&task, since).await),
TaskStateLookup::Retrying(status) => Ok(empty_task_report(status)),
TaskStateLookup::Completed(completed) => Ok(completed_task_report(&completed, since)),
@@ -2135,7 +2207,7 @@ impl HealManager {
/// recently completed task, a different token is invalid for that path.
pub async fn get_task_status_for_path(&self, heal_path: &str, task_id: &str) -> Result<HealTaskStatus> {
let canonical_task_id = self.canonical_task_id(task_id).await;
match self.lookup_task_state(&canonical_task_id, Some(heal_path)).await {
match self.lookup_task_state(&canonical_task_id, Some(heal_path)).await? {
TaskStateLookup::Active(task) => Ok(task.get_status().await),
TaskStateLookup::Retrying(status) => Ok(status),
TaskStateLookup::Completed(completed) => Ok(completed.status.clone()),
@@ -2189,11 +2261,18 @@ impl HealManager {
}
drop(completed_heals);
let mut displaced_terminals = lock_displaced_terminals(&self.displaced_terminals);
prune_completed_heal_statuses(&mut displaced_terminals);
displaced_terminals
.values()
.any(|terminal| heal_type_matches_path(&terminal.heal_type, heal_path))
{
let mut displaced_terminals = lock_displaced_terminals(&self.displaced_terminals);
prune_completed_heal_statuses(&mut displaced_terminals);
if displaced_terminals
.values()
.any(|terminal| heal_type_matches_path(&terminal.heal_type, heal_path))
{
return true;
}
}
self.root_recovery.completed_matches_path(heal_path).await.unwrap_or(false)
}
/// Get task progress
@@ -2205,7 +2284,7 @@ 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 progress = match self.lookup_task_state(&canonical_task_id, None).await {
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,
@@ -2221,9 +2300,11 @@ impl HealManager {
{
let mut active_heals = self.active_heals.lock().await;
if let Some(task) = active_heals.get(&canonical_task_id) {
let completed = CompletedHealStatus::snapshot(task, HealTaskStatus::Cancelled).await;
self.publish_admin_terminal(&canonical_task_id, &task.heal_type, task.source, &completed)
.await?;
self.root_recovery.remove(&task.id, &task.heal_type, task.source).await?;
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);
@@ -2246,6 +2327,8 @@ impl HealManager {
{
let mut retrying_heals = self.retrying_heals.lock().await;
if let Some(retrying) = retrying_heals.get(&canonical_task_id) {
self.publish_admin_cancelled_terminal(&canonical_task_id, &retrying.request.heal_type, retrying.request.source)
.await?;
self.root_recovery
.remove(&canonical_task_id, &retrying.request.heal_type, retrying.request.source)
.await?;
@@ -2271,6 +2354,8 @@ impl HealManager {
let mut queue = self.heal_queue.lock().await;
if let Some(request) = queue.requests().find(|request| request.id == canonical_task_id) {
self.publish_admin_cancelled_terminal(&canonical_task_id, &request.heal_type, request.source)
.await?;
self.root_recovery
.remove(&request.id, &request.heal_type, request.source)
.await?;
@@ -2314,9 +2399,11 @@ impl HealManager {
for task_id in &task_ids {
if let Some(task) = active_heals.get(task_id) {
let completed = CompletedHealStatus::snapshot(task, HealTaskStatus::Cancelled).await;
self.publish_admin_terminal(task_id, &task.heal_type, task.source, &completed)
.await?;
self.root_recovery.remove(&task.id, &task.heal_type, task.source).await?;
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);
@@ -2344,6 +2431,8 @@ impl HealManager {
for task_id in &task_ids {
if let Some(retrying) = retrying_heals.get(task_id) {
self.publish_admin_cancelled_terminal(task_id, &retrying.request.heal_type, retrying.request.source)
.await?;
self.root_recovery
.remove(task_id, &retrying.request.heal_type, retrying.request.source)
.await?;
@@ -2369,15 +2458,22 @@ impl HealManager {
}
}
let mut queue = self.heal_queue.lock().await;
for request in queue
.requests()
.filter(|request| heal_type_matches_path(&request.heal_type, heal_path))
{
let queued_matches = {
let queue = self.heal_queue.lock().await;
queue
.requests()
.filter(|request| heal_type_matches_path(&request.heal_type, heal_path))
.cloned()
.collect::<Vec<_>>()
};
for request in &queued_matches {
self.publish_admin_cancelled_terminal(&request.id, &request.heal_type, request.source)
.await?;
self.root_recovery
.remove(&request.id, &request.heal_type, request.source)
.await?;
}
let mut queue = self.heal_queue.lock().await;
let queued_cancelled = queue.remove_matching(|request| heal_type_matches_path(&request.heal_type, heal_path));
if !queued_cancelled.is_empty() {
publish_heal_queue_length(&queue);
@@ -2389,11 +2485,15 @@ impl HealManager {
self.remove_mrf_repair_notice_targets_for_task(&request.id);
}
if heal_type_matches_path(&HealType::Cluster, heal_path) {
for pending in self.root_recovery.pending().await? {
if self.root_recovery.cancel_pending(&pending.id).await? {
cancelled += 1;
}
for pending in self
.root_recovery
.pending()
.await?
.into_iter()
.filter(|pending| heal_type_matches_path(&pending.heal_type, heal_path))
{
if self.root_recovery.cancel_pending(&pending.id).await? {
cancelled += 1;
}
}
if cancelled == 0 {
crates/heal/src/heal/manager/queue.rs
+22 -4
View File
@@ -288,16 +288,29 @@ impl PriorityHealQueue {
QueuePushOutcome::Accepted
}
pub(super) fn can_displace_lower_priority(&self, priority: HealPriority) -> bool {
self.heap.iter().any(|item| item.priority < priority)
pub(super) fn can_displace_lower_priority_where<F>(&self, priority: HealPriority, can_displace: F) -> bool
where
F: Fn(&HealRequest) -> bool,
{
self.heap
.iter()
.any(|item| item.priority < priority && can_displace(&item.request))
}
pub(super) fn push_displacing_lower_priority(&mut self, request: HealRequest) -> Option<HealRequest> {
#[cfg(test)]
pub(super) fn can_displace_lower_priority(&self, priority: HealPriority) -> bool {
self.can_displace_lower_priority_where(priority, |_| true)
}
pub(super) fn push_displacing_lower_priority_where<F>(&mut self, request: HealRequest, can_displace: F) -> Option<HealRequest>
where
F: Fn(&HealRequest) -> bool,
{
let mut retained = BinaryHeap::new();
let mut displaced: Option<PriorityQueueItem> = None;
while let Some(item) = self.heap.pop() {
if item.priority < request.priority {
if item.priority < request.priority && can_displace(&item.request) {
let should_displace = displaced
.as_ref()
.map(|current| {
@@ -337,6 +350,11 @@ impl PriorityHealQueue {
displaced
}
#[cfg(test)]
pub(super) fn push_displacing_lower_priority(&mut self, request: HealRequest) -> Option<HealRequest> {
self.push_displacing_lower_priority_where(request, |_| true)
}
/// Get statistics about queue contents by priority
pub(super) fn get_priority_stats(&self) -> HashMap<HealPriority, usize> {
let mut stats = HashMap::new();
crates/heal/src/heal/manager/root_recovery.rs
+428 -12
View File
@@ -12,10 +12,11 @@
// See the License for the specific language governing permissions and
// limitations under the License.
//! Graceful-shutdown handoff for administrator root heals. This namespace is
//! Graceful-shutdown handoff for administrator heals. This namespace is
//! separate from erasure-set checkpoints and replacement generations, which
//! cannot represent a cluster traversal. One coordinator disk owns each
//! record; never create a fallback copy after an uncertain write or deletion.
//! cannot represent an admitted admin control-plane request. One coordinator
//! disk owns each record; never create a fallback copy after an uncertain write
//! or deletion.
use super::*;
use crate::heal::storage_api::owner::{EcstoreConditionalFileUpdate, EcstoreDiskAPI, EcstoreDiskBytes};
@@ -25,13 +26,172 @@ use serde::{Deserialize, Serialize};
// The metadata bucket already exists and its parent is durable. Creating a
// nested journal directory here would also require syncing every ancestor.
const ROOT_RECOVERY_PREFIX: &str = "root-heal-";
const ROOT_RECOVERY_SCHEMA: u32 = 1;
const ROOT_TERMINAL_PREFIX: &str = "terminal-root-heal-";
const LEGACY_ROOT_RECOVERY_SCHEMA: u32 = 1;
const ROOT_RECOVERY_SCHEMA: u32 = 2;
const ROOT_TERMINAL_SCHEMA: u32 = 1;
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
enum RecoveryHealType {
Cluster,
Bucket {
bucket: String,
},
Object {
bucket: String,
object: String,
version_id: Option<String>,
},
Prefix {
bucket: String,
prefix: String,
},
ErasureSet {
buckets: Vec<String>,
set_disk_id: String,
},
Metadata {
bucket: String,
object: String,
},
EcDecode {
bucket: String,
object: String,
version_id: Option<String>,
},
}
impl RecoveryHealType {
fn validate(&self) -> Result<()> {
match self {
Self::Cluster => {}
Self::Bucket { bucket } => validate_recovery_component("bucket", bucket)?,
Self::Object {
bucket,
object,
version_id,
}
| Self::EcDecode {
bucket,
object,
version_id,
} => {
validate_recovery_component("bucket", bucket)?;
validate_recovery_component("object", object)?;
if let Some(version_id) = version_id {
validate_recovery_component("version id", version_id)?;
}
}
Self::Prefix { bucket, prefix } => {
validate_recovery_component("bucket", bucket)?;
validate_recovery_component("prefix", prefix)?;
}
Self::ErasureSet { buckets, set_disk_id } => {
validate_recovery_component("set disk id", set_disk_id)?;
if buckets.is_empty() {
return Err(Error::Other("Admin heal recovery erasure set must name buckets".to_string()));
}
for bucket in buckets {
validate_recovery_component("bucket", bucket)?;
}
}
Self::Metadata { bucket, object } => {
validate_recovery_component("bucket", bucket)?;
validate_recovery_component("object", object)?;
}
}
Ok(())
}
}
impl From<&HealType> for RecoveryHealType {
fn from(heal_type: &HealType) -> Self {
match heal_type {
HealType::Cluster => Self::Cluster,
HealType::Bucket { bucket } => Self::Bucket { bucket: bucket.clone() },
HealType::Object {
bucket,
object,
version_id,
} => Self::Object {
bucket: bucket.clone(),
object: object.clone(),
version_id: version_id.clone(),
},
HealType::Prefix { bucket, prefix } => Self::Prefix {
bucket: bucket.clone(),
prefix: prefix.clone(),
},
HealType::ErasureSet { buckets, set_disk_id } => Self::ErasureSet {
buckets: buckets.clone(),
set_disk_id: set_disk_id.clone(),
},
HealType::Metadata { bucket, object } => Self::Metadata {
bucket: bucket.clone(),
object: object.clone(),
},
HealType::ECDecode {
bucket,
object,
version_id,
} => Self::EcDecode {
bucket: bucket.clone(),
object: object.clone(),
version_id: version_id.clone(),
},
}
}
}
impl From<RecoveryHealType> for HealType {
fn from(heal_type: RecoveryHealType) -> Self {
match heal_type {
RecoveryHealType::Cluster => Self::Cluster,
RecoveryHealType::Bucket { bucket } => Self::Bucket { bucket },
RecoveryHealType::Object {
bucket,
object,
version_id,
} => Self::Object {
bucket,
object,
version_id,
},
RecoveryHealType::Prefix { bucket, prefix } => Self::Prefix { bucket, prefix },
RecoveryHealType::ErasureSet { buckets, set_disk_id } => Self::ErasureSet { buckets, set_disk_id },
RecoveryHealType::Metadata { bucket, object } => Self::Metadata { bucket, object },
RecoveryHealType::EcDecode {
bucket,
object,
version_id,
} => Self::ECDecode {
bucket,
object,
version_id,
},
}
}
}
fn validate_recovery_component(label: &str, value: &str) -> Result<()> {
if value.is_empty() || value.contains('\0') {
return Err(Error::Other(format!("Invalid admin heal recovery {label}")));
}
Ok(())
}
fn default_recovery_heal_type() -> RecoveryHealType {
RecoveryHealType::Cluster
}
#[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct RootHealIntent {
schema: u32,
task_id: String,
#[serde(default = "default_recovery_heal_type")]
heal_type: RecoveryHealType,
#[serde(deserialize_with = "decode_options")]
options: HealOptions,
priority: HealPriority,
@@ -39,11 +199,62 @@ struct RootHealIntent {
created_at: SystemTime,
}
#[derive(Debug, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct RootHealTerminal {
schema: u32,
task_id: String,
heal_type: RecoveryHealType,
status: HealTaskStatus,
progress: Option<HealProgress>,
completed_at: SystemTime,
}
impl RootHealTerminal {
fn from_completed(task_id: &str, completed: &CompletedHealStatus) -> Self {
Self {
schema: ROOT_TERMINAL_SCHEMA,
task_id: task_id.to_owned(),
heal_type: RecoveryHealType::from(&completed.heal_type),
status: completed.status.clone(),
progress: completed.progress.clone(),
completed_at: completed.completed_at,
}
}
fn cancelled(task_id: &str, heal_type: &HealType) -> Self {
Self {
schema: ROOT_TERMINAL_SCHEMA,
task_id: task_id.to_owned(),
heal_type: RecoveryHealType::from(heal_type),
status: HealTaskStatus::Cancelled,
progress: None,
completed_at: SystemTime::now(),
}
}
fn into_completed(self) -> CompletedHealStatus {
CompletedHealStatus {
outcome: None,
progress: self.progress,
retained_bytes: std::sync::OnceLock::new(),
heal_type: self.heal_type.into(),
status: self.status,
result_items_truncated: false,
completed_at: self.completed_at,
seqed_items: Vec::new(),
next_seq: 0,
min_seq: 0,
}
}
}
impl RootHealIntent {
fn from_request(request: &HealRequest) -> Self {
Self {
schema: ROOT_RECOVERY_SCHEMA,
task_id: request.id.clone(),
heal_type: RecoveryHealType::from(&request.heal_type),
options: request.options.clone(),
priority: request.priority,
retry_attempts: request.retry_attempts,
@@ -52,7 +263,7 @@ impl RootHealIntent {
}
fn into_request(self) -> HealRequest {
let mut request = HealRequest::new(HealType::Cluster, self.options, self.priority);
let mut request = HealRequest::new(self.heal_type.into(), self.options, self.priority);
request.id = self.task_id;
request.source = HealRequestSource::Admin;
request.retry_attempts = self.retry_attempts;
@@ -68,8 +279,18 @@ pub(super) struct RootHealRecovery {
disks: Option<Vec<DiskStore>>,
}
pub(super) fn is_root_heal(heal_type: &HealType, source: HealRequestSource) -> bool {
source == HealRequestSource::Admin && matches!(heal_type, HealType::Cluster)
pub(super) fn is_admin_heal_recovery(heal_type: &HealType, source: HealRequestSource) -> bool {
source == HealRequestSource::Admin
&& matches!(
heal_type,
HealType::Cluster
| HealType::Bucket { .. }
| HealType::Object { .. }
| HealType::Prefix { .. }
| HealType::ErasureSet { .. }
| HealType::Metadata { .. }
| HealType::ECDecode { .. }
)
}
fn decode_options<'de, D: serde::Deserializer<'de>>(deserializer: D) -> std::result::Result<HealOptions, D::Error> {
@@ -107,16 +328,41 @@ fn intent_path(task_id: &str) -> Result<String> {
Ok(format!("{ROOT_RECOVERY_PREFIX}{task_id}.json"))
}
fn terminal_path(task_id: &str) -> Result<String> {
let parsed = uuid::Uuid::parse_str(task_id).map_err(|_| Error::Other("Invalid root heal terminal task id".to_string()))?;
if parsed.to_string() != task_id {
return Err(Error::Other("Noncanonical root heal terminal task id".to_string()));
}
Ok(format!("{ROOT_TERMINAL_PREFIX}{task_id}.json"))
}
fn decode_intent(task_id: &str, bytes: &[u8]) -> Result<RootHealIntent> {
let _ = intent_path(task_id)?;
let intent: RootHealIntent = serde_json::from_slice(bytes)
.map_err(|error| Error::Other(format!("Invalid root heal recovery record {task_id}: {error}")))?;
if intent.schema != ROOT_RECOVERY_SCHEMA || intent.task_id != task_id {
if intent.task_id != task_id {
return Err(Error::Other(format!("Unsupported or mismatched root heal recovery record {task_id}")));
}
match intent.schema {
LEGACY_ROOT_RECOVERY_SCHEMA if intent.heal_type == RecoveryHealType::Cluster => {}
ROOT_RECOVERY_SCHEMA => {}
_ => return Err(Error::Other(format!("Unsupported or mismatched root heal recovery record {task_id}"))),
}
intent.heal_type.validate()?;
Ok(intent)
}
fn decode_terminal(task_id: &str, bytes: &[u8]) -> Result<RootHealTerminal> {
let _ = terminal_path(task_id)?;
let terminal: RootHealTerminal = serde_json::from_slice(bytes)
.map_err(|error| Error::Other(format!("Invalid root heal terminal record {task_id}: {error}")))?;
if terminal.schema != ROOT_TERMINAL_SCHEMA || terminal.task_id != task_id {
return Err(Error::Other(format!("Unsupported or mismatched root heal terminal record {task_id}")));
}
terminal.heal_type.validate()?;
Ok(terminal)
}
impl RootHealRecovery {
#[cfg(test)]
pub(super) fn with_disks(disks: Vec<DiskStore>) -> Self {
@@ -162,8 +408,55 @@ impl RootHealRecovery {
Ok(found)
}
async fn find_terminal(disks: &[DiskStore], task_id: &str) -> Result<Option<(DiskStore, EcstoreDiskBytes)>> {
let path = terminal_path(task_id)?;
let mut found = None;
for disk in disks {
EcstoreDiskAPI::stat_volume(disk.as_ref(), RUSTFS_META_BUCKET).await?;
match EcstoreDiskAPI::read_all(disk.as_ref(), RUSTFS_META_BUCKET, &path).await {
Ok(bytes) => {
decode_terminal(task_id, &bytes)?;
if found.is_some() {
return Err(Error::Other(format!("Multiple root heal terminal owners for {task_id}")));
}
found = Some((disk.clone(), bytes));
}
Err(DiskError::FileNotFound) => {}
Err(error) => return Err(Error::Disk(error)),
}
}
Ok(found)
}
async fn persist_terminal_locked(
disks: &[DiskStore],
task_id: &str,
terminal: RootHealTerminal,
) -> Result<Option<(DiskStore, EcstoreDiskBytes)>> {
let path = terminal_path(task_id)?;
if let Some((_, bytes)) = Self::find_terminal(disks, task_id).await? {
let current = decode_terminal(task_id, &bytes)?;
if current == terminal {
return Self::find(disks, task_id).await;
}
return Err(Error::Other(format!("Root heal terminal record changed for {task_id}")));
}
let pending = Self::find(disks, task_id).await?;
let disk = pending
.as_ref()
.map(|(disk, _)| disk.clone())
.or_else(|| disks.first().cloned())
.ok_or_else(|| Error::Other("No local disk available for root heal terminal receipt".to_string()))?;
let bytes = serde_json::to_vec(&terminal)
.map_err(|error| Error::Other(format!("Serialize root heal terminal receipt: {error}")))?;
match EcstoreDiskAPI::compare_and_update_file(disk.as_ref(), RUSTFS_META_BUCKET, &path, None, Some(bytes.into())).await? {
EcstoreConditionalFileUpdate::Updated => Ok(pending),
_ => Err(Error::Other(format!("Root heal terminal record changed for {task_id}"))),
}
}
pub(super) async fn persist(&self, request: &HealRequest) -> Result<()> {
if !is_root_heal(&request.heal_type, request.source) {
if !is_admin_heal_recovery(&request.heal_type, request.source) {
return Ok(());
}
let _guard = self.mutation.lock().await;
@@ -199,9 +492,13 @@ impl RootHealRecovery {
}
pub(super) async fn remove(&self, task_id: &str, heal_type: &HealType, source: HealRequestSource) -> Result<bool> {
if !is_root_heal(heal_type, source) {
if !is_admin_heal_recovery(heal_type, source) {
return Ok(false);
}
self.remove_pending_by_id(task_id).await
}
async fn remove_pending_by_id(&self, task_id: &str) -> Result<bool> {
let _guard = self.mutation.lock().await;
let Some((disk, bytes)) = Self::find(&self.disks().await?, task_id).await? else {
return Ok(false);
@@ -221,7 +518,7 @@ impl RootHealRecovery {
}
pub(super) async fn checkpoint_failed_execution(&self, task: &HealTask) -> Result<()> {
if !is_root_heal(&task.heal_type, task.source) {
if !is_admin_heal_recovery(&task.heal_type, task.source) {
return Ok(());
}
let remaining = match task.retry_request_with_remaining_timeout().await {
@@ -235,6 +532,9 @@ impl RootHealRecovery {
return Ok(());
};
let mut intent = decode_intent(&task.id, &expected)?;
if HealType::from(intent.heal_type.clone()) != task.heal_type {
return Err(Error::Other(format!("Root heal recovery owner changed for {}", task.id)));
}
let mut expected_options = intent.options.clone();
expected_options.timeout = task.options.timeout;
if intent.created_at != task.created_at || intent.priority != task.priority || expected_options != task.options {
@@ -268,7 +568,120 @@ impl RootHealRecovery {
if intent_path(task_id).is_err() {
return Ok(false);
}
self.remove(task_id, &HealType::Cluster, HealRequestSource::Admin).await
let _guard = self.mutation.lock().await;
let disks = self.disks().await?;
if Self::find_terminal(&disks, task_id).await?.is_some() {
if let Some((disk, bytes)) = Self::find(&disks, task_id).await? {
match EcstoreDiskAPI::compare_and_update_file(
disk.as_ref(),
RUSTFS_META_BUCKET,
&intent_path(task_id)?,
Some(bytes),
None,
)
.await?
{
EcstoreConditionalFileUpdate::Updated => {}
_ => return Err(Error::Other(format!("Root heal recovery record changed while cancelling {task_id}"))),
}
}
return Ok(true);
}
let Some((disk, bytes)) = Self::find(&disks, task_id).await? else {
return Ok(false);
};
let pending = decode_intent(task_id, &bytes)?;
let heal_type = HealType::from(pending.heal_type);
let terminal = RootHealTerminal::cancelled(task_id, &heal_type);
let _ = Self::persist_terminal_locked(&disks, task_id, terminal).await?;
match EcstoreDiskAPI::compare_and_update_file(
disk.as_ref(),
RUSTFS_META_BUCKET,
&intent_path(task_id)?,
Some(bytes),
None,
)
.await?
{
EcstoreConditionalFileUpdate::Updated => Ok(true),
_ => Err(Error::Other(format!("Root heal recovery record changed while cancelling {task_id}"))),
}
}
pub(super) async fn persist_terminal(
&self,
task_id: &str,
heal_type: &HealType,
source: HealRequestSource,
completed: &CompletedHealStatus,
) -> Result<bool> {
if !is_admin_heal_recovery(heal_type, source) || completed.heal_type != *heal_type {
return Ok(false);
}
let _guard = self.mutation.lock().await;
let disks = self.disks().await?;
let pending =
Self::persist_terminal_locked(&disks, task_id, RootHealTerminal::from_completed(task_id, completed)).await?;
if let Some((disk, bytes)) = pending {
match EcstoreDiskAPI::compare_and_update_file(
disk.as_ref(),
RUSTFS_META_BUCKET,
&intent_path(task_id)?,
Some(bytes),
None,
)
.await?
{
EcstoreConditionalFileUpdate::Updated => {}
_ => {
return Err(Error::Other(format!(
"Root heal recovery record changed while publishing terminal {task_id}"
)));
}
}
}
Ok(true)
}
pub(super) async fn completed(&self, task_id: &str) -> Result<Option<CompletedHealStatus>> {
if terminal_path(task_id).is_err() {
return Ok(None);
}
let _guard = self.mutation.lock().await;
let disks = self.disks().await?;
let Some((_, bytes)) = Self::find_terminal(&disks, task_id).await? else {
return Ok(None);
};
Ok(Some(decode_terminal(task_id, &bytes)?.into_completed()))
}
pub(super) async fn completed_matches_path(&self, heal_path: &str) -> Result<bool> {
let _guard = self.mutation.lock().await;
let disks = self.disks().await?;
for disk in &disks {
EcstoreDiskAPI::stat_volume(disk.as_ref(), RUSTFS_META_BUCKET).await?;
let entries = match EcstoreDiskAPI::list_dir(disk.as_ref(), "", RUSTFS_META_BUCKET, "", -1).await {
Ok(entries) => entries,
Err(DiskError::FileNotFound) => continue,
Err(error) => return Err(Error::Disk(error)),
};
for entry in entries {
let Some(task_id) = entry
.strip_prefix(ROOT_TERMINAL_PREFIX)
.and_then(|entry| entry.strip_suffix(".json"))
else {
continue;
};
let Some((_, bytes)) = Self::find_terminal(&disks, task_id).await? else {
continue;
};
let heal_type = HealType::from(decode_terminal(task_id, &bytes)?.heal_type);
if heal_type_matches_path(&heal_type, heal_path) {
return Ok(true);
}
}
}
Ok(false)
}
pub(super) async fn pending(&self) -> Result<Vec<HealRequest>> {
@@ -295,6 +708,9 @@ impl RootHealRecovery {
}
let mut requests = Vec::new();
for task_id in ids {
if Self::find_terminal(&disks, &task_id).await?.is_some() {
continue;
}
if let Some((_, bytes)) = Self::find(&disks, &task_id).await? {
requests.push(decode_intent(&task_id, &bytes)?.into_request());
}
crates/heal/src/heal/manager/scheduler.rs
+41 -33
View File
@@ -302,38 +302,6 @@ impl HealManager {
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);
if owns_completion
&& result.is_ok()
&& let Err(error) = root_recovery_clone.remove(&task_id, &task.heal_type, task.source).await
{
// Keep the durable responsibility if retirement fails.
// Replaying a completed traversal is idempotent.
warn!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_SCHEDULER_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
task_id,
state = "root_recovery_retirement_failed",
error = %error,
"Failed to retire root heal recovery record"
);
}
if owns_completion
&& result.is_err()
&& let Err(error) = root_recovery_clone.checkpoint_failed_execution(&task).await
{
warn!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_SCHEDULER_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
task_id,
state = "root_recovery_checkpoint_failed",
error = %error,
"Failed to checkpoint root heal recovery execution budget"
);
}
let cancelled_completion = if owns_completion {
false
} else {
@@ -352,7 +320,47 @@ impl HealManager {
completed_status_entry.status = HealTaskStatus::Cancelled;
completed_status_entry.outcome = Some(Arc::new(task.get_outcome().await));
}
let terminal_completion = !matches!(completed_status, HealTaskStatus::Retrying { .. });
let terminal_completion = matches!(
completed_status,
HealTaskStatus::Completed | HealTaskStatus::Cancelled | HealTaskStatus::Failed { .. }
);
if owns_completion
&& terminal_completion
&& root_recovery::is_admin_heal_recovery(&task.heal_type, task.source)
&& let Err(error) = root_recovery_clone
.persist_terminal(&task_id, &task.heal_type, task.source, &completed_status_entry)
.await
{
// Keep the durable responsibility if terminal
// publication fails. Replaying the task is preferable
// to losing the final receipt across restart.
warn!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_SCHEDULER_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
task_id,
state = "root_recovery_terminal_publish_failed",
error = %error,
"Failed to publish heal terminal receipt"
);
}
if owns_completion
&& !terminal_completion
&& result.is_err()
&& let Err(error) = root_recovery_clone.checkpoint_failed_execution(&task).await
{
warn!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_SCHEDULER_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
task_id,
state = "root_recovery_checkpoint_failed",
error = %error,
"Failed to checkpoint root heal recovery execution budget"
);
}
let completed_status_for_verified_events = completed_status_entry.clone();
// Keep retry ownership continuous: status snapshots acquire
// these locks in the same active -> retrying order.
crates/heal/src/heal/manager/tests/root_recovery.rs
+645 -17
View File
@@ -15,6 +15,7 @@
use super::super::root_recovery::RootHealRecovery;
use super::*;
use crate::heal::RUSTFS_META_BUCKET;
use std::collections::HashSet;
async fn recovery_disk() -> (TempDir, DiskStore) {
let temp = TempDir::new().expect("temporary root recovery disk");
@@ -48,7 +49,11 @@ fn recovery_manager(disks: Vec<DiskStore>) -> HealManager {
}
fn root_request() -> HealRequest {
let mut request = HealRequest::new(HealType::Cluster, HealOptions::default(), HealPriority::High);
admin_request(HealType::Cluster)
}
fn admin_request(heal_type: HealType) -> HealRequest {
let mut request = HealRequest::new(heal_type, HealOptions::default(), HealPriority::High);
request.source = HealRequestSource::Admin;
request
}
@@ -106,6 +111,549 @@ async fn root_recovery_shutdown_restart_replays_same_id_and_success_retires_inte
assert!(restarted.root_recovery.pending().await.expect("read completion").is_empty());
}
#[tokio::test]
async fn root_recovery_admin_start_persists_before_shutdown() {
let (_temp, disk) = recovery_disk().await;
let manager = recovery_manager(vec![disk.clone()]);
let mut request = root_request();
request.options.recursive = true;
let receipt = manager
.submit_heal_request_with_receipt(request.clone())
.await
.expect("root admission should persist");
assert_eq!(receipt.result, HealAdmissionResult::Accepted);
assert_eq!(receipt.task_id, request.id);
let pending = manager.root_recovery.pending().await.expect("read durable admission");
assert_eq!(
pending.iter().map(|request| request.id.as_str()).collect::<Vec<_>>(),
[request.id.as_str()]
);
drop(manager);
let restarted = recovery_manager(vec![disk]);
restarted.replay_root_heals().await.expect("replay durable admission");
let queued = restarted
.heal_queue
.lock()
.await
.requests()
.map(|request| request.id.clone())
.collect::<Vec<_>>();
assert_eq!(queued, [request.id]);
}
#[tokio::test]
async fn root_recovery_admin_non_root_types_persist_and_replay() {
for heal_type in [
HealType::Bucket {
bucket: "bucket".to_string(),
},
HealType::Prefix {
bucket: "bucket".to_string(),
prefix: "logs/2026".to_string(),
},
HealType::Object {
bucket: "bucket".to_string(),
object: "object".to_string(),
version_id: Some("version-1".to_string()),
},
HealType::Metadata {
bucket: "bucket".to_string(),
object: "object".to_string(),
},
HealType::ECDecode {
bucket: "bucket".to_string(),
object: "object".to_string(),
version_id: None,
},
HealType::ErasureSet {
buckets: vec!["bucket".to_string()],
set_disk_id: "pool_0_set_1".to_string(),
},
] {
let (_temp, disk) = recovery_disk().await;
let manager = recovery_manager(vec![disk.clone()]);
let mut request = admin_request(heal_type.clone());
request.options.recursive = true;
let receipt = manager
.submit_heal_request_with_receipt(request.clone())
.await
.expect("admin heal admission should persist");
assert_eq!(receipt.result, HealAdmissionResult::Accepted);
assert_eq!(
manager
.root_recovery
.pending()
.await
.expect("read durable admin state")
.iter()
.map(|request| (&request.id, &request.heal_type))
.collect::<Vec<_>>(),
[(&request.id, &request.heal_type)]
);
drop(manager);
let restarted = recovery_manager(vec![disk]);
restarted.replay_root_heals().await.expect("replay durable admin heal");
let queued = restarted.heal_queue.lock().await.requests().cloned().collect::<Vec<_>>();
assert_eq!(queued.len(), 1);
assert_eq!(queued[0].id, request.id);
assert_eq!(queued[0].heal_type, request.heal_type);
assert_eq!(queued[0].options, request.options);
assert_eq!(queued[0].source, HealRequestSource::Admin);
}
}
#[tokio::test]
async fn root_recovery_non_admin_request_is_not_persisted() {
let (_temp, disk) = recovery_disk().await;
let manager = recovery_manager(vec![disk.clone()]);
let mut request = HealRequest::new(
HealType::Bucket {
bucket: "scanner".to_string(),
},
HealOptions::default(),
HealPriority::Low,
);
request.source = HealRequestSource::Scanner;
assert_eq!(
manager
.submit_heal_request(request)
.await
.expect("scanner request should still queue"),
HealAdmissionResult::Accepted
);
assert!(manager.root_recovery.pending().await.expect("read durable state").is_empty());
drop(manager);
let restarted = recovery_manager(vec![disk]);
restarted.replay_root_heals().await.expect("replay empty durable state");
assert_eq!(restarted.get_queue_length().await, 0);
}
#[tokio::test]
async fn root_recovery_path_cancel_covers_durable_only_non_root_record() {
let (_temp, disk) = recovery_disk().await;
let manager = recovery_manager(vec![disk.clone()]);
let request = admin_request(HealType::Bucket {
bucket: "bucket".to_string(),
});
manager
.root_recovery
.persist(&request)
.await
.expect("durable bucket responsibility");
assert_eq!(
manager
.cancel_tasks_for_path("bucket")
.await
.expect("cancel durable-only bucket path"),
1
);
drop(manager);
let restarted = recovery_manager(vec![disk]);
restarted
.replay_root_heals()
.await
.expect("restart after durable path cancellation");
assert_eq!(restarted.get_queue_length().await, 0);
assert_eq!(
restarted
.get_task_status_for_path("bucket", &request.id)
.await
.expect("durable cancellation remains queryable by path"),
HealTaskStatus::Cancelled
);
assert_eq!(
restarted
.get_task_status(&request.id)
.await
.expect("durable cancellation remains queryable by id"),
HealTaskStatus::Cancelled
);
}
#[tokio::test]
async fn root_recovery_active_cancel_is_queryable_after_restart_for_scoped_admin() {
let (_temp, disk) = recovery_disk().await;
let manager = recovery_manager(vec![disk.clone()]);
let request = admin_request(HealType::Bucket {
bucket: "bucket".to_string(),
});
active_root(&manager, request.clone()).await;
manager
.root_recovery
.persist(&request)
.await
.expect("durable active bucket responsibility");
manager.cancel_task(&request.id).await.expect("cancel active bucket");
assert!(
manager
.root_recovery
.pending()
.await
.expect("active terminal retires intent")
.is_empty()
);
drop(manager);
let restarted = recovery_manager(vec![disk]);
restarted
.replay_root_heals()
.await
.expect("restart after active cancellation");
assert_eq!(restarted.get_queue_length().await, 0);
assert_eq!(
restarted
.get_task_status(&request.id)
.await
.expect("active cancellation remains queryable by id"),
HealTaskStatus::Cancelled
);
assert_eq!(
restarted
.get_task_status_for_path("bucket", &request.id)
.await
.expect("active cancellation remains queryable by path"),
HealTaskStatus::Cancelled
);
}
#[tokio::test]
async fn root_recovery_terminal_receipt_wins_over_stale_pending_scoped_intent_after_restart() {
let (_temp, disk) = recovery_disk().await;
let manager = recovery_manager(vec![disk.clone()]);
let request = admin_request(HealType::Bucket {
bucket: "bucket".to_string(),
});
manager
.root_recovery
.persist(&request)
.await
.expect("durable bucket responsibility");
manager
.publish_admin_cancelled_terminal(&request.id, &request.heal_type, request.source)
.await
.expect("publish terminal receipt");
manager
.root_recovery
.persist(&request)
.await
.expect("restore stale pending intent after terminal publication");
assert!(
manager
.root_recovery
.pending()
.await
.expect("terminal masks stale pending")
.is_empty()
);
drop(manager);
let restarted = recovery_manager(vec![disk]);
restarted
.replay_root_heals()
.await
.expect("restart with terminal and stale pending");
assert_eq!(restarted.get_queue_length().await, 0);
assert_eq!(
restarted
.get_task_status(&request.id)
.await
.expect("terminal status survives stale pending"),
HealTaskStatus::Cancelled
);
}
#[tokio::test]
async fn root_recovery_completed_non_root_admin_is_queryable_after_restart() {
let (_temp, disk) = recovery_disk().await;
let manager = recovery_manager(vec![disk.clone()]);
let request = admin_request(HealType::Bucket {
bucket: "bucket".to_string(),
});
manager
.root_recovery
.persist(&request)
.await
.expect("durable bucket responsibility");
let completed = CompletedHealStatus {
outcome: None,
heal_type: request.heal_type.clone(),
status: HealTaskStatus::Completed,
progress: Some(HealProgress {
objects_scanned: 2,
objects_healed: 2,
bytes_processed: 128,
..Default::default()
}),
retained_bytes: std::sync::OnceLock::new(),
result_items_truncated: false,
completed_at: SystemTime::now(),
seqed_items: Vec::new(),
next_seq: 0,
min_seq: 0,
};
assert!(
manager
.publish_admin_terminal(&request.id, &request.heal_type, request.source, &completed)
.await
.expect("publish completed terminal")
);
assert!(
manager
.root_recovery
.pending()
.await
.expect("completed terminal retires intent")
.is_empty()
);
drop(manager);
let restarted = recovery_manager(vec![disk]);
restarted.replay_root_heals().await.expect("restart after completed terminal");
assert_eq!(restarted.get_queue_length().await, 0);
assert_eq!(
restarted
.get_task_status_for_path("bucket", &request.id)
.await
.expect("completed terminal remains queryable by path"),
HealTaskStatus::Completed
);
let progress = restarted
.get_task_progress(&request.id)
.await
.expect("completed terminal exposes progress");
assert_eq!(progress.objects_scanned, 2);
assert_eq!(progress.objects_healed, 2);
}
#[tokio::test]
async fn root_recovery_admin_start_fails_closed_when_owner_is_unavailable() {
let (_temp, disk) = recovery_disk().await;
let (unavailable_temp, unavailable) = recovery_disk().await;
std::fs::remove_dir_all(unavailable_temp.path().join(RUSTFS_META_BUCKET)).expect("make owner volume unavailable");
let manager = recovery_manager(vec![unavailable, disk.clone()]);
let request = admin_request(HealType::Bucket {
bucket: "bucket".to_string(),
});
assert!(
manager.submit_heal_request_with_receipt(request).await.is_err(),
"admin admission must fail closed when the durable owner cannot be checked"
);
assert_eq!(manager.get_queue_length().await, 0);
assert!(
RootHealRecovery::with_disks(vec![disk])
.pending()
.await
.expect("other disk remains empty")
.is_empty()
);
}
#[tokio::test]
async fn root_recovery_force_start_cancels_only_overlapping_durable_admin_records() {
let (_temp, disk) = recovery_disk().await;
let manager = recovery_manager(vec![disk.clone()]);
let old = admin_request(HealType::Bucket {
bucket: "bucket-a".to_string(),
});
let disjoint = admin_request(HealType::Bucket {
bucket: "bucket-b".to_string(),
});
manager.root_recovery.persist(&old).await.expect("old bucket owner");
manager.root_recovery.persist(&disjoint).await.expect("disjoint bucket owner");
let mut replacement = admin_request(HealType::Prefix {
bucket: "bucket-a".to_string(),
prefix: "logs/".to_string(),
});
replacement.force_start = true;
assert_eq!(
manager
.submit_heal_request(replacement.clone())
.await
.expect("forceStart should replace only the overlapping durable owner"),
HealAdmissionResult::Accepted
);
let mut pending = manager
.root_recovery
.pending()
.await
.expect("read durable owners")
.into_iter()
.map(|request| (request.id, request.heal_type))
.collect::<Vec<_>>();
pending.sort_by(|left, right| left.0.cmp(&right.0));
let mut expected = vec![
(disjoint.id.clone(), disjoint.heal_type.clone()),
(replacement.id.clone(), replacement.heal_type.clone()),
];
expected.sort_by(|left, right| left.0.cmp(&right.0));
assert_eq!(pending, expected);
drop(manager);
let restarted = recovery_manager(vec![disk]);
restarted.replay_root_heals().await.expect("replay surviving owners");
let queued_ids = restarted
.heal_queue
.lock()
.await
.requests()
.map(|request| request.id.clone())
.collect::<HashSet<_>>();
assert_eq!(queued_ids, HashSet::from([disjoint.id, replacement.id]));
}
#[tokio::test]
async fn root_recovery_queued_non_root_admin_owner_is_not_priority_displaced() {
let (_temp, disk) = recovery_disk().await;
let manager = recovery_manager(vec![disk.clone()]);
manager.config.write().await.queue_size = 1;
let mut durable = admin_request(HealType::Bucket {
bucket: "bucket".to_string(),
});
durable.priority = HealPriority::Low;
assert_eq!(
manager
.submit_heal_request(durable.clone())
.await
.expect("low-priority admin bucket should queue durably"),
HealAdmissionResult::Accepted
);
let mut urgent = HealRequest::new(
HealType::Object {
bucket: "other".to_string(),
object: "object".to_string(),
version_id: None,
},
HealOptions::default(),
HealPriority::Urgent,
);
urgent.source = HealRequestSource::Internal;
assert_eq!(
manager
.submit_heal_request(urgent)
.await
.expect("durable admin owner cannot be displaced"),
HealAdmissionResult::Full
);
assert_eq!(
manager
.root_recovery
.pending()
.await
.expect("read durable bucket")
.iter()
.map(|request| request.id.as_str())
.collect::<Vec<_>>(),
[durable.id.as_str()]
);
}
#[tokio::test]
async fn root_recovery_legacy_schema_replays_as_cluster() {
#[derive(serde::Serialize)]
struct LegacyRootHealIntent<'a> {
schema: u32,
task_id: &'a str,
options: &'a HealOptions,
priority: HealPriority,
retry_attempts: u32,
created_at: SystemTime,
}
let (_temp, disk) = recovery_disk().await;
let request = root_request();
let path = format!("root-heal-{}.json", request.id);
let bytes = serde_json::to_vec(&LegacyRootHealIntent {
schema: 1,
task_id: &request.id,
options: &request.options,
priority: request.priority,
retry_attempts: request.retry_attempts,
created_at: request.created_at,
})
.expect("legacy root recovery JSON");
disk.write_all(RUSTFS_META_BUCKET, &path, bytes.into())
.await
.expect("write legacy root record");
let restarted = recovery_manager(vec![disk]);
restarted.replay_root_heals().await.expect("replay legacy root record");
let queued = restarted.heal_queue.lock().await.requests().cloned().collect::<Vec<_>>();
assert_eq!(queued.len(), 1);
assert_eq!(queued[0].id, request.id);
assert_eq!(queued[0].heal_type, HealType::Cluster);
}
#[tokio::test]
async fn root_recovery_rejected_admin_start_does_not_persist() {
let (_temp, disk) = recovery_disk().await;
let manager = recovery_manager(vec![disk.clone()]);
manager.config.write().await.queue_size = 0;
let request = root_request();
let receipt = manager
.submit_heal_request_with_receipt(request.clone())
.await
.expect("full admission reports a receipt");
assert_eq!(receipt.result, HealAdmissionResult::Full);
assert!(manager.root_recovery.pending().await.expect("read durable state").is_empty());
drop(manager);
let restarted = recovery_manager(vec![disk]);
restarted.replay_root_heals().await.expect("replay empty durable state");
assert_eq!(restarted.get_queue_length().await, 0);
}
#[tokio::test]
async fn root_recovery_queued_owner_is_not_priority_displaced() {
let (_temp, disk) = recovery_disk().await;
let manager = recovery_manager(vec![disk.clone()]);
manager.config.write().await.queue_size = 1;
let mut root = root_request();
root.priority = HealPriority::Low;
assert_eq!(
manager
.submit_heal_request(root.clone())
.await
.expect("low-priority root should queue"),
HealAdmissionResult::Accepted
);
let mut bucket = HealRequest::new(
HealType::Bucket {
bucket: "bucket".to_string(),
},
HealOptions::default(),
HealPriority::Urgent,
);
bucket.source = HealRequestSource::Admin;
assert_eq!(
manager
.submit_heal_request(bucket)
.await
.expect("durable root owner cannot be displaced"),
HealAdmissionResult::Full
);
let pending = manager.root_recovery.pending().await.expect("read durable root");
assert_eq!(pending.iter().map(|request| request.id.as_str()).collect::<Vec<_>>(), [root.id.as_str()]);
let queued = manager
.heal_queue
.lock()
.await
.requests()
.map(|request| request.id.clone())
.collect::<Vec<_>>();
assert_eq!(queued, [root.id]);
}
#[tokio::test]
async fn root_recovery_explicit_cancel_covers_active_queued_retrying_and_durable_only() {
for state in ["active", "queued", "retrying", "durable_only", "root_path"] {
@@ -159,7 +707,17 @@ async fn root_recovery_force_start_cancels_durable_only_responsibility() {
.expect("force start replacement"),
HealAdmissionResult::Accepted
);
assert!(manager.root_recovery.pending().await.expect("old owner retired").is_empty());
assert_eq!(
manager
.root_recovery
.pending()
.await
.expect("new owner retained")
.iter()
.map(|request| request.id.as_str())
.collect::<Vec<_>>(),
[new.id.as_str()]
);
manager.stop().await.expect("persist new root only");
let restarted = recovery_manager(vec![disk]);
restarted.replay_root_heals().await.expect("restart replacement");
@@ -173,6 +731,58 @@ async fn root_recovery_force_start_cancels_durable_only_responsibility() {
assert_eq!(ids, [new.id]);
}
#[tokio::test]
async fn root_recovery_force_start_preserves_disjoint_durable_only_admin_work() {
let (_temp, disk) = recovery_disk().await;
let manager = recovery_manager(vec![disk.clone()]);
let old_overlap = admin_request(HealType::Bucket {
bucket: "overlap".to_string(),
});
let old_disjoint = admin_request(HealType::Bucket {
bucket: "disjoint".to_string(),
});
manager
.root_recovery
.persist(&old_overlap)
.await
.expect("durable overlapping bucket");
manager
.root_recovery
.persist(&old_disjoint)
.await
.expect("durable disjoint bucket");
let mut replacement = admin_request(HealType::Bucket {
bucket: "overlap".to_string(),
});
replacement.force_start = true;
assert_eq!(
manager
.submit_heal_request(replacement.clone())
.await
.expect("forceStart replaces overlapping durable owner"),
HealAdmissionResult::Accepted
);
drop(manager);
let restarted = recovery_manager(vec![disk]);
restarted
.replay_root_heals()
.await
.expect("restart after selective forceStart");
let mut ids = restarted
.heal_queue
.lock()
.await
.requests()
.map(|request| request.id.clone())
.collect::<Vec<_>>();
ids.sort();
let mut expected = vec![old_disjoint.id, replacement.id];
expected.sort();
assert_eq!(ids, expected);
}
#[tokio::test]
async fn root_recovery_force_start_replaces_fresh_queued_and_retrying_admin_roots() {
for retrying in [false, true] {
@@ -184,7 +794,28 @@ async fn root_recovery_force_start_replaces_fresh_queued_and_retrying_admin_root
} else {
manager.submit_heal_request(old.clone()).await.expect("queue original root");
}
assert!(manager.root_recovery.pending().await.expect("not handed off yet").is_empty());
if retrying {
assert!(
manager
.root_recovery
.pending()
.await
.expect("retrying not handed off yet")
.is_empty()
);
} else {
assert_eq!(
manager
.root_recovery
.pending()
.await
.expect("queued root is durable immediately")
.iter()
.map(|request| request.id.as_str())
.collect::<Vec<_>>(),
[old.id.as_str()]
);
}
let mut new = root_request();
new.force_start = true;
assert_eq!(
@@ -222,7 +853,7 @@ async fn root_recovery_invalid_records_are_retained_without_partial_replay() {
let original = disk.read_all(RUSTFS_META_BUCKET, &path).await.expect("read root record");
let mut value: serde_json::Value = serde_json::from_slice(&original).expect("record JSON");
match kind {
"schema" => value["schema"] = 2.into(),
"schema" => value["schema"] = 3.into(),
"identity" => value["task_id"] = valid.id.clone().into(),
"option" => value["options"]["future_delete_mode"] = true.into(),
"no_lock" => value["options"]["no_lock"] = true.into(),
@@ -448,18 +1079,15 @@ async fn root_recovery_terminal_timeout_updates_only_existing_journal_before_sec
.await
.expect("second restart after terminal timeout");
let queue = restarted.heal_queue.lock().await;
if durable {
assert_eq!(
queue
.requests()
.next()
.expect("remaining timeout responsibility")
.options
.timeout,
Some(Duration::ZERO)
);
} else {
assert!(queue.is_empty(), "terminal failure must not create a new durable responsibility");
}
assert_eq!(
queue
.requests()
.next()
.expect("remaining timeout responsibility")
.options
.timeout,
Some(Duration::ZERO),
"durable={durable}"
);
}
}