fix(tier): add operator transition reconciliation

crates/ecstore/src/api/mod.rs

@@ -67,6 +67,14 @@ pub mod bucket {
             };
         }
 
+        pub mod transition_transaction {
+            pub use crate::bucket::lifecycle::transition_transaction::{
+                TransitionOperatorDeleteResult, TransitionOperatorError, TransitionOperatorProbe, TransitionOperatorStatus,
+                delete_transition_candidate_for_operator, finalize_missing_transition_transaction_for_operator,
+                inspect_transition_transaction_for_operator,
+            };
+        }
+
         pub mod evaluator {
             pub use crate::bucket::lifecycle::evaluator::Evaluator;
         }

crates/ecstore/src/bucket/lifecycle/transition_transaction.rs

@@ -23,6 +23,7 @@ use uuid::Uuid;
 use crate::bucket::lifecycle::config_boundary;
 use crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE;
 use crate::bucket::lifecycle::tier_sweeper::{
+    delete_confirmed_transition_candidate_exact_with_lease_idempotent,
     delete_confirmed_transition_candidate_exact_with_manager_and_identity,
     delete_object_from_remote_tier_idempotent_with_manager_and_identity,
 };
@@ -616,6 +617,183 @@ pub enum TransitionTransactionRecoveryOutcome {
     Retained,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum TransitionOperatorProbe {
+    Missing,
+    UnversionedPresent,
+    VersionedPresent(String),
+    Ambiguous,
+    Unsupported,
+}
+
+impl From<TransitionCandidateProbe> for TransitionOperatorProbe {
+    fn from(value: TransitionCandidateProbe) -> Self {
+        match value {
+            TransitionCandidateProbe::Missing => Self::Missing,
+            TransitionCandidateProbe::UnversionedPresent => Self::UnversionedPresent,
+            TransitionCandidateProbe::VersionedPresent(version_id) => Self::VersionedPresent(version_id),
+            TransitionCandidateProbe::Ambiguous => Self::Ambiguous,
+            TransitionCandidateProbe::Unsupported => Self::Unsupported,
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
+pub struct TransitionOperatorStatus {
+    pub transaction_id: Uuid,
+    pub state: TransitionTransactionState,
+    pub tier_name: String,
+    pub remote_object: String,
+    pub not_after_unix_nanos: i64,
+    pub probe: TransitionOperatorProbe,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
+pub struct TransitionOperatorDeleteResult {
+    pub status: TransitionOperatorStatus,
+    pub journal_observed_after_delete: bool,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum TransitionOperatorError {
+    #[error("transition transaction was not found")]
+    NotFound,
+    #[error("transition transaction is still inside its active ownership window")]
+    NotExpired,
+    #[error("transition transaction state is not eligible for operator reconciliation: {0:?}")]
+    InvalidState(TransitionTransactionState),
+    #[error("an exact non-empty remote version is required")]
+    RemoteVersionRequired,
+    #[error("remote candidate is not proven missing: {0:?}")]
+    CandidateNotMissing(TransitionOperatorProbe),
+    #[error("transition transaction store failed: {0}")]
+    Store(#[source] Error),
+    #[error("remote tier reconciliation failed: {0}")]
+    Remote(#[source] std::io::Error),
+}
+
+type TransitionOperatorResult<T> = std::result::Result<T, TransitionOperatorError>;
+
+fn validate_operator_reconcile_transaction(
+    transaction: &TransitionTransaction,
+    now_unix_nanos: i128,
+) -> TransitionOperatorResult<()> {
+    transaction
+        .validate()
+        .map_err(|err| TransitionOperatorError::Store(Error::other(err)))?;
+    if transaction.state != TransitionTransactionState::UploadOutcomeUnknown {
+        return Err(TransitionOperatorError::InvalidState(transaction.state));
+    }
+    if now_unix_nanos < i128::from(transaction.not_after_unix_nanos) {
+        return Err(TransitionOperatorError::NotExpired);
+    }
+    Ok(())
+}
+
+async fn load_operator_reconcile_transaction(
+    api: Arc<ECStore>,
+    transaction_id: Uuid,
+) -> TransitionOperatorResult<TransitionTransaction> {
+    match load_transition_transaction_record(api, transaction_id).await {
+        Ok(transaction) => Ok(transaction),
+        Err(Error::ConfigNotFound) => Err(TransitionOperatorError::NotFound),
+        Err(err) => Err(TransitionOperatorError::Store(err)),
+    }
+}
+
+async fn operator_probe_transition_candidate(
+    api: Arc<ECStore>,
+    transaction: &TransitionTransaction,
+) -> TransitionOperatorResult<TransitionOperatorProbe> {
+    let lease = TierConfigMgr::acquire_operation_lease_for_backend_identity(
+        &api.tier_config_mgr(),
+        &transaction.tier_name,
+        transaction.backend_fingerprint,
+    )
+    .await
+    .map_err(|err| TransitionOperatorError::Remote(std::io::Error::other(err)))?;
+    lease
+        .probe_transition_candidate_for(&transaction.remote_object, transaction.transaction_id)
+        .await
+        .map(TransitionOperatorProbe::from)
+        .map_err(TransitionOperatorError::Remote)
+}
+
+pub async fn inspect_transition_transaction_for_operator(
+    api: Arc<ECStore>,
+    transaction_id: Uuid,
+) -> TransitionOperatorResult<TransitionOperatorStatus> {
+    let transaction = load_operator_reconcile_transaction(api.clone(), transaction_id).await?;
+    validate_operator_reconcile_transaction(&transaction, time::OffsetDateTime::now_utc().unix_timestamp_nanos())?;
+    let probe = operator_probe_transition_candidate(api, &transaction).await?;
+    Ok(TransitionOperatorStatus {
+        transaction_id,
+        state: transaction.state,
+        tier_name: transaction.tier_name,
+        remote_object: transaction.remote_object,
+        not_after_unix_nanos: transaction.not_after_unix_nanos,
+        probe,
+    })
+}
+
+pub async fn delete_transition_candidate_for_operator(
+    api: Arc<ECStore>,
+    transaction_id: Uuid,
+    remote_version_id: &str,
+) -> TransitionOperatorResult<TransitionOperatorDeleteResult> {
+    if remote_version_id.is_empty() {
+        return Err(TransitionOperatorError::RemoteVersionRequired);
+    }
+    let transaction = load_operator_reconcile_transaction(api.clone(), transaction_id).await?;
+    validate_operator_reconcile_transaction(&transaction, time::OffsetDateTime::now_utc().unix_timestamp_nanos())?;
+    let lease = TierConfigMgr::acquire_operation_lease_for_backend_identity(
+        &api.tier_config_mgr(),
+        &transaction.tier_name,
+        transaction.backend_fingerprint,
+    )
+    .await
+    .map_err(|err| TransitionOperatorError::Remote(std::io::Error::other(err)))?;
+    lease
+        .validate_remote_version_id(remote_version_id)
+        .map_err(TransitionOperatorError::Remote)?;
+    delete_confirmed_transition_candidate_exact_with_lease_idempotent(&transaction.remote_object, remote_version_id, &lease)
+        .await
+        .map_err(TransitionOperatorError::Remote)?;
+    let probe = operator_probe_transition_candidate(api.clone(), &transaction).await?;
+    let journal_observed_after_delete = match load_transition_transaction_record(api, transaction_id).await {
+        Ok(_) => true,
+        Err(Error::ConfigNotFound) => false,
+        Err(err) => return Err(TransitionOperatorError::Store(err)),
+    };
+    Ok(TransitionOperatorDeleteResult {
+        status: TransitionOperatorStatus {
+            transaction_id,
+            state: transaction.state,
+            tier_name: transaction.tier_name,
+            remote_object: transaction.remote_object,
+            not_after_unix_nanos: transaction.not_after_unix_nanos,
+            probe,
+        },
+        journal_observed_after_delete,
+    })
+}
+
+pub async fn finalize_missing_transition_transaction_for_operator(
+    api: Arc<ECStore>,
+    transaction_id: Uuid,
+) -> TransitionOperatorResult<()> {
+    let transaction = load_operator_reconcile_transaction(api.clone(), transaction_id).await?;
+    validate_operator_reconcile_transaction(&transaction, time::OffsetDateTime::now_utc().unix_timestamp_nanos())?;
+    let probe = operator_probe_transition_candidate(api.clone(), &transaction).await?;
+    if probe != TransitionOperatorProbe::Missing {
+        return Err(TransitionOperatorError::CandidateNotMissing(probe));
+    }
+    delete_transition_transaction_record(api, transaction_id)
+        .await
+        .map_err(TransitionOperatorError::Store)
+}
+
 pub(crate) fn decode_transition_transaction_record(object: &str, data: &[u8]) -> Result<TransitionTransaction> {
     let transaction_id = transition_transaction_id_from_record_object_name(object)?;
     TransitionTransaction::decode(transaction_id, data)
@@ -1097,6 +1275,27 @@ mod tests {
             .expect("upload state change should succeed")
     }
 
+    #[test]
+    fn operator_reconcile_requires_expired_unknown_upload_outcome() {
+        let mut transaction = new_transaction();
+        let active_deadline = transaction.not_after_unix_nanos;
+
+        assert!(matches!(
+            validate_operator_reconcile_transaction(&transaction, i128::from(active_deadline) + 1),
+            Err(TransitionOperatorError::InvalidState(TransitionTransactionState::UploadStarted))
+        ));
+
+        transaction
+            .advance(transaction.fence(), TransitionTransactionState::UploadOutcomeUnknown, None)
+            .expect("unknown upload outcome should be recorded");
+        assert!(matches!(
+            validate_operator_reconcile_transaction(&transaction, i128::from(active_deadline) - 1),
+            Err(TransitionOperatorError::NotExpired)
+        ));
+        validate_operator_reconcile_transaction(&transaction, i128::from(active_deadline))
+            .expect("expired unknown upload outcome should be eligible");
+    }
+
     fn cleanup_proof(transaction: &TransitionTransaction, decision: TransitionCleanupDecision) -> TransitionCleanupProof {
         TransitionCleanupProof {
             transaction_id: transaction.transaction_id,

crates/ecstore/src/store/init.rs

@@ -562,10 +562,12 @@ mod tests {
             },
             tier_sweeper::Jentry,
             transition_transaction::{
-                TRANSITION_TRANSACTION_RECORD_PREFIX, TransitionCleanupDecision, TransitionCleanupProof, TransitionRemoteVersion,
-                TransitionSourceIdentity, TransitionSourceVersionMode, TransitionTransaction, TransitionTransactionInit,
-                TransitionTransactionState, load_transition_transaction_record, recover_transition_transaction_records,
-                save_transition_transaction_record,
+                TRANSITION_TRANSACTION_RECORD_PREFIX, TransitionCleanupDecision, TransitionCleanupProof, TransitionOperatorError,
+                TransitionOperatorProbe, TransitionRemoteVersion, TransitionSourceIdentity, TransitionSourceVersionMode,
+                TransitionTransaction, TransitionTransactionInit, TransitionTransactionState,
+                delete_transition_candidate_for_operator, finalize_missing_transition_transaction_for_operator,
+                inspect_transition_transaction_for_operator, load_transition_transaction_record,
+                recover_transition_transaction_records, save_transition_transaction_record,
             },
         },
         client::transition_api::ReaderImpl,
@@ -575,6 +577,7 @@ mod tests {
         services::tier::{
             test_util::{MockWarmBackend, MockWarmOp, TransitionCleanupStoreBarrier, register_mock_tier},
             tier::{TIER_CONFIG_FILE, TierConfigMgr},
+            tier_config::{TierConfig, TierType, TierWasabi},
             tier_mutation_intent::{
                 TIER_MUTATION_INTENT_RECORD_PREFIX, TierMutationIntent, TierMutationIntentKind, TierMutationIntentState,
                 TierMutationIntentTarget, advance_tier_mutation_intent_record_idempotent, delete_tier_mutation_intent_record,
@@ -1163,6 +1166,37 @@ mod tests {
             .len()
     }
 
+    #[cfg(feature = "test-util")]
+    async fn register_operator_reconcile_test_tier(
+        handle: &Arc<tokio::sync::RwLock<TierConfigMgr>>,
+        tier_name: &str,
+    ) -> MockWarmBackend {
+        let backend = MockWarmBackend::new();
+        let mut manager = handle.write().await;
+        manager.tiers.insert(
+            tier_name.to_string(),
+            TierConfig {
+                version: "v1".to_string(),
+                tier_type: TierType::Wasabi,
+                name: tier_name.to_string(),
+                wasabi: Some(TierWasabi {
+                    name: tier_name.to_string(),
+                    endpoint: "https://s3.wasabisys.com".to_string(),
+                    access_key: "test-access-key".to_string(),
+                    secret_key: "test-secret-key".to_string(),
+                    bucket: "mock-tier".to_string(),
+                    prefix: format!("mock/{}/", uuid::Uuid::new_v4()),
+                    region: "us-east-1".to_string(),
+                }),
+                ..Default::default()
+            },
+        );
+        manager
+            .install_test_driver(tier_name, Box::new(backend.clone()))
+            .expect("mock fallback tier driver should install");
+        backend
+    }
+
     #[cfg(feature = "test-util")]
     async fn wait_for_tier_delete_journal_recovery(
         store: Arc<crate::store::ECStore>,
@@ -2809,6 +2843,154 @@ mod tests {
         }
     }
 
+    #[cfg(feature = "test-util")]
+    #[tokio::test]
+    #[serial_test::serial(storage_class_env)]
+    async fn operator_reconcile_deletes_exact_candidate_before_finalizing_record() {
+        let temp_dir = tempfile::tempdir().expect("create temp store dir");
+        let (ctx, store, _shutdown) =
+            without_storage_class_env(build_isolated_test_store(temp_dir.path(), "transition-operator-reconcile", &[4])).await;
+        crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+
+        let tier_name = "TXOPERATOR";
+        let backend = register_operator_reconcile_test_tier(&ctx.tier_config_mgr(), tier_name).await;
+        let backend_identity = TierConfigMgr::acquire_operation_lease(&ctx.tier_config_mgr(), tier_name)
+            .await
+            .expect("tier lease should resolve")
+            .backend_identity();
+        let mut transaction = TransitionTransaction::new(TransitionTransactionInit {
+            deployment_id: ctx.deployment_id().expect("test store should initialize deployment id"),
+            transaction_id: uuid::Uuid::new_v4(),
+            owner_epoch: uuid::Uuid::new_v4(),
+            write_id: uuid::Uuid::new_v4(),
+            source: TransitionSourceIdentity {
+                bucket: "source-bucket".to_string(),
+                object: "source-object".to_string(),
+                version_id: None,
+                data_dir: uuid::Uuid::new_v4(),
+                mod_time_unix_nanos: 1_770_000_000_000_000_000,
+                size: 42,
+                etag: "source-etag".to_string(),
+                version_mode: TransitionSourceVersionMode::Unversioned,
+            },
+            tier_name: tier_name.to_string(),
+            backend_fingerprint: backend_identity,
+            not_after_unix_nanos: 1,
+        })
+        .expect("transaction should build");
+        transaction
+            .advance(transaction.fence(), TransitionTransactionState::UploadOutcomeUnknown, None)
+            .expect("transaction should enter unknown upload outcome state");
+
+        let remote_version = uuid::Uuid::new_v4().to_string();
+        backend.set_put_remote_version(Some(remote_version.clone())).await;
+        let candidate = bytes::Bytes::from_static(b"operator-confirmed transition candidate");
+        backend
+            .put(
+                &transaction.remote_object,
+                ReaderImpl::Body(candidate.clone()),
+                i64::try_from(candidate.len()).expect("test candidate length should fit i64"),
+            )
+            .await
+            .expect("mock backend should accept candidate");
+        save_transition_transaction_record(store.clone(), &transaction)
+            .await
+            .expect("transaction record should persist");
+
+        let status = inspect_transition_transaction_for_operator(store.clone(), transaction.transaction_id)
+            .await
+            .expect("operator inspection should probe the candidate");
+        assert_eq!(status.probe, TransitionOperatorProbe::VersionedPresent(remote_version.clone()));
+
+        let wrong_version = uuid::Uuid::new_v4().to_string();
+        let result = delete_transition_candidate_for_operator(store.clone(), transaction.transaction_id, &wrong_version)
+            .await
+            .expect("an exact delete of an absent version should be idempotent");
+        assert_eq!(
+            result.status.probe,
+            TransitionOperatorProbe::VersionedPresent(remote_version.clone()),
+            "an incorrect exact version must not delete the provider-confirmed candidate"
+        );
+        assert!(result.journal_observed_after_delete);
+        load_transition_transaction_record(store.clone(), transaction.transaction_id)
+            .await
+            .expect("an incorrect exact version must retain the transaction journal");
+
+        let result = delete_transition_candidate_for_operator(store.clone(), transaction.transaction_id, &remote_version)
+            .await
+            .expect("operator-confirmed exact candidate should be deleted");
+        assert_eq!(result.status.probe, TransitionOperatorProbe::Missing);
+        assert!(result.journal_observed_after_delete);
+        assert_eq!(backend.exact_remove_count(), 2);
+        assert_eq!(backend.remove_versions().await, vec![(transaction.remote_object.clone(), remote_version)]);
+        load_transition_transaction_record(store.clone(), transaction.transaction_id)
+            .await
+            .expect("candidate deletion must retain the transaction journal");
+
+        finalize_missing_transition_transaction_for_operator(store.clone(), transaction.transaction_id)
+            .await
+            .expect("a separately confirmed missing candidate should permit finalization");
+        assert!(matches!(
+            load_transition_transaction_record(store, transaction.transaction_id).await,
+            Err(Error::ConfigNotFound)
+        ));
+    }
+
+    #[cfg(feature = "test-util")]
+    #[tokio::test]
+    #[serial_test::serial(storage_class_env)]
+    async fn operator_finalize_retains_record_without_missing_proof() {
+        let temp_dir = tempfile::tempdir().expect("create temp store dir");
+        let (ctx, store, _shutdown) =
+            without_storage_class_env(build_isolated_test_store(temp_dir.path(), "transition-operator-fail-closed", &[4])).await;
+        crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+
+        let tier_name = "TXOPERATORFAIL";
+        let backend = register_operator_reconcile_test_tier(&ctx.tier_config_mgr(), tier_name).await;
+        let backend_identity = TierConfigMgr::acquire_operation_lease(&ctx.tier_config_mgr(), tier_name)
+            .await
+            .expect("tier lease should resolve")
+            .backend_identity();
+        let mut transaction = TransitionTransaction::new(TransitionTransactionInit {
+            deployment_id: ctx.deployment_id().expect("test store should initialize deployment id"),
+            transaction_id: uuid::Uuid::new_v4(),
+            owner_epoch: uuid::Uuid::new_v4(),
+            write_id: uuid::Uuid::new_v4(),
+            source: TransitionSourceIdentity {
+                bucket: "source-bucket".to_string(),
+                object: "source-object".to_string(),
+                version_id: None,
+                data_dir: uuid::Uuid::new_v4(),
+                mod_time_unix_nanos: 1_770_000_000_000_000_000,
+                size: 42,
+                etag: "source-etag".to_string(),
+                version_mode: TransitionSourceVersionMode::Unversioned,
+            },
+            tier_name: tier_name.to_string(),
+            backend_fingerprint: backend_identity,
+            not_after_unix_nanos: 1,
+        })
+        .expect("transaction should build");
+        transaction
+            .advance(transaction.fence(), TransitionTransactionState::UploadOutcomeUnknown, None)
+            .expect("transaction should enter unknown upload outcome state");
+        save_transition_transaction_record(store.clone(), &transaction)
+            .await
+            .expect("transaction record should persist");
+        backend
+            .set_transition_candidate_probe_override(Some(TransitionCandidateProbe::Unsupported))
+            .await;
+
+        assert!(matches!(
+            finalize_missing_transition_transaction_for_operator(store.clone(), transaction.transaction_id).await,
+            Err(TransitionOperatorError::CandidateNotMissing(TransitionOperatorProbe::Unsupported))
+        ));
+        load_transition_transaction_record(store, transaction.transaction_id)
+            .await
+            .expect("an unsupported probe must retain the transaction journal");
+        assert_eq!(backend.remove_count().await, 0);
+    }
+
     #[cfg(feature = "test-util")]
     #[tokio::test]
     #[serial_test::serial(storage_class_env)]