diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 6381bf837..640c4779a 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -1794,6 +1794,7 @@ fn spawn_manual_transition_job_recovery_once(api: Arc) -> Option, @@ -1833,6 +1835,7 @@ pub struct ManualTransitionJobRecoveryStats { enum ManualTransitionJobRecoveryOutcome { Resumed, Cancelled, + Unknown, Skipped, } @@ -1845,6 +1848,7 @@ async fn recover_manual_transition_jobs(api: Arc, limit: usize) -> Resu total.scanned = total.scanned.saturating_add(stats.scanned); total.resumed = total.resumed.saturating_add(stats.resumed); total.cancelled = total.cancelled.saturating_add(stats.cancelled); + total.unknown = total.unknown.saturating_add(stats.unknown); total.skipped = total.skipped.saturating_add(stats.skipped); total.failed = total.failed.saturating_add(stats.failed); @@ -1908,9 +1912,10 @@ pub async fn recover_manual_transition_jobs_once( continue; } }; - match recover_manual_transition_job(api.clone(), job_id).await { + match recover_manual_transition_job(api.clone(), job_id, manual_transition_queue_snapshot()).await { Ok(ManualTransitionJobRecoveryOutcome::Resumed) => stats.resumed = stats.resumed.saturating_add(1), Ok(ManualTransitionJobRecoveryOutcome::Cancelled) => stats.cancelled = stats.cancelled.saturating_add(1), + Ok(ManualTransitionJobRecoveryOutcome::Unknown) => stats.unknown = stats.unknown.saturating_add(1), Ok(ManualTransitionJobRecoveryOutcome::Skipped) => stats.skipped = stats.skipped.saturating_add(1), Err(err) => { stats.failed = stats.failed.saturating_add(1); @@ -1930,7 +1935,11 @@ pub async fn recover_manual_transition_jobs_once( Ok(stats) } -async fn recover_manual_transition_job(api: Arc, job_id: Uuid) -> Result { +async fn recover_manual_transition_job( + api: Arc, + job_id: Uuid, + queue_snapshot: ManualTransitionQueueSnapshot, +) -> Result { let (mut record, etag) = match load_manual_transition_job_record_with_etag(api.clone(), job_id).await { Ok(record) => record, Err(Error::ConfigNotFound) => return Ok(ManualTransitionJobRecoveryOutcome::Skipped), @@ -1940,10 +1949,22 @@ async fn recover_manual_transition_job(api: Arc, job_id: Uuid) -> Resul return Ok(ManualTransitionJobRecoveryOutcome::Skipped); } + let recovery_unknown_snapshot = ManualTransitionQueueSnapshot::default(); + if record.mark_unknown_if_worker_results_lost(recovery_unknown_snapshot) + || record.mark_unknown_if_recovery_would_skip_pending_page(recovery_unknown_snapshot) + { + return match save_manual_transition_job_record_if_current(api.clone(), &record, &etag).await { + Ok(()) => { + release_manual_transition_recovery_admission(api, &record).await; + Ok(ManualTransitionJobRecoveryOutcome::Unknown) + } + Err(Error::PreconditionFailed) => Ok(ManualTransitionJobRecoveryOutcome::Skipped), + Err(err) => Err(err), + }; + } + if record.cancel_requested { - let mut report = record.report.clone(); - report.cancelled = true; - record.complete(report, manual_transition_queue_snapshot()); + record.cancel_after_recovery(queue_snapshot); return match save_manual_transition_job_record_if_current(api.clone(), &record, &etag).await { Ok(()) => { release_manual_transition_recovery_admission(api, &record).await; @@ -1954,7 +1975,7 @@ async fn recover_manual_transition_job(api: Arc, job_id: Uuid) -> Resul }; } - record.claim_recovery_lease(manual_transition_recovery_owner_id(), manual_transition_queue_snapshot()); + record.claim_recovery_lease(manual_transition_recovery_owner_id(), queue_snapshot); let recovery_lease_id = record.lease_id; match save_manual_transition_job_record_if_current(api.clone(), &record, &etag).await { Ok(()) => {} @@ -4544,10 +4565,11 @@ mod tests { use super::{ DATE_EXPIRY_EXISTING_OBJECTS_GRACE_SECS, DEFAULT_TRANSITION_QUEUE_CAPACITY, DEFAULT_TRANSITION_WORKERS_ABSOLUTE_MAX, DEFAULT_TRANSITION_WORKERS_CAP, EVENT_LIFECYCLE_EXPIRED_DETECTED, EVENT_LIFECYCLE_NOT_ENQUEUED, ExpiryState, - FreeVersionTask, ManualTransitionQueueSnapshot, ManualTransitionRunOptions, ManualTransitionRunReport, - StaleMultipartUploadCandidate, TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL, TIER_FREE_VERSION_RECOVERY_MAX_IDLE_INTERVAL, - TRANSITION_COMPLETE, TierFreeVersionRecoverySchedule, TransitionEnqueueOutcome, TransitionState, TransitionedObject, - VersionReplicationScan, cleanup_empty_multipart_sha_dirs_on_local_disks, cleanup_stale_multipart_uploads_once_at, + FreeVersionTask, ManualTransitionJobRecoveryOutcome, ManualTransitionQueueSnapshot, ManualTransitionRunOptions, + ManualTransitionRunReport, StaleMultipartUploadCandidate, TIER_FREE_VERSION_RECOVERY_BASE_INTERVAL, + TIER_FREE_VERSION_RECOVERY_MAX_IDLE_INTERVAL, TRANSITION_COMPLETE, 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_for_existing_objects_scoped, enqueue_transition_with_lifecycle, enqueue_transition_with_lifecycle_report, eval_action_from_lifecycle, jitter_tier_free_version_recovery_delay, lifecycle_action_blocked_by_replication, @@ -4555,13 +4577,13 @@ mod tests { lifecycle_rule_has_date_expiration, lifecycle_version_purge_state_from_completed_targets, manual_transition_duration_elapsed, manual_transition_has_more_after_limit, manual_transition_version_marker, mark_delete_opts_skip_decommissioned_on_remote_success, merge_stale_multipart_candidate, - persist_manual_transition_page_checkpoint, recover_manual_transition_jobs, recover_manual_transition_jobs_once, - replication_state_for_delete, resolve_tier_free_version_recovery_enabled, resolve_transition_queue_capacity, - resolve_transition_queue_send_timeout, resolve_transition_worker_count, resolve_transition_workers_absolute_max, - run_tier_free_version_recovery_loop, select_restore_s3_location, set_lifecycle_observability_observer, - set_recovered_free_version_enqueue_observer, should_defer_date_expiry_for_recent_config_update, - should_reuse_lifecycle_delete_replication_state, transitioned_cleanup_tuple, transitioned_object_delete_opts, - wait_for_tier_free_version_recovery, + persist_manual_transition_page_checkpoint, recover_manual_transition_job, recover_manual_transition_jobs, + recover_manual_transition_jobs_once, replication_state_for_delete, resolve_tier_free_version_recovery_enabled, + resolve_transition_queue_capacity, resolve_transition_queue_send_timeout, resolve_transition_worker_count, + resolve_transition_workers_absolute_max, run_tier_free_version_recovery_loop, select_restore_s3_location, + set_lifecycle_observability_observer, set_recovered_free_version_enqueue_observer, + should_defer_date_expiry_for_recent_config_update, should_reuse_lifecycle_delete_replication_state, + transitioned_cleanup_tuple, transitioned_object_delete_opts, wait_for_tier_free_version_recovery, }; #[cfg(feature = "test-util")] use super::{delete_free_version_remote_object_then, encode_dir_object, get_transitioned_object_reader_with_tier_manager}; @@ -4569,11 +4591,14 @@ mod tests { use crate::bucket::lifecycle::bucket_lifecycle_ops::{ decode_manual_transition_continuation_token, encode_manual_transition_continuation_token, }; + use crate::bucket::lifecycle::config_boundary; use crate::bucket::lifecycle::manual_transition_job::{ ManualTransitionJobRecord, ManualTransitionJobState, ManualTransitionScopeAdmission, ManualTransitionScopeAdmissionClaim, - claim_manual_transition_scope_admission, legacy_manual_transition_scope_key, load_manual_transition_job_record, - load_manual_transition_scope_admission, renew_manual_transition_job_lease, request_manual_transition_job_cancel, - save_manual_transition_job_record, save_manual_transition_scope_admission_if_absent, + claim_manual_transition_scope_admission, delete_manual_transition_scope_admission_if_current, + legacy_manual_transition_scope_key, load_manual_transition_job_record, load_manual_transition_scope_admission, + load_manual_transition_scope_admission_with_etag, manual_transition_scope_record_object_name, + renew_manual_transition_job_lease, request_manual_transition_job_cancel, save_manual_transition_job_record, + save_manual_transition_scope_admission_if_absent, save_manual_transition_scope_admission_if_current, }; use crate::bucket::lifecycle::replication_sink::{ ReplicateDecision, ReplicateTargetDecision, ReplicationStatusType, VersionPurgeStatusType, @@ -7607,6 +7632,203 @@ mod tests { ); } + #[tokio::test] + #[serial] + async fn manual_transition_recovery_marks_unknown_when_cursor_would_skip_pending_work() { + let (_paths, ecstore) = setup_test_env().await; + let job_id = Uuid::new_v4(); + let continuation_token = + encode_manual_transition_continuation_token(Some("logs/page-end".to_string()), Some("null".to_string())) + .expect("resume token should encode"); + let options = ManualTransitionRunOptions { + prefix: "logs/".to_string(), + continuation_token: Some(continuation_token.clone()), + ..Default::default() + }; + let mut record = ManualTransitionJobRecord::new(job_id, "manual-recovery-pending-page-bucket", &options, "old-owner"); + record.report.continuation_token = Some(continuation_token); + record.report.enqueued = 2; + record.report.transition_completed = 1; + record.lease_expires_at_unix_nanos = 0; + save_manual_transition_job_record(ecstore.clone(), &record) + .await + .expect("expired job record should save"); + save_manual_transition_scope_admission_if_absent(ecstore.clone(), &ManualTransitionScopeAdmission::from_job(&record)) + .await + .expect("expired scope admission should save"); + + let outcome = recover_manual_transition_job(ecstore.clone(), job_id, ManualTransitionQueueSnapshot::default()) + .await + .expect("manual transition recovery should process pending cursor jobs"); + + assert_eq!(outcome, ManualTransitionJobRecoveryOutcome::Unknown); + let recovered = load_manual_transition_job_record(ecstore.clone(), job_id) + .await + .expect("unknown job should load"); + assert_eq!(recovered.state, ManualTransitionJobState::Unknown); + assert_eq!(recovered.report.enqueued, 2); + assert_eq!(recovered.report.transition_completed, 1); + assert!( + recovered + .error + .as_deref() + .is_some_and(|error| error.contains("page/task journal")) + ); + assert!( + matches!( + load_manual_transition_scope_admission(ecstore, &record.scope_key).await, + Err(Error::ConfigNotFound) + ), + "unknown recovery must release the scope admission" + ); + } + + #[tokio::test] + #[serial] + async fn manual_transition_recovery_marks_unknown_for_cursor_pending_work_when_queue_is_busy() { + let (_paths, ecstore) = setup_test_env().await; + let job_id = Uuid::new_v4(); + let continuation_token = + encode_manual_transition_continuation_token(Some("logs/page-end".to_string()), Some("null".to_string())) + .expect("resume token should encode"); + let options = ManualTransitionRunOptions { + prefix: "logs/".to_string(), + continuation_token: Some(continuation_token.clone()), + ..Default::default() + }; + let mut record = ManualTransitionJobRecord::new(job_id, "manual-recovery-busy-queue-bucket", &options, "old-owner"); + record.report.continuation_token = Some(continuation_token); + record.report.enqueued = 2; + record.report.transition_completed = 1; + record.lease_expires_at_unix_nanos = 0; + save_manual_transition_job_record(ecstore.clone(), &record) + .await + .expect("expired job record should save"); + save_manual_transition_scope_admission_if_absent(ecstore.clone(), &ManualTransitionScopeAdmission::from_job(&record)) + .await + .expect("expired scope admission should save"); + + let outcome = recover_manual_transition_job( + ecstore.clone(), + job_id, + ManualTransitionQueueSnapshot { + queued: 1, + ..Default::default() + }, + ) + .await + .expect("busy queue recovery should fail closed for pending cursor jobs"); + + assert_eq!(outcome, ManualTransitionJobRecoveryOutcome::Unknown); + let loaded = load_manual_transition_job_record(ecstore.clone(), job_id) + .await + .expect("unknown job should remain loadable"); + assert_eq!(loaded.state, ManualTransitionJobState::Unknown); + assert!( + matches!( + load_manual_transition_scope_admission(ecstore, &record.scope_key).await, + Err(Error::ConfigNotFound) + ), + "unknown recovery must release the scope admission" + ); + } + + #[tokio::test] + #[serial] + async fn manual_transition_recovery_marks_unknown_before_cancel_for_cursor_pending_work() { + let (_paths, ecstore) = setup_test_env().await; + let job_id = Uuid::new_v4(); + let continuation_token = + encode_manual_transition_continuation_token(Some("logs/page-end".to_string()), Some("null".to_string())) + .expect("resume token should encode"); + let options = ManualTransitionRunOptions { + prefix: "logs/".to_string(), + continuation_token: Some(continuation_token.clone()), + ..Default::default() + }; + let mut record = ManualTransitionJobRecord::new(job_id, "manual-recovery-cancel-pending-bucket", &options, "old-owner"); + record.report.continuation_token = Some(continuation_token); + record.report.enqueued = 2; + record.report.transition_completed = 1; + record.lease_expires_at_unix_nanos = 0; + record.mark_cancel_requested(); + save_manual_transition_job_record(ecstore.clone(), &record) + .await + .expect("expired cancelled job record should save"); + save_manual_transition_scope_admission_if_absent(ecstore.clone(), &ManualTransitionScopeAdmission::from_job(&record)) + .await + .expect("expired cancelled scope admission should save"); + + let outcome = recover_manual_transition_job(ecstore.clone(), job_id, ManualTransitionQueueSnapshot::default()) + .await + .expect("manual transition recovery should fail closed before cancelled pending cursor jobs"); + + assert_eq!(outcome, ManualTransitionJobRecoveryOutcome::Unknown); + let recovered = load_manual_transition_job_record(ecstore.clone(), job_id) + .await + .expect("unknown cancelled job should load"); + assert_eq!(recovered.state, ManualTransitionJobState::Unknown); + assert!(recovered.cancel_requested); + assert!( + recovered + .error + .as_deref() + .is_some_and(|error| error.contains("page/task journal")) + ); + } + + #[tokio::test] + #[serial] + async fn manual_transition_recovery_marks_unknown_when_completed_scan_lost_worker_results() { + let (_paths, ecstore) = setup_test_env().await; + let job_id = Uuid::new_v4(); + let options = ManualTransitionRunOptions { + prefix: "logs/".to_string(), + ..Default::default() + }; + let mut record = + ManualTransitionJobRecord::new(job_id, "manual-recovery-lost-worker-result-bucket", &options, "old-owner"); + record.complete( + ManualTransitionRunReport { + bucket: record.bucket.clone(), + prefix: record.prefix.clone(), + enqueued: 2, + ..Default::default() + }, + ManualTransitionQueueSnapshot::default(), + ); + record.lease_expires_at_unix_nanos = 0; + save_manual_transition_job_record(ecstore.clone(), &record) + .await + .expect("expired job record should save"); + save_manual_transition_scope_admission_if_absent(ecstore.clone(), &ManualTransitionScopeAdmission::from_job(&record)) + .await + .expect("expired scope admission should save"); + + let outcome = recover_manual_transition_job(ecstore.clone(), job_id, ManualTransitionQueueSnapshot::default()) + .await + .expect("manual transition recovery should process completed scans with lost worker results"); + + assert_eq!(outcome, ManualTransitionJobRecoveryOutcome::Unknown); + let recovered = load_manual_transition_job_record(ecstore.clone(), job_id) + .await + .expect("unknown job should load"); + assert_eq!(recovered.state, ManualTransitionJobState::Unknown); + assert!( + recovered + .error + .as_deref() + .is_some_and(|error| error.contains("worker result")) + ); + assert!( + matches!( + load_manual_transition_scope_admission(ecstore, &record.scope_key).await, + Err(Error::ConfigNotFound) + ), + "unknown recovery must release the scope admission" + ); + } + #[tokio::test] #[serial] async fn manual_transition_recovery_drains_multiple_record_pages() { @@ -7658,11 +7880,18 @@ mod tests { async fn manual_transition_recovery_completes_expired_cancelled_record() { let (_paths, ecstore) = setup_test_env().await; let job_id = Uuid::new_v4(); + let continuation_token = + encode_manual_transition_continuation_token(Some("logs/page-end".to_string()), Some("null".to_string())) + .expect("resume token should encode"); let options = ManualTransitionRunOptions { prefix: "logs/".to_string(), + continuation_token: Some(continuation_token.clone()), ..Default::default() }; let mut record = ManualTransitionJobRecord::new(job_id, "manual-recovery-cancel-bucket", &options, "old-owner"); + record.report.continuation_token = Some(continuation_token); + record.report.enqueued = 2; + record.report.transition_completed = 2; record.lease_expires_at_unix_nanos = 0; record.mark_cancel_requested(); save_manual_transition_job_record(ecstore.clone(), &record) @@ -7677,6 +7906,7 @@ mod tests { .expect("manual transition recovery should process cancelled jobs"); assert_eq!(stats.cancelled, 1); + assert_eq!(stats.unknown, 0); assert_eq!(stats.resumed, 0); assert_eq!(stats.failed, 0); let recovered = load_manual_transition_job_record(ecstore.clone(), job_id) @@ -7858,6 +8088,50 @@ mod tests { ); } + #[tokio::test] + #[serial] + async fn manual_transition_scope_release_preserves_replaced_admission() { + let (_paths, ecstore) = setup_test_env().await; + let options = ManualTransitionRunOptions { + prefix: "logs/".to_string(), + ..Default::default() + }; + let first = ManualTransitionJobRecord::new(Uuid::new_v4(), "manual-release-race-bucket", &options, "first-owner"); + save_manual_transition_scope_admission_if_absent(ecstore.clone(), &ManualTransitionScopeAdmission::from_job(&first)) + .await + .expect("first scope admission should save"); + let (_loaded, etag) = load_manual_transition_scope_admission_with_etag(ecstore.clone(), &first.scope_key) + .await + .expect("first scope admission should load with an ETag"); + + let second = ManualTransitionJobRecord::new(Uuid::new_v4(), "manual-release-race-bucket", &options, "second-owner"); + save_manual_transition_scope_admission_if_current( + ecstore.clone(), + &ManualTransitionScopeAdmission::from_job(&second), + &etag, + ) + .await + .expect("second scope admission should replace first admission"); + + let object = manual_transition_scope_record_object_name(&first.scope_key).expect("scope admission path should encode"); + let stale_delete = config_boundary::delete_config_if_match(ecstore.clone(), &object, &etag) + .await + .expect_err("stale ETag must not delete a replaced scope admission"); + assert_eq!(stale_delete, Error::PreconditionFailed); + + let released = + delete_manual_transition_scope_admission_if_current(ecstore.clone(), &first.scope_key, first.job_id, first.lease_id) + .await + .expect("stale release should not fail"); + + assert!(!released); + let current = load_manual_transition_scope_admission(ecstore, &first.scope_key) + .await + .expect("replaced scope admission should remain"); + assert_eq!(current.job_id, second.job_id); + assert_eq!(current.lease_id, second.lease_id); + } + #[tokio::test] async fn existing_object_lifecycle_allows_expired_marker_after_replication_completed() { let lc = expired_delete_marker_lifecycle(); diff --git a/crates/ecstore/src/bucket/lifecycle/config_boundary.rs b/crates/ecstore/src/bucket/lifecycle/config_boundary.rs index 9f4deab7d..9dbfe7fd4 100644 --- a/crates/ecstore/src/bucket/lifecycle/config_boundary.rs +++ b/crates/ecstore/src/bucket/lifecycle/config_boundary.rs @@ -18,10 +18,11 @@ use http::HeaderMap; use rustfs_filemeta::FileInfo; use crate::config::com; +use crate::disk::RUSTFS_META_BUCKET; use crate::error::{Error, Result}; use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader}; use crate::storage_api_contracts::{ - object::{DeletedObject, ObjectIO, ObjectOperations, ObjectToDelete}, + object::{DeletedObject, HTTPPreconditions, ObjectIO, ObjectOperations, ObjectToDelete}, range::HTTPRangeSpec, }; @@ -98,3 +99,39 @@ where { com::delete_config(api, file).await } + +pub(crate) async fn delete_config_if_match(api: Arc, file: &str, etag: &str) -> Result<()> +where + S: ObjectOperations< + Error = Error, + ObjectInfo = ObjectInfo, + ObjectOptions = ObjectOptions, + FileInfo = FileInfo, + ObjectToDelete = ObjectToDelete, + DeletedObject = DeletedObject, + >, +{ + match api + .delete_object( + RUSTFS_META_BUCKET, + file, + ObjectOptions { + http_preconditions: Some(HTTPPreconditions { + if_match: Some(etag.to_string()), + ..Default::default() + }), + ..Default::default() + }, + ) + .await + { + Ok(_) => Ok(()), + Err(err) => { + if err == Error::FileNotFound || matches!(err, Error::ObjectNotFound(_, _)) { + Err(Error::ConfigNotFound) + } else { + Err(err) + } + } + } +} diff --git a/crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs b/crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs index d803feb1f..2e5487087 100644 --- a/crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs +++ b/crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs @@ -141,6 +141,17 @@ impl ManualTransitionJobRecord { self.mark_updated_terminal(); } + pub fn cancel_after_recovery(&mut self, queue_snapshot: ManualTransitionQueueSnapshot) { + if self.state == ManualTransitionJobState::Running && self.cancel_requested { + self.scan_completed = true; + self.report.cancelled = true; + self.queue_snapshot = queue_snapshot; + self.error = None; + self.state = ManualTransitionJobState::Cancelled; + self.mark_updated_terminal(); + } + } + pub fn record_worker_result(&mut self, result: ManualTransitionWorkerResult, queue_snapshot: ManualTransitionQueueSnapshot) { if self.is_terminal() { return; @@ -213,6 +224,23 @@ impl ManualTransitionJobRecord { true } + pub fn mark_unknown_if_recovery_would_skip_pending_page(&mut self, queue_snapshot: ManualTransitionQueueSnapshot) -> bool { + if self.state != ManualTransitionJobState::Running + || self.report.continuation_token.is_none() + || !self.report.worker_transition_pending() + || queue_snapshot.queued > 0 + || queue_snapshot.active > 0 + { + return false; + } + self.queue_snapshot = queue_snapshot; + self.state = ManualTransitionJobState::Unknown; + self.error = + Some("manual transition page/task journal is missing for pending work before the durable cursor".to_string()); + self.mark_updated_terminal(); + true + } + pub fn update_running_progress(&mut self, report: ManualTransitionRunReport, queue_snapshot: ManualTransitionQueueSnapshot) { if self.state == ManualTransitionJobState::Running { self.report.merge_scan_report_preserving_worker(&report); @@ -838,15 +866,16 @@ pub async fn delete_manual_transition_scope_admission_if_current( job_id: Uuid, lease_id: Uuid, ) -> EcstoreResult { - match load_manual_transition_scope_admission(api.clone(), scope_key).await { - Ok(admission) if admission.job_id == job_id && admission.lease_id == lease_id => {} + let etag = match load_manual_transition_scope_admission_with_etag(api.clone(), scope_key).await { + Ok((admission, etag)) if admission.job_id == job_id && admission.lease_id == lease_id => etag, Ok(_) => return Ok(false), Err(Error::ConfigNotFound) => return Ok(true), Err(err) => return Err(err), - } + }; let object = manual_transition_scope_record_object_name(scope_key).map_err(manual_transition_job_store_error)?; - match config_boundary::delete_config(api, &object).await { + match config_boundary::delete_config_if_match(api, &object, &etag).await { Ok(()) | Err(Error::ConfigNotFound) => Ok(true), + Err(Error::PreconditionFailed) => Ok(false), Err(err) => Err(err), } } @@ -961,6 +990,22 @@ mod tests { assert_eq!(record.report.tier_failure, 1); } + #[test] + fn manual_transition_job_cancel_recovery_finishes_after_worker_drain() { + let options = ManualTransitionRunOptions::default(); + let mut record = ManualTransitionJobRecord::new(Uuid::new_v4(), "bucket", &options, TEST_OWNER); + record.report.enqueued = 2; + record.report.transition_completed = 2; + record.mark_cancel_requested(); + + record.cancel_after_recovery(ManualTransitionQueueSnapshot::default()); + + assert_eq!(record.state, ManualTransitionJobState::Cancelled); + assert!(record.report.cancelled); + assert_eq!(record.report.enqueued, 2); + assert_eq!(record.report.transition_completed, 2); + } + #[test] fn manual_transition_job_record_decode_normalizes_legacy_cancelled_report() { let options = ManualTransitionRunOptions::default(); @@ -1031,6 +1076,30 @@ mod tests { assert!(record.completed_at_unix_nanos.is_some()); } + #[test] + fn manual_transition_job_marks_unknown_when_recovery_would_skip_pending_page() { + let options = ManualTransitionRunOptions::default(); + let mut record = ManualTransitionJobRecord::new(Uuid::new_v4(), "bucket", &options, TEST_OWNER); + record.report.enqueued = 2; + record.report.transition_completed = 1; + + assert!(!record.mark_unknown_if_recovery_would_skip_pending_page(ManualTransitionQueueSnapshot { + queued: 1, + ..Default::default() + })); + + record.report.continuation_token = Some("opaque".to_string()); + assert!(!record.mark_unknown_if_recovery_would_skip_pending_page(ManualTransitionQueueSnapshot { + active: 1, + ..Default::default() + })); + + assert!(record.mark_unknown_if_recovery_would_skip_pending_page(ManualTransitionQueueSnapshot::default())); + assert_eq!(record.state, ManualTransitionJobState::Unknown); + assert!(record.error.as_deref().is_some_and(|err| err.contains("page/task journal"))); + assert!(record.completed_at_unix_nanos.is_some()); + } + #[test] fn manual_transition_job_record_builds_resume_options() { let options = ManualTransitionRunOptions { diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 3b0fc82c3..3b15a94a1 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -3070,6 +3070,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { } if version_found { + opts.precondition_check(&goi)?; check_object_lock_delete(bucket, object, &goi, &opts).await?; } diff --git a/crates/ecstore/src/store/rebalance/support.rs b/crates/ecstore/src/store/rebalance/support.rs index 9c653e5db..6035e58d8 100644 --- a/crates/ecstore/src/store/rebalance/support.rs +++ b/crates/ecstore/src/store/rebalance/support.rs @@ -72,7 +72,13 @@ pub(super) fn resolve_rebalance_delete_from_all_pools_result( bucket: &str, object: &str, ) -> Result { - result.map_err(|err| Error::other(format!("failed to delete rebalance source object {bucket}/{object}: {err}"))) + result.map_err(|err| { + if err == Error::PreconditionFailed { + err + } else { + Error::other(format!("failed to delete rebalance source object {bucket}/{object}: {err}")) + } + }) } fn is_ignorable_rebalance_delete_error(err: &Error) -> bool { @@ -80,7 +86,11 @@ fn is_ignorable_rebalance_delete_error(err: &Error) -> bool { } fn rebalance_delete_pool_error(pool_idx: usize, bucket: &str, object: &str, err: Error) -> Error { - Error::other(format!("pool {pool_idx} delete failed for {bucket}/{object}: {err}")) + if err == Error::PreconditionFailed { + err + } else { + Error::other(format!("pool {pool_idx} delete failed for {bucket}/{object}: {err}")) + } } pub(super) fn resolve_rebalance_delete_from_all_pools_results( @@ -168,3 +178,31 @@ pub(super) fn resolve_latest_object_info_candidates( Err(pool_lookup_not_found_error(bucket, object, opts)) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn rebalance_delete_result_preserves_precondition_failed() { + let err = resolve_rebalance_delete_from_all_pools_result(Err(Error::PreconditionFailed), "bucket", "object") + .expect_err("precondition failure should remain structured"); + + assert_eq!(err, Error::PreconditionFailed); + } + + #[test] + fn rebalance_delete_pool_result_preserves_precondition_failed() { + let err = resolve_rebalance_delete_from_all_pools_results( + vec![RebalanceDeletePoolResult { + pool_idx: 0, + result: Err(Error::PreconditionFailed), + }], + "bucket", + "object", + ) + .expect_err("precondition failure should remain structured"); + + assert_eq!(err, Error::PreconditionFailed); + } +}