From 921cc7ad94bdac8b3a9464065e01072011c221c0 Mon Sep 17 00:00:00 2001 From: houseme Date: Tue, 8 Sep 2026 22:53:21 +0800 Subject: [PATCH] fix(scanner): confirm recovery intent accept readback (#7516) Co-authored-by: zhi22915 --- crates/scanner/src/scanner/cycle_state.rs | 99 +++++++++++++++- .../src/scanner/tests/recovery_control.rs | 107 +++++++++++++++++- 2 files changed, 204 insertions(+), 2 deletions(-) diff --git a/crates/scanner/src/scanner/cycle_state.rs b/crates/scanner/src/scanner/cycle_state.rs index f20374ae0..5f6ea4437 100644 --- a/crates/scanner/src/scanner/cycle_state.rs +++ b/crates/scanner/src/scanner/cycle_state.rs @@ -128,6 +128,79 @@ pub(super) mod cleanup_io_fault { } } +#[cfg(test)] +pub(super) mod recovery_intent_accept_fault { + use super::*; + + enum Fault { + Corrupt, + Running, + } + + static NEXT_ACCEPT_READBACK_FAULT: StdMutex> = StdMutex::new(None); + + pub(in crate::scanner) struct Guard; + + impl Drop for Guard { + fn drop(&mut self) { + *NEXT_ACCEPT_READBACK_FAULT + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) = None; + } + } + + pub(in crate::scanner) fn corrupt_next_accept_readback() -> Guard { + install(Fault::Corrupt) + } + + pub(in crate::scanner) fn advance_next_accept_readback_to_running() -> Guard { + install(Fault::Running) + } + + fn install(fault: Fault) -> Guard { + let mut slot = NEXT_ACCEPT_READBACK_FAULT + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + assert!(slot.is_none(), "only one recovery intent accept readback fault may be installed"); + *slot = Some(fault); + Guard + } + + pub(super) async fn maybe_apply(storeapi: Arc, path: &str) -> Result<(), ScannerError> + where + S: ScannerObjectIO, + { + let Some(fault) = NEXT_ACCEPT_READBACK_FAULT + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .take() + else { + return Ok(()); + }; + match fault { + Fault::Corrupt => save_config(storeapi, path, b"{corrupt".to_vec()).await.map_err(|err| { + ScannerError::Other(format!("failed to inject scanner recovery intent accept readback fault: {err}")) + }), + Fault::Running => { + let mut record = read_recovery_intent_record(storeapi.clone(), path).await?.ok_or_else(|| { + ScannerError::Other("scanner recovery intent disappeared before fault injection".to_string()) + })?; + record.state = SCANNER_RECOVERY_INTENT_STATE_RUNNING.to_string(); + save_config( + storeapi, + path, + serde_json::to_vec(&record) + .map_err(|err| ScannerError::Other(format!("failed to encode scanner recovery intent fault: {err}")))?, + ) + .await + .map_err(|err| { + ScannerError::Other(format!("failed to inject scanner recovery intent accept readback fault: {err}")) + }) + } + } + } +} + #[derive(Clone, Debug, Default, Serialize)] pub struct ScannerCycleRecoveryStatus { /// The immutable primary object whose revision is being guarded. @@ -564,6 +637,17 @@ fn compare_recovery_intent( } } +fn confirm_recovery_intent_acceptance( + expected: ScannerRecoveryIntentRecord, + persisted: ScannerRecoveryIntentRecord, +) -> ScannerRecoveryIntentAcceptResult { + if persisted == expected { + ScannerRecoveryIntentAcceptResult::Accepted { record: persisted } + } else { + compare_recovery_intent(&expected, persisted) + } +} + async fn read_recovery_intent_record( storeapi: Arc, path: &str, @@ -757,7 +841,20 @@ pub async fn accept_scanner_usage_recovery_intent( .map_err(|err| ScannerError::Other(format!("failed to encode scanner recovery intent: {err}")))?; match save_config_with_preconditions(storeapi.clone(), &path, encoded, DataUsageCacheRevision::Missing.preconditions()).await { - Ok(_) => Ok(ScannerRecoveryIntentAcceptResult::Accepted { record: candidate }), + Ok(_) => { + #[cfg(test)] + recovery_intent_accept_fault::maybe_apply(storeapi.clone(), &path).await?; + let persisted = match read_recovery_intent_record(storeapi.clone(), &path).await { + Ok(Some(record)) => record, + Ok(None) => { + return Err(ScannerError::Other( + "scanner recovery intent disappeared before acceptance confirmation".to_string(), + )); + } + Err(error) => return Err(error), + }; + Ok(confirm_recovery_intent_acceptance(candidate, persisted)) + } Err(EcstoreError::PreconditionFailed) => { let existing = read_recovery_intent_record(storeapi, &path).await?; let Some(existing) = existing else { diff --git a/crates/scanner/src/scanner/tests/recovery_control.rs b/crates/scanner/src/scanner/tests/recovery_control.rs index 9191c238c..76ed49f4e 100644 --- a/crates/scanner/src/scanner/tests/recovery_control.rs +++ b/crates/scanner/src/scanner/tests/recovery_control.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use super::super::cycle_state::cleanup_io_fault; +use super::super::cycle_state::{cleanup_io_fault, recovery_intent_accept_fault}; use super::*; use crate::storage_api::owner::{EcstoreRebalStatus, EcstoreRebalanceInfo, EcstoreRebalanceMeta, EcstoreRebalanceStats}; @@ -256,6 +256,100 @@ async fn scanner_recovery_intent_accept_is_durable_and_idempotent() { assert_eq!(replay, ScannerRecoveryIntentAcceptResult::Replayed { record }); } +#[tokio::test] +#[serial] +async fn scanner_recovery_intent_accept_requires_confirmed_readback() { + let (_dir, store) = setup_scanner_cycle_store().await; + let _fault = recovery_intent_accept_fault::corrupt_next_accept_readback(); + let error = + accept_scanner_usage_recovery_intent(store.clone(), recovery_intent_request("intent-key-0001-readback", "operator-a")) + .await + .expect_err("accept must fail when the just-written intent cannot be confirmed"); + assert!(error.to_string().contains("scanner recovery intent is invalid"), "{error}"); + + let restarted = restart_scanner_cycle_store_from(&store).await; + let error = scanner_usage_recovery_intents_for_startup(&CancellationToken::new(), restarted) + .await + .expect_err("unconfirmed corrupt intent must remain a fail-closed startup error"); + assert!(error.to_string().contains("scanner recovery intent is invalid"), "{error}"); +} + +#[tokio::test] +#[serial] +async fn scanner_recovery_intent_accept_replays_if_execution_advances_before_readback() { + let (_dir, store) = setup_scanner_cycle_store().await; + let request = recovery_intent_request("intent-key-0001-running", "operator-a"); + let _fault = recovery_intent_accept_fault::advance_next_accept_readback_to_running(); + let replay = accept_scanner_usage_recovery_intent(store.clone(), request.clone()) + .await + .expect("same request advanced by execution remains idempotent"); + let record = match replay { + ScannerRecoveryIntentAcceptResult::Replayed { record } => record, + other => panic!("advanced same-request record must replay instead of conflict: {other:?}"), + }; + assert_eq!(record.state, "running"); + + let retry = accept_scanner_usage_recovery_intent(store, request) + .await + .expect("lost response retry observes the running durable record"); + assert_eq!(retry, ScannerRecoveryIntentAcceptResult::Replayed { record }); +} + +#[tokio::test] +#[serial] +async fn concurrent_scanner_recovery_intent_acceptance_uses_one_durable_record() { + let (_dir, store) = setup_scanner_cycle_store().await; + let request = recovery_intent_request("intent-key-0001-concurrent", "operator-a"); + let mut tasks = Vec::new(); + for _ in 0..12 { + let store = store.clone(); + let request = request.clone(); + tasks.push(tokio::spawn(async move { + accept_scanner_usage_recovery_intent(store, request) + .await + .expect("concurrent same-key accept should converge") + })); + } + + let mut accepted = 0usize; + let mut replayed = 0usize; + let mut records = Vec::new(); + for task in tasks { + match task.await.expect("accept task should not panic") { + ScannerRecoveryIntentAcceptResult::Accepted { record } => { + accepted += 1; + records.push(record); + } + ScannerRecoveryIntentAcceptResult::Replayed { record } => { + replayed += 1; + records.push(record); + } + other => panic!("same-key accepts must not conflict: {other:?}"), + } + } + assert_eq!(accepted, 1, "exactly one request may win the missing-record CAS"); + assert_eq!(replayed, 11, "all other same-key requests must replay the durable winner"); + assert!( + records.windows(2).all(|pair| pair[0] == pair[1]), + "all accepts must return the same durable identity" + ); + + let restarted = restart_scanner_cycle_store_from(&store).await; + let replayable = scanner_usage_recovery_intents_for_startup(&CancellationToken::new(), restarted.clone()) + .await + .expect("startup should rediscover the single indexed intent"); + assert_eq!(replayable, vec![records[0].intent_id.clone()]); + let replay = accept_scanner_usage_recovery_intent(restarted, request) + .await + .expect("lost response after restart should replay the same record"); + assert_eq!( + replay, + ScannerRecoveryIntentAcceptResult::Replayed { + record: records[0].clone() + } + ); +} + #[tokio::test] #[serial] async fn scanner_recovery_intent_executor_persists_completed_progress() { @@ -547,6 +641,17 @@ async fn scanner_recovery_intent_query_rejects_corrupt_or_unknown_records() { .await .expect_err("corrupt intent must not decode as absent"); assert!(error.to_string().contains("scanner recovery intent is invalid")); + + let mut future = serde_json::to_value(&record).expect("record value"); + future["future_writer_capability"] = serde_json::json!("durable-accept-v2"); + save_config(store.clone(), &path, serde_json::to_vec(&future).expect("future record should encode")) + .await + .expect("future durable record"); + let error = get_scanner_usage_recovery_intent(store.clone(), &record.intent_id) + .await + .expect_err("future writer payload must not decode as a known terminal state"); + assert!(error.to_string().contains("scanner recovery intent is invalid")); + let unknown = get_scanner_usage_recovery_intent(store, &scanner_recovery_actor_sha256("missing")) .await .expect("missing intent should read as absent");