From c02143d334a3cd44586add8878316316beda382b Mon Sep 17 00:00:00 2001 From: houseme Date: Fri, 24 Jul 2026 12:13:30 +0800 Subject: [PATCH] feat(ilm): report manual transition backfill outcomes Add scoped lifecycle transition backfill reporting for backlog #1478 and expose enqueue outcomes needed by #1479 without changing the existing scanner/compensation bool API. Co-Authored-By: heihutu --- crates/ecstore/src/api/mod.rs | 10 +- .../bucket/lifecycle/bucket_lifecycle_ops.rs | 272 ++++++++++++++++-- 2 files changed, 261 insertions(+), 21 deletions(-) diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index c7f186c91..4f707c4eb 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -43,10 +43,12 @@ pub mod bucket { pub mod bucket_lifecycle_ops { pub use crate::bucket::lifecycle::bucket_lifecycle_ops::{ - ExpiryState, LifecycleOps, RestoreRequestOps, TransitionState, TransitionedObject, apply_expiry_rule, - apply_transition_rule, enqueue_expiry_for_existing_objects, enqueue_transition_for_existing_objects, - enqueue_transition_immediate, expire_transitioned_object, get_global_expiry_state, get_global_transition_state, - init_background_expiry, post_restore_opts, run_stale_multipart_upload_cleanup_once, validate_transition_tier, + ExpiryState, LifecycleOps, ManualTransitionRunOptions, ManualTransitionRunReport, RestoreRequestOps, + TransitionState, TransitionedObject, apply_expiry_rule, apply_transition_rule, + enqueue_expiry_for_existing_objects, enqueue_transition_for_existing_objects, + enqueue_transition_for_existing_objects_scoped, enqueue_transition_immediate, expire_transitioned_object, + get_global_expiry_state, get_global_transition_state, init_background_expiry, post_restore_opts, + run_stale_multipart_upload_cleanup_once, validate_transition_tier, }; } diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index f9ef2e1cc..276ceda3d 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -991,6 +991,21 @@ enum ImmediateEnqueueFailure { QueueSendTimedOut { timeout_ms: u64 }, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum TransitionEnqueueOutcome { + Queued, + AlreadyInFlight, + QueueFull, + QueueClosed, + QueueSendTimedOut, +} + +impl TransitionEnqueueOutcome { + fn is_handled(self) -> bool { + matches!(self, Self::Queued | Self::AlreadyInFlight) + } +} + impl TransitionState { #[allow(clippy::new_ret_no_self)] pub fn new() -> Arc { @@ -1213,12 +1228,17 @@ impl TransitionState { } } - pub async fn queue_transition_task(self: &Arc, oi: &ObjectInfo, event: &lifecycle::Event, src: &LcEventSrc) -> bool { + async fn queue_transition_task_outcome( + self: &Arc, + oi: &ObjectInfo, + event: &lifecycle::Event, + src: &LcEventSrc, + ) -> TransitionEnqueueOutcome { if is_immediate_transition_source(src) && should_force_immediate_transition_enqueue_timeout() { self.handle_immediate_enqueue_failure(oi, src, ImmediateEnqueueFailure::ForcedTimeout); record_scanner_transition_enqueue_result(src, 1, false); self.record_scanner_transition_state(); - return false; + return TransitionEnqueueOutcome::QueueSendTimedOut; } // Deduplicate concurrent enqueues of the same object version. The claim is @@ -1227,7 +1247,7 @@ impl TransitionState { // counters (the first enqueue already counted this object). if !self.reserve_transition(oi) { self.record_scanner_transition_state(); - return true; + return TransitionEnqueueOutcome::AlreadyInFlight; } let task = TransitionTask { @@ -1237,6 +1257,7 @@ impl TransitionState { }; let mut queued = false; if is_immediate_transition_source(src) { + let mut failure = None; match self.transition_tx.try_send(Some(task)) { Ok(()) => queued = true, Err(async_channel::TrySendError::Full(task)) => { @@ -1252,6 +1273,7 @@ impl TransitionState { timeout_ms: Some(send_timeout.as_millis() as u64), }, ); + failure = Some(TransitionEnqueueOutcome::QueueClosed); } Err(_) => { self.handle_immediate_enqueue_failure( @@ -1261,11 +1283,13 @@ impl TransitionState { timeout_ms: send_timeout.as_millis() as u64, }, ); + failure = Some(TransitionEnqueueOutcome::QueueSendTimedOut); } } } Err(async_channel::TrySendError::Closed(_task)) => { self.handle_immediate_enqueue_failure(oi, src, ImmediateEnqueueFailure::QueueClosed { timeout_ms: None }); + failure = Some(TransitionEnqueueOutcome::QueueClosed); } } if !queued { @@ -1273,13 +1297,19 @@ impl TransitionState { } record_scanner_transition_enqueue_result(src, 1, queued); self.record_scanner_transition_state(); - return queued; + return if queued { + TransitionEnqueueOutcome::Queued + } else { + failure.unwrap_or(TransitionEnqueueOutcome::QueueFull) + }; } + let mut failure = None; match self.transition_tx.try_send(Some(task)) { Ok(()) => queued = true, Err(async_channel::TrySendError::Full(_)) => { Self::inc_counter(&self.queue_full_tasks); + failure = Some(TransitionEnqueueOutcome::QueueFull); debug!( bucket = %oi.bucket, object = %oi.name, @@ -1298,6 +1328,7 @@ impl TransitionState { state = "queue_closed", "transition enqueue failed because the queue is closed" ); + failure = Some(TransitionEnqueueOutcome::QueueClosed); } } if !queued { @@ -1305,7 +1336,15 @@ impl TransitionState { } record_scanner_transition_enqueue_result(src, 1, queued); self.record_scanner_transition_state(); - queued + if queued { + TransitionEnqueueOutcome::Queued + } else { + failure.unwrap_or(TransitionEnqueueOutcome::QueueFull) + } + } + + pub async fn queue_transition_task(self: &Arc, oi: &ObjectInfo, event: &lifecycle::Event, src: &LcEventSrc) -> bool { + self.queue_transition_task_outcome(oi, event, src).await.is_handled() } pub async fn init(api: Arc) { @@ -2402,10 +2441,72 @@ pub async fn enqueue_immediate_expiry(oi: &ObjectInfo, src: LcEventSrc) { } } +#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize)] +pub struct ManualTransitionRunOptions { + pub prefix: String, + pub tier: Option, + pub dry_run: bool, + pub max_objects: Option, +} + +#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize)] +pub struct ManualTransitionRunReport { + pub bucket: String, + pub prefix: String, + pub tier: Option, + pub dry_run: bool, + pub lifecycle_config_found: bool, + pub scanned: u64, + pub eligible: u64, + pub enqueued: u64, + pub dry_run_eligible: u64, + pub skipped_not_transition: u64, + pub skipped_tier: u64, + pub skipped_delete_marker: u64, + pub skipped_directory: u64, + pub skipped_replication: u64, + pub skipped_already_in_flight: u64, + pub skipped_queue_full: u64, + pub skipped_queue_closed: u64, + pub skipped_queue_timeout: u64, + pub truncated_by_limit: bool, + pub next_marker: Option, + pub next_version_idmarker: Option, +} + +impl ManualTransitionRunReport { + fn new(bucket: &str, options: &ManualTransitionRunOptions) -> Self { + Self { + bucket: bucket.to_string(), + prefix: options.prefix.clone(), + tier: options.tier.clone(), + dry_run: options.dry_run, + ..Default::default() + } + } + + pub fn has_partial_enqueue(&self) -> bool { + self.skipped_queue_full > 0 || self.skipped_queue_closed > 0 || self.skipped_queue_timeout > 0 + } +} + pub async fn enqueue_transition_for_existing_objects(api: Arc, bucket: &str) -> Result<(), Error> { + let _ = enqueue_transition_for_existing_objects_scoped(api, bucket, ManualTransitionRunOptions::default()).await?; + Ok(()) +} + +pub async fn enqueue_transition_for_existing_objects_scoped( + api: Arc, + bucket: &str, + options: ManualTransitionRunOptions, +) -> Result { + const LIST_PAGE_SIZE: i32 = 1000; + + let mut report = ManualTransitionRunReport::new(bucket, &options); let Some(lc) = runtime_sources::bucket_lifecycle_config(bucket).await else { - return Ok(()); + return Ok(report); }; + report.lifecycle_config_found = true; let mut marker = None; let mut version_marker = None; let src = LcEventSrc::Scanner; @@ -2413,15 +2514,22 @@ pub async fn enqueue_transition_for_existing_objects(api: Arc, bucket: loop { let page = api .clone() - .list_object_versions(bucket, "", marker.clone(), version_marker.clone(), None, 1000) + .list_object_versions(bucket, &options.prefix, marker.clone(), version_marker.clone(), None, LIST_PAGE_SIZE) .await?; for object in &page.objects { - enqueue_transition_with_lifecycle(object, &lc, &src).await; + report.scanned = report.scanned.saturating_add(1); + enqueue_transition_with_lifecycle_report(object, &lc, &src, &options, &mut report).await; + if options.max_objects.is_some_and(|max_objects| report.scanned >= max_objects) { + report.truncated_by_limit = true; + report.next_marker = Some(object.name.clone()); + report.next_version_idmarker = object.version_id.map(|version| version.to_string()); + return Ok(report); + } } if !page.is_truncated { - return Ok(()); + return Ok(report); } marker = page.next_marker; @@ -2615,25 +2723,72 @@ pub async fn enqueue_expiry_for_existing_objects(api: Arc, bucket: &str } } -async fn enqueue_transition_with_lifecycle(oi: &ObjectInfo, lc: &BucketLifecycleConfiguration, src: &LcEventSrc) -> bool { +async fn enqueue_transition_with_lifecycle_report( + oi: &ObjectInfo, + lc: &BucketLifecycleConfiguration, + src: &LcEventSrc, + options: &ManualTransitionRunOptions, + report: &mut ManualTransitionRunReport, +) -> bool { let event = lc.eval(&oi.to_lifecycle_opts()).await; match event.action { IlmAction::TransitionAction | IlmAction::TransitionVersionAction => { if oi.delete_marker || oi.is_dir { + if oi.delete_marker { + report.skipped_delete_marker = report.skipped_delete_marker.saturating_add(1); + } else { + report.skipped_directory = report.skipped_directory.saturating_add(1); + } return false; } if lifecycle_action_blocked_by_replication(event.action, oi) { + report.skipped_replication = report.skipped_replication.saturating_add(1); return false; } - return runtime_sources::transition_state_handle() - .queue_transition_task(oi, &event, src) + if options + .tier + .as_deref() + .is_some_and(|tier| !event.storage_class.eq_ignore_ascii_case(tier)) + { + report.skipped_tier = report.skipped_tier.saturating_add(1); + return false; + } + report.eligible = report.eligible.saturating_add(1); + if options.dry_run { + report.dry_run_eligible = report.dry_run_eligible.saturating_add(1); + return true; + } + let outcome = runtime_sources::transition_state_handle() + .queue_transition_task_outcome(oi, &event, src) .await; + match outcome { + TransitionEnqueueOutcome::Queued => report.enqueued = report.enqueued.saturating_add(1), + TransitionEnqueueOutcome::AlreadyInFlight => { + report.skipped_already_in_flight = report.skipped_already_in_flight.saturating_add(1); + } + TransitionEnqueueOutcome::QueueFull => { + report.skipped_queue_full = report.skipped_queue_full.saturating_add(1); + } + TransitionEnqueueOutcome::QueueClosed => { + report.skipped_queue_closed = report.skipped_queue_closed.saturating_add(1); + } + TransitionEnqueueOutcome::QueueSendTimedOut => { + report.skipped_queue_timeout = report.skipped_queue_timeout.saturating_add(1); + } + } + return outcome.is_handled(); } - _ => (), + _ => report.skipped_not_transition = report.skipped_not_transition.saturating_add(1), } false } +async fn enqueue_transition_with_lifecycle(oi: &ObjectInfo, lc: &BucketLifecycleConfiguration, src: &LcEventSrc) -> bool { + let options = ManualTransitionRunOptions::default(); + let mut report = ManualTransitionRunReport::new(&oi.bucket, &options); + enqueue_transition_with_lifecycle_report(oi, lc, src, &options, &mut report).await +} + /// Build the delete options for a lifecycle expiry event on a transitioned /// object. Versioned events target the exact version; restore-expiry events /// (`DeleteRestoredAction`/`DeleteRestoredVersionAction`) set @@ -3459,10 +3614,11 @@ pub async fn apply_lifecycle_action(event: &lifecycle::Event, src: &LcEventSrc, mod tests { use super::{ DATE_EXPIRY_EXISTING_OBJECTS_GRACE_SECS, DEFAULT_TRANSITION_QUEUE_CAPACITY, DEFAULT_TRANSITION_WORKERS_ABSOLUTE_MAX, - DEFAULT_TRANSITION_WORKERS_CAP, ExpiryState, FreeVersionTask, StaleMultipartUploadCandidate, - TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL, TIER_FREE_VERSION_RECOVERY_MAX_IDLE_INTERVAL, TierFreeVersionRecoverySchedule, - TransitionState, TransitionedObject, VersionReplicationScan, cleanup_empty_multipart_sha_dirs_on_local_disks, - cleanup_stale_multipart_uploads_once_at, enqueue_recovered_free_version_with_state, enqueue_transition_with_lifecycle, + DEFAULT_TRANSITION_WORKERS_CAP, ExpiryState, FreeVersionTask, ManualTransitionRunOptions, ManualTransitionRunReport, + StaleMultipartUploadCandidate, TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL, TIER_FREE_VERSION_RECOVERY_MAX_IDLE_INTERVAL, + TierFreeVersionRecoverySchedule, TransitionEnqueueOutcome, TransitionState, TransitionedObject, VersionReplicationScan, + cleanup_empty_multipart_sha_dirs_on_local_disks, cleanup_stale_multipart_uploads_once_at, + enqueue_recovered_free_version_with_state, enqueue_transition_with_lifecycle, enqueue_transition_with_lifecycle_report, eval_action_from_lifecycle, jitter_tier_free_version_recovery_delay, lifecycle_action_blocked_by_replication, lifecycle_delete_all_versions_replication_scan, lifecycle_deleted_object, lifecycle_replication_blocks_action, lifecycle_rule_has_date_expiration, lifecycle_version_purge_state_from_completed_targets, @@ -5594,6 +5750,32 @@ mod tests { assert_eq!(state.transition_rx.len(), 1, "only one task must actually be queued"); } + #[tokio::test] + #[serial] + async fn queue_transition_task_outcome_reports_duplicate_separately() { + let state = TransitionState::new_with_capacity(4); + let object = ObjectInfo { + bucket: "bucket".to_string(), + name: "object".to_string(), + ..Default::default() + }; + let event = crate::bucket::lifecycle::lifecycle::Event { + action: IlmAction::TransitionAction, + ..Default::default() + }; + + let first = state + .queue_transition_task_outcome(&object, &event, &LcEventSrc::Scanner) + .await; + let second = state + .queue_transition_task_outcome(&object, &event, &LcEventSrc::Scanner) + .await; + + assert_eq!(first, TransitionEnqueueOutcome::Queued); + assert_eq!(second, TransitionEnqueueOutcome::AlreadyInFlight); + assert_eq!(state.transition_rx.len(), 1); + } + #[tokio::test] #[serial] async fn queue_transition_task_dedupes_immediate_and_scanner_sources_for_same_version() { @@ -6031,6 +6213,62 @@ mod tests { assert!(!queued); } + #[tokio::test] + async fn manual_transition_dry_run_counts_due_object_without_enqueue() { + let lc = latest_transition_lifecycle(); + let object = current_object(ReplicationStatusType::Completed); + let options = ManualTransitionRunOptions { + dry_run: true, + ..Default::default() + }; + let mut report = ManualTransitionRunReport::new(&object.bucket, &options); + + let handled = enqueue_transition_with_lifecycle_report(&object, &lc, &LcEventSrc::Scanner, &options, &mut report).await; + + assert!(handled); + assert_eq!(report.eligible, 1); + assert_eq!(report.dry_run_eligible, 1); + assert_eq!(report.enqueued, 0); + } + + #[tokio::test] + async fn manual_transition_respects_not_yet_due_lifecycle_rule() { + let lc = latest_transition_lifecycle(); + let object = ObjectInfo { + mod_time: Some(OffsetDateTime::now_utc()), + ..current_object(ReplicationStatusType::Completed) + }; + let options = ManualTransitionRunOptions { + dry_run: true, + ..Default::default() + }; + let mut report = ManualTransitionRunReport::new(&object.bucket, &options); + + let handled = enqueue_transition_with_lifecycle_report(&object, &lc, &LcEventSrc::Scanner, &options, &mut report).await; + + assert!(!handled); + assert_eq!(report.eligible, 0); + assert_eq!(report.skipped_not_transition, 1); + } + + #[tokio::test] + async fn manual_transition_tier_filter_skips_non_matching_transition() { + let lc = latest_transition_lifecycle(); + let object = current_object(ReplicationStatusType::Completed); + let options = ManualTransitionRunOptions { + tier: Some("COLD".to_string()), + dry_run: true, + ..Default::default() + }; + let mut report = ManualTransitionRunReport::new(&object.bucket, &options); + + let handled = enqueue_transition_with_lifecycle_report(&object, &lc, &LcEventSrc::Scanner, &options, &mut report).await; + + assert!(!handled); + assert_eq!(report.eligible, 0); + assert_eq!(report.skipped_tier, 1); + } + #[tokio::test] async fn existing_object_lifecycle_allows_expired_marker_after_replication_completed() { let lc = expired_delete_marker_lifecycle();