diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 43c467b14..2fe32b834 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -221,10 +221,13 @@ jobs: github-token: ${{ secrets.GITHUB_TOKEN }} cache-save-if: ${{ github.ref == 'refs/heads/main' }} - # The #4877 restore self-deadlock is fixed in this PR, which re-enabled - # test_multipart_restore_preserves_parts_and_etag. The remaining exclusions - # each hit a DIFFERENT, independent issue (all tracked under - # rustfs/backlog#1148; they keep #[ignore] with a backlog reference): + # restore_object_usecase_reports_ongoing_conflict_and_completion was + # re-enabled by backlog#1304 (restore accepts serialize on a short CAS + # guard; the copy-back no longer holds the #4877 whole-copy-back lock, + # so the mid-restore ongoing read and fast 409 rejection it asserts are + # the implemented contract). The remaining exclusions each hit a + # DIFFERENT, independent issue (all tracked under rustfs/backlog#1148; + # they keep #[ignore] with a backlog reference): # - test_noncurrent_{expiry,transition}_still_works_after_immediate_compensation_transition: # noncurrent transition/expiry after an immediate compensation transition. # - test_transition_and_restore_flows: transition metadata is missing on @@ -235,15 +238,11 @@ jobs: # delete path reads that flag, so cleanup deletes the whole object # instead of only the local restored copy (ObjectNotFound afterwards). # The expire_restored delete semantics are unimplemented. - # - restore_object_usecase_reports_ongoing_conflict_and_completion: asserts - # a concurrent get_object_info observes ongoing-request=true mid-restore, - # which #4877's read-vs-restore serialization rules out (see backlog#1148 - # ilm-8 criterion 1 - an API-semantics decision, not a bug). - name: Run ignored ILM integration tests serially run: | cargo nextest run -j1 --run-ignored ignored-only \ -p rustfs-scanner -p rustfs \ - -E '(binary(lifecycle_integration_test) or (package(rustfs) and test(lifecycle_transition_api_test))) and not (test(test_transition_and_restore_flows) or test(test_noncurrent_expiry_still_works_after_immediate_compensation_transition) or test(test_noncurrent_transition_still_works_after_immediate_compensation_transition) or test(restore_object_usecase_reports_ongoing_conflict_and_completion) or test(test_restore_chain_local_read_expiry_keeps_remote_and_allows_re_restore))' + -E '(binary(lifecycle_integration_test) or (package(rustfs) and test(lifecycle_transition_api_test))) and not (test(test_transition_and_restore_flows) or test(test_noncurrent_expiry_still_works_after_immediate_compensation_transition) or test(test_noncurrent_transition_still_works_after_immediate_compensation_transition) or test(test_restore_chain_local_read_expiry_keeps_remote_and_allows_re_restore))' test-and-lint-rio-v2: name: Test and Lint (rio-v2) diff --git a/crates/ecstore/src/store/object.rs b/crates/ecstore/src/store/object.rs index 8231b1a6f..d9bd0beed 100644 --- a/crates/ecstore/src/store/object.rs +++ b/crates/ecstore/src/store/object.rs @@ -103,6 +103,21 @@ impl ObjectLockDiagGuard { } } +/// Opaque write-lock guard for the RestoreObject accept path; see +/// [`ECStore::acquire_restore_accept_guard`]. Deliberately not a general lock +/// API — it only exists so the accept path's restore-status compare-and-set +/// can span the ecstore/API layer boundary. +pub struct RestoreAcceptGuard(ObjectLockDiagGuard); + +impl RestoreAcceptGuard { + /// True when the underlying namespace lock was lost (e.g. heartbeat + /// refresh lost quorum). Callers must check this before committing the + /// restore-status write the guard exists to serialize. + pub fn is_lock_lost(&self) -> bool { + self.0.guard.is_lock_lost() + } +} + impl Drop for ObjectLockDiagGuard { fn drop(&mut self) { if !self.enabled || self.guard.is_released() { @@ -439,6 +454,23 @@ impl ECStore { Ok(Some(guard)) } + /// Serializes the RestoreObject accept path — the read of the current + /// `x-amz-restore` status, the ongoing/already-restored decision, and the + /// metadata write that flips `ongoing-request="true"` — against concurrent + /// accepts of the same object, making that read-check-write an atomic + /// compare-and-set (backlog#1304). While the guard is held the caller must + /// pass `no_lock: true` on its reads/writes of this object, check + /// [`RestoreAcceptGuard::is_lock_lost`] before the status write, and drop + /// the guard before starting the tier copy-back so concurrent + /// HEAD/`get_object_info` stay non-blocking during the restore. + pub async fn acquire_restore_accept_guard(&self, bucket: &str, object: &str) -> Result { + let object = encode_dir_object(object); + let guard = self + .acquire_object_write_lock("restore_object_accept", bucket, &object) + .await?; + Ok(RestoreAcceptGuard(guard)) + } + async fn acquire_delete_objects_write_locks( &self, bucket: &str, @@ -1308,19 +1340,25 @@ impl ECStore { opts: &ObjectOptions, ) -> Result<()> { let object = encode_dir_object(object); - let mut opts = opts.clone(); - let _object_lock_guard = self - .acquire_object_write_lock_if_needed("restore_transitioned_object", bucket, &object, &mut opts) - .await?; - + // Deliberately NOT holding the object write lock across the tier + // copy-back (backlog#1304): non-SELECT restore-vs-restore is + // serialized by the accept path's compare-and-set of the ongoing flag + // (see acquire_restore_accept_guard), and while ongoing-request="true" + // the restore header parses with no expiry, so DeleteRestoredAction + // cannot fire mid-copy-back. Torn-write protection against concurrent + // readers and writers stays with the inner put_object / + // complete_multipart_upload commit phases, which take this object's + // write lock themselves. A delete (user or lifecycle) landing between + // the tier read and that commit can still be overwritten by the + // commit — the same window MinIO accepts; a commit-time existence + // re-check is tracked separately. Holding the lock here instead + // (#4877) blocked HEAD/get_object_info for the whole copy-back and + // self-deadlocked on the inner commits. if self.single_pool() { - return self.pools[0] - .clone() - .restore_transitioned_object(bucket, &object, &opts) - .await; + return self.pools[0].clone().restore_transitioned_object(bucket, &object, opts).await; } - let opts = transition_restore_pool_opts(&opts); + let opts = transition_restore_pool_opts(opts); let (_, idx) = self .get_latest_accessible_object_info_with_idx(bucket, object.as_str(), &opts) .await?; @@ -2179,25 +2217,32 @@ mod tests { ); } + // NOTE: #4877's `restore_transitioned_object_waits_for_existing_reader` + // was removed with the whole-copy-back write lock it asserted + // (backlog#1304): restore entry no longer serializes on the object lock. + // The replacement semantics — non-blocking reads during the copy-back and + // fast rejection of a concurrent restore — are covered end-to-end by + // `restore_object_usecase_reports_ongoing_conflict_and_completion` + // (rustfs/src/app/lifecycle_transition_api_test.rs) and at the lock level + // by the accept-guard test below; restore-vs-reader data protection lives + // in the inner put_object/complete_multipart_upload commit locks. #[tokio::test] #[serial_test::serial] - async fn restore_transitioned_object_waits_for_existing_reader() { + async fn restore_accept_guard_serializes_concurrent_accepts() { + // backlog#1304: the accept guard is the compare-and-set boundary for + // the restore ongoing flag — a second accept of the same object must + // wait behind (here: time out on) the first. let store = Arc::new(new_read_lock_test_store().await); - let ns_lock = store - .handle_new_ns_lock("bucket", "object") + let _first = store + .acquire_restore_accept_guard("bucket", "object") .await - .expect("namespace lock should be created"); - let _reader = ns_lock - .get_read_lock(Duration::from_secs(1)) - .await - .expect("reader should acquire the object lock"); + .expect("first accept guard should be acquired"); let err = temp_env::async_with_vars([(rustfs_config::ENV_OBJECT_LOCK_ACQUIRE_TIMEOUT, Some("1"))], async { - store - .clone() - .handle_restore_transitioned_object("bucket", "object", &ObjectOptions::default()) - .await - .expect_err("restore must wait behind an existing reader") + match store.acquire_restore_accept_guard("bucket", "object").await { + Ok(_) => panic!("second accept guard must wait behind the first"), + Err(err) => err, + } }) .await; diff --git a/rustfs/src/app/lifecycle_transition_api_test.rs b/rustfs/src/app/lifecycle_transition_api_test.rs index f5d06a30a..9b8d284c3 100644 --- a/rustfs/src/app/lifecycle_transition_api_test.rs +++ b/rustfs/src/app/lifecycle_transition_api_test.rs @@ -1526,24 +1526,19 @@ async fn put_bucket_lifecycle_configuration_rejects_zero_day_expiration() { /// POST restore(days=1) is accepted and immediately flips the object to /// `x-amz-restore: ongoing-request="true"` (the mock tier's injected GET /// latency keeps the background copy-back in flight); a second POST during -/// that window is rejected with `ErrObjectRestoreAlreadyInProgress`; once the +/// that window is rejected with 409 `RestoreAlreadyInProgress`; once the /// copy-back completes the object reports `ongoing-request="false"` with a /// future expiry-date; and a full GET is then served from the local restored /// copy (the mock tier records no further `get` calls). /// -/// CURRENTLY EXCLUDED from the CI ILM Integration (serial) lane (see ci.yml): -/// the #4877 restore self-deadlock is now fixed, so restore completes, but this -/// test asserts a concurrent `get_object_info` observes `ongoing-request="true"` -/// mid-restore. #4877 serializes reads against the restore write lock, so the -/// concurrent read only returns once the copy-back has finished and already -/// cleared the ongoing flag — it reads `false`, failing the assertion. Whether -/// a concurrent restore should instead reject fast (`ErrObjectRestoreAlreadyInProgress`) -/// while keeping HEAD non-blocking (backlog#1148 ilm-8 criterion 1) is an -/// API-semantics decision, not a bug; revisit before re-enabling. The `test/` -/// prefix and the rest of the contract are correct. +/// Re-enabled in the serial lane by backlog#1304: the accept path now flips +/// the ongoing flag under a short compare-and-set guard and the copy-back +/// runs without the #4877 whole-copy-back lock, so the mid-restore +/// `ongoing-request="true"` read and the fast 409 rejection asserted here are +/// the implemented contract. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[serial] -#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-8); currently excluded there — asserts a concurrent ongoing-request=true read that #4877's read-vs-restore serialization rules out (API-semantics decision)"] +#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-8)"] async fn restore_object_usecase_reports_ongoing_conflict_and_completion() { let (_disk_paths, ecstore) = setup_test_env().await; let usecase = DefaultObjectUsecase::from_global(); @@ -1615,7 +1610,7 @@ async fn restore_object_usecase_reports_ongoing_conflict_and_completion() { .expect_err("a second restore while one is ongoing must be rejected"); assert_eq!( err.code(), - &s3s::S3ErrorCode::Custom("ErrObjectRestoreAlreadyInProgress".into()), + &s3s::S3ErrorCode::RestoreAlreadyInProgress, "unexpected rejection for a repeated restore: {err:?}" ); @@ -1660,6 +1655,112 @@ async fn restore_object_usecase_reports_ongoing_conflict_and_completion() { ); } +/// backlog#1304: the restore-accept compare-and-set itself, under real +/// concurrency. Two POST ?restore for the same transitioned object race each +/// other from separate tasks; the accept guard must let exactly one through — +/// the winner flips `ongoing-request="true"`, the loser must read that flag +/// and be rejected with 409 `RestoreAlreadyInProgress` — and the tier must +/// see exactly one copy-back GET, never two. Without the accept guard both +/// requests can observe ongoing=false and double-start the copy-back (the +/// #4859 race), which this test catches via the tier GET count. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[serial] +#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-8)"] +async fn restore_object_usecase_accepts_exactly_one_of_two_concurrent_restores() { + let (_disk_paths, ecstore) = setup_test_env().await; + + let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase(); + let backend = register_mock_tier(&tier_name).await; + + let bucket = format!("test-api-restore-cas-{}", &Uuid::new_v4().simple().to_string()[..8]); + // Must live under the `test/` prefix — the shared transition rule filters on it. + let object = "test/restore/cas-object.bin"; + let payload: Vec = (0..128 * 1024).map(|i| (i % 251) as u8).collect(); + + create_test_bucket(&ecstore, bucket.as_str()).await; + set_bucket_lifecycle_transition_with_tier(bucket.as_str(), &tier_name) + .await + .expect("Failed to set lifecycle configuration"); + let _ = upload_test_object(&ecstore, bucket.as_str(), object, &payload).await; + + lifecycle::bucket_lifecycle_ops::enqueue_transition_for_existing_objects(ecstore.clone(), bucket.as_str()) + .await + .expect("Failed to enqueue transitioned object"); + let _ = wait_for_transition(&ecstore, bucket.as_str(), object, TRANSITION_WAIT_TIMEOUT) + .await + .expect("object should transition before the concurrent restores run"); + + // Keep the winner's copy-back in flight while the loser's accept runs, so + // the loser cannot slip into the already-restored path after a completed + // copy-back. + backend.set_latency(Some(Duration::from_millis(1500))).await; + + let tier_gets_before_restore = backend.get_count().await; + + let restore_input = || { + RestoreObjectInput::builder() + .bucket(bucket.clone()) + .key(object.to_string()) + .restore_request(Some(RestoreRequest { + days: Some(1), + description: None, + glacier_job_parameters: None, + output_location: None, + select_parameters: None, + tier: None, + type_: None, + })) + .build() + .unwrap() + }; + let spawn_restore = |input: RestoreObjectInput| { + tokio::spawn(async move { + let usecase = DefaultObjectUsecase::from_global(); + Box::pin(usecase.execute_restore_object(build_request(input, Method::POST))).await + }) + }; + + let (first, second) = tokio::join!(spawn_restore(restore_input()), spawn_restore(restore_input())); + let results = [ + first.expect("first concurrent restore task panicked"), + second.expect("second concurrent restore task panicked"), + ]; + + let accepted = results.iter().filter(|result| result.is_ok()).count(); + assert_eq!(accepted, 1, "exactly one of two concurrent restores must be accepted, got {accepted}"); + let rejection = results + .iter() + .find_map(|result| result.as_ref().err()) + .expect("the losing concurrent restore must surface an error"); + assert_eq!( + rejection.code(), + &s3s::S3ErrorCode::RestoreAlreadyInProgress, + "the losing concurrent restore must be rejected as already in progress: {rejection:?}" + ); + + // Let the single accepted copy-back complete, then verify the tier saw + // exactly one restore read — a second GET means a double copy-back. + let mut completed = false; + for _ in 0..40 { + let info = ecstore + .get_object_info(bucket.as_str(), object, &ObjectOptions::default()) + .await + .expect("Failed to poll object info for restore completion"); + if !info.restore_ongoing && info.restore_expires.is_some() { + completed = true; + break; + } + tokio::time::sleep(Duration::from_millis(500)).await; + } + backend.clear_faults().await; + assert!(completed, "the accepted restore copy-back should complete within the poll window"); + assert_eq!( + backend.get_count().await - tier_gets_before_restore, + 1, + "two concurrent restore requests must trigger exactly one tier copy-back GET" + ); +} + /// rustfs/backlog#1320: a single PUT must compute the replication decision /// exactly once. Historically the PUT path computed `must_replicate_object` /// twice with the same inputs — once before commit to persist the pending diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index d1ffb505d..a783c68ce 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -6369,10 +6369,36 @@ impl DefaultObjectUsecase { }; let version_id_str = version_id.clone().unwrap_or_default(); - let opts = post_restore_opts(&version_id_str, &bucket, &object) + let mut opts = post_restore_opts(&version_id_str, &bucket, &object) .await .map_err(|_| S3Error::with_message(S3ErrorCode::Custom("ErrPostRestoreOpts".into()), "restore object failed."))?; + // SELECT-type restores skip both the ongoing check and the metadata + // write below, so the accept guard would protect nothing for them — + // they keep the plain (read-locked) accept path. + let is_select = rreq.type_.as_ref().is_some_and(|t| t.as_str() == "SELECT"); + + // Hold the restore-accept guard across the restore-status read, the + // ongoing/already-restored decision, and the metadata write below, so + // two concurrent (non-SELECT) POST ?restore cannot both observe + // ongoing=false and both start a copy-back (backlog#1304). Reads and + // writes inside this scope run with no_lock; the guard is dropped + // before the copy-back is spawned so it never blocks readers. + // Contention on the accept guard (e.g. a concurrent accept or an + // in-flight commit on the same object) is transient — answer 503 + // SlowDown so SDK clients back off and retry instead of treating it + // as a hard failure. + let accept_guard = if is_select { + None + } else { + let guard = store + .acquire_restore_accept_guard(&bucket, &object) + .await + .map_err(|_| S3Error::with_message(S3ErrorCode::SlowDown, "restore object failed."))?; + opts.no_lock = true; + Some(guard) + }; + let mut obj_info = store .get_object_info(&bucket, &object, &opts) .await @@ -6394,11 +6420,13 @@ impl DefaultObjectUsecase { )); } - // Check if restore is already in progress - if obj_info.restore_ongoing && (rreq.type_.as_ref().is_none_or(|t| t.as_str() != "SELECT")) { + // Check if restore is already in progress. AWS answers this with + // 409 RestoreAlreadyInProgress; a Custom code would serialize as a + // retryable 500 and make SDK clients retry the conflict (backlog#1304). + if obj_info.restore_ongoing && !is_select { return Err(S3Error::with_message( - S3ErrorCode::Custom("ErrObjectRestoreAlreadyInProgress".into()), - "restore object failed.", + S3ErrorCode::RestoreAlreadyInProgress, + "Object restore is already in progress.", )); } @@ -6417,7 +6445,7 @@ impl DefaultObjectUsecase { let event_object_info = obj_info.clone(); let obj_info_ = obj_info.clone(); - if rreq.type_.as_ref().is_none_or(|t| t.as_str() != "SELECT") { + if !is_select { obj_info.metadata_only = true; metadata.insert(AMZ_RESTORE_EXPIRY_DAYS.to_string(), rreq.days.unwrap_or(1).to_string()); let request_date = OffsetDateTime::now_utc().format(&Rfc3339).map_err(|e| { @@ -6445,6 +6473,14 @@ impl DefaultObjectUsecase { } obj_info.user_defined = Arc::new(metadata); + // Fence the compare-and-set write: if the accept guard was lost + // (lock-service degradation), another node may have concurrently + // accepted this restore — back off instead of committing a second + // ongoing flag and double-starting the copy-back. + if accept_guard.as_ref().is_some_and(|g| g.is_lock_lost()) { + return Err(S3Error::with_message(S3ErrorCode::SlowDown, "restore object failed.")); + } + store .clone() .copy_object( @@ -6455,11 +6491,14 @@ impl DefaultObjectUsecase { &mut obj_info, &ObjectOptions { version_id: obj_info_.version_id.map(|v| v.to_string()), + // Inside the accept-guard critical section (see above). + no_lock: true, ..Default::default() }, &ObjectOptions { version_id: obj_info_.version_id.map(|v| v.to_string()), mod_time: obj_info_.mod_time, + no_lock: true, ..Default::default() }, ) @@ -6482,6 +6521,10 @@ impl DefaultObjectUsecase { } } + // The accept decision is committed; release the object write lock so + // the background copy-back and concurrent reads are never blocked on it. + drop(accept_guard); + // Handle output location for SELECT requests if let Some(output_location) = &rreq.output_location && let Some(s3) = &output_location.s3 @@ -6493,12 +6536,17 @@ impl DefaultObjectUsecase { } } - // Spawn restoration task in the background + // Spawn restoration task in the background. Pin the copy-back to the + // version the accept resolved and flagged: with a versionless request + // on a versioned bucket, a PUT landing between the accept and the + // copy-back would otherwise re-resolve "latest" to the new version, + // fail (not transitioned), and strand the flagged version at + // ongoing=true forever (backlog#1304). let store_clone = store.clone(); let bucket_clone = bucket.clone(); let object_clone = object.clone(); let rreq_clone = rreq.clone(); - let version_id_clone = version_id.clone(); + let version_id_clone = obj_info_.version_id.map(|v| v.to_string()); spawn_traced(async move { let opts = ObjectOptions {