fix(ilm): add transition transaction foundation (#5065)

Co-authored-by: heihutu <heihutu@gmail.com>

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

@@ -28,3 +28,4 @@ pub mod tier_delete_journal;
 pub mod tier_free_version_recovery;
 pub mod tier_last_day_stats;
 pub mod tier_sweeper;
+pub mod transition_transaction;

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

@@ -0,0 +1,925 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use rustfs_utils::crypto::{hex_sha256, is_sha256_checksum};
+use serde::{Deserialize, Serialize};
+use uuid::Uuid;
+
+pub const TRANSITION_TRANSACTION_SCHEMA: &str = "rustfs-transition-transaction-v1";
+pub const TRANSITION_TRANSACTION_PREFIX: &str = "ilm/transition-transactions";
+pub const MAX_TRANSITION_TRANSACTION_SIZE: usize = 64 * 1024;
+
+pub type Result<T> = std::result::Result<T, TransitionTransactionError>;
+
+#[derive(Debug, thiserror::Error)]
+pub enum TransitionTransactionError {
+    #[error("transition transaction already exists")]
+    AlreadyExists,
+    #[error("transition transaction is not found")]
+    NotFound,
+    #[error("transition transaction is corrupt: {0}")]
+    Corrupt(&'static str),
+    #[error("transition transaction schema is unsupported: {0}")]
+    UnsupportedSchema(String),
+    #[error("transition transaction checksum mismatch")]
+    ChecksumMismatch,
+    #[error("transition transaction owner or revision fence is stale")]
+    Fenced,
+    #[error("invalid transition transaction state change from {from:?} to {to:?}")]
+    InvalidStateChange {
+        from: TransitionTransactionState,
+        to: TransitionTransactionState,
+    },
+    #[error("transition transaction json error: {0}")]
+    Json(#[from] serde_json::Error),
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum TransitionTransactionState {
+    UploadStarted,
+    UploadOutcomeUnknown,
+    AbortedNoRemote,
+    Uploaded,
+    LocalCommitStarted,
+    Committed,
+    CleanupPending,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum TransitionRemoteVersionKind {
+    Unknown,
+    Unversioned,
+    Versioned,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct TransitionRemoteVersion {
+    pub kind: TransitionRemoteVersionKind,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub version_id: Option<String>,
+}
+
+impl TransitionRemoteVersion {
+    pub fn known_from_put_response(version_id: impl Into<String>) -> Self {
+        let version_id = version_id.into();
+        if version_id.is_empty() || Uuid::parse_str(&version_id).is_ok_and(|parsed| parsed.is_nil()) {
+            Self::unversioned()
+        } else {
+            Self::versioned(version_id)
+        }
+    }
+
+    pub fn unknown() -> Self {
+        Self {
+            kind: TransitionRemoteVersionKind::Unknown,
+            version_id: None,
+        }
+    }
+
+    pub fn unversioned() -> Self {
+        Self {
+            kind: TransitionRemoteVersionKind::Unversioned,
+            version_id: None,
+        }
+    }
+
+    pub fn versioned(version_id: impl Into<String>) -> Self {
+        Self {
+            kind: TransitionRemoteVersionKind::Versioned,
+            version_id: Some(version_id.into()),
+        }
+    }
+
+    pub fn is_unknown(&self) -> bool {
+        self.kind == TransitionRemoteVersionKind::Unknown
+    }
+
+    pub fn tier_delete_version_id(&self) -> Option<&str> {
+        match self.kind {
+            TransitionRemoteVersionKind::Unknown | TransitionRemoteVersionKind::Unversioned => None,
+            TransitionRemoteVersionKind::Versioned => self.version_id.as_deref(),
+        }
+    }
+
+    fn validate(&self) -> Result<()> {
+        match (&self.kind, self.version_id.as_deref()) {
+            (TransitionRemoteVersionKind::Unknown | TransitionRemoteVersionKind::Unversioned, None) => Ok(()),
+            (TransitionRemoteVersionKind::Unknown | TransitionRemoteVersionKind::Unversioned, Some(_)) => Err(
+                TransitionTransactionError::Corrupt("remote version must be absent for non-versioned state"),
+            ),
+            (TransitionRemoteVersionKind::Versioned, Some(version_id)) => {
+                if version_id.is_empty() {
+                    return Err(TransitionTransactionError::Corrupt("versioned remote version is empty"));
+                }
+                if Uuid::parse_str(version_id).is_ok_and(|parsed| parsed.is_nil()) {
+                    return Err(TransitionTransactionError::Corrupt("versioned remote version is nil uuid"));
+                }
+                Ok(())
+            }
+            (TransitionRemoteVersionKind::Versioned, None) => {
+                Err(TransitionTransactionError::Corrupt("versioned remote version is missing"))
+            }
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct TransitionSourceIdentity {
+    pub bucket: String,
+    pub object: String,
+    pub version_id: Option<Uuid>,
+    pub data_dir: Uuid,
+    pub mod_time_unix_nanos: i64,
+    pub size: i64,
+    pub etag: String,
+    pub version_mode: TransitionSourceVersionMode,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum TransitionSourceVersionMode {
+    Unversioned,
+    Versioned,
+    VersionSuspended,
+}
+
+impl TransitionSourceIdentity {
+    pub fn validate(&self) -> Result<()> {
+        if self.bucket.is_empty() {
+            return Err(TransitionTransactionError::Corrupt("source bucket is empty"));
+        }
+        if self.object.is_empty() {
+            return Err(TransitionTransactionError::Corrupt("source object is empty"));
+        }
+        if self.data_dir.is_nil() {
+            return Err(TransitionTransactionError::Corrupt("source data_dir is nil"));
+        }
+        if self.version_id.is_some_and(|version_id| version_id.is_nil()) {
+            return Err(TransitionTransactionError::Corrupt("source version_id is nil"));
+        }
+        if self.version_mode == TransitionSourceVersionMode::Versioned && self.version_id.is_none() {
+            return Err(TransitionTransactionError::Corrupt("versioned source is missing version_id"));
+        }
+        if matches!(
+            self.version_mode,
+            TransitionSourceVersionMode::Unversioned | TransitionSourceVersionMode::VersionSuspended
+        ) && self.version_id.is_some()
+        {
+            return Err(TransitionTransactionError::Corrupt("non-versioned source mode must not carry version_id"));
+        }
+        if self.size < 0 {
+            return Err(TransitionTransactionError::Corrupt("source size is negative"));
+        }
+        if self.etag.is_empty() {
+            return Err(TransitionTransactionError::Corrupt("source etag is empty"));
+        }
+        Ok(())
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct TransitionTransactionFence {
+    pub revision: u64,
+    pub owner_epoch: Uuid,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct TransitionTransaction {
+    pub deployment_id: Uuid,
+    pub transaction_id: Uuid,
+    pub revision: u64,
+    pub owner_epoch: Uuid,
+    pub write_id: Uuid,
+    pub source: TransitionSourceIdentity,
+    pub tier_name: String,
+    pub backend_fingerprint: String,
+    pub remote_object: String,
+    pub remote_version: TransitionRemoteVersion,
+    pub state: TransitionTransactionState,
+    pub not_after_unix_nanos: i64,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct TransitionTransactionInit {
+    pub deployment_id: Uuid,
+    pub transaction_id: Uuid,
+    pub owner_epoch: Uuid,
+    pub write_id: Uuid,
+    pub source: TransitionSourceIdentity,
+    pub tier_name: String,
+    pub backend_fingerprint: String,
+    pub not_after_unix_nanos: i64,
+}
+
+impl TransitionTransaction {
+    pub fn new(init: TransitionTransactionInit) -> Result<Self> {
+        let remote_object =
+            canonical_transition_remote_object(init.deployment_id, &init.source.bucket, init.transaction_id, init.write_id)?;
+        let transaction = Self {
+            deployment_id: init.deployment_id,
+            transaction_id: init.transaction_id,
+            revision: 1,
+            owner_epoch: init.owner_epoch,
+            write_id: init.write_id,
+            source: init.source,
+            tier_name: init.tier_name,
+            backend_fingerprint: init.backend_fingerprint,
+            remote_object,
+            remote_version: TransitionRemoteVersion::unknown(),
+            state: TransitionTransactionState::UploadStarted,
+            not_after_unix_nanos: init.not_after_unix_nanos,
+        };
+        transaction.validate()?;
+        Ok(transaction)
+    }
+
+    pub fn fence(&self) -> TransitionTransactionFence {
+        TransitionTransactionFence {
+            revision: self.revision,
+            owner_epoch: self.owner_epoch,
+        }
+    }
+
+    pub fn validate(&self) -> Result<()> {
+        if self.deployment_id.is_nil() {
+            return Err(TransitionTransactionError::Corrupt("deployment_id is nil"));
+        }
+        if self.transaction_id.is_nil() {
+            return Err(TransitionTransactionError::Corrupt("transaction_id is nil"));
+        }
+        if self.revision == 0 {
+            return Err(TransitionTransactionError::Corrupt("revision is zero"));
+        }
+        if self.owner_epoch.is_nil() {
+            return Err(TransitionTransactionError::Corrupt("owner_epoch is nil"));
+        }
+        if self.write_id.is_nil() {
+            return Err(TransitionTransactionError::Corrupt("write_id is nil"));
+        }
+        self.source.validate()?;
+        if self.tier_name.is_empty() {
+            return Err(TransitionTransactionError::Corrupt("tier name is empty"));
+        }
+        if !is_sha256_checksum(&self.backend_fingerprint) {
+            return Err(TransitionTransactionError::Corrupt("backend fingerprint is not a sha256 checksum"));
+        }
+        if self.remote_object
+            != canonical_transition_remote_object(self.deployment_id, &self.source.bucket, self.transaction_id, self.write_id)?
+        {
+            return Err(TransitionTransactionError::Corrupt("remote object is not canonical"));
+        }
+        self.remote_version.validate()?;
+        if state_requires_known_remote_version(self.state) && self.remote_version.is_unknown() {
+            return Err(TransitionTransactionError::Corrupt("state requires known remote version"));
+        }
+        if matches!(
+            self.state,
+            TransitionTransactionState::UploadStarted
+                | TransitionTransactionState::UploadOutcomeUnknown
+                | TransitionTransactionState::AbortedNoRemote
+        ) && !self.remote_version.is_unknown()
+        {
+            return Err(TransitionTransactionError::Corrupt(
+                "pre-upload-completion state must not carry a known remote version",
+            ));
+        }
+        Ok(())
+    }
+
+    pub fn advance(
+        &mut self,
+        fence: TransitionTransactionFence,
+        next: TransitionTransactionState,
+        remote_version: Option<TransitionRemoteVersion>,
+    ) -> Result<TransitionTransactionFence> {
+        self.check_fence(fence)?;
+        if !state_change_allowed(self.state, next) {
+            return Err(TransitionTransactionError::InvalidStateChange {
+                from: self.state,
+                to: next,
+            });
+        }
+
+        match next {
+            TransitionTransactionState::Uploaded => {
+                let remote_version =
+                    remote_version.ok_or(TransitionTransactionError::Corrupt("uploaded state requires remote version"))?;
+                if remote_version.is_unknown() {
+                    return Err(TransitionTransactionError::Corrupt("uploaded state requires known remote version"));
+                }
+                self.remote_version = remote_version;
+            }
+            TransitionTransactionState::AbortedNoRemote => {
+                if remote_version.is_some() {
+                    return Err(TransitionTransactionError::Corrupt(
+                        "aborted no remote state must not carry remote version",
+                    ));
+                }
+                self.remote_version = TransitionRemoteVersion::unknown();
+            }
+            TransitionTransactionState::UploadOutcomeUnknown => {
+                if remote_version.is_some() {
+                    return Err(TransitionTransactionError::Corrupt(
+                        "unknown upload outcome state must not carry remote version",
+                    ));
+                }
+                self.remote_version = TransitionRemoteVersion::unknown();
+            }
+            TransitionTransactionState::LocalCommitStarted | TransitionTransactionState::Committed => {
+                if let Some(remote_version) = remote_version
+                    && remote_version != self.remote_version
+                {
+                    return Err(TransitionTransactionError::Corrupt("remote version changed after upload"));
+                }
+                if self.remote_version.is_unknown() {
+                    return Err(TransitionTransactionError::Corrupt("local commit requires known remote version"));
+                }
+            }
+            TransitionTransactionState::CleanupPending | TransitionTransactionState::UploadStarted => {
+                return Err(TransitionTransactionError::InvalidStateChange {
+                    from: self.state,
+                    to: next,
+                });
+            }
+        }
+
+        self.state = next;
+        self.bump_revision()?;
+        self.validate()?;
+        Ok(self.fence())
+    }
+
+    pub fn mark_cleanup_pending(
+        &mut self,
+        fence: TransitionTransactionFence,
+        proof: TransitionCleanupProof,
+    ) -> Result<TransitionTransactionFence> {
+        self.check_fence(fence)?;
+        self.validate_cleanup_proof(&proof)?;
+
+        match (&self.state, &proof.decision) {
+            (TransitionTransactionState::Uploaded, TransitionCleanupDecision::UploadAbortedBeforeLocalCommit)
+            | (
+                TransitionTransactionState::UploadOutcomeUnknown,
+                TransitionCleanupDecision::RemoteVersionRecoveredAfterCancellation,
+            ) => {}
+            (
+                TransitionTransactionState::LocalCommitStarted,
+                TransitionCleanupDecision::SourceReconciledUnchanged { observed_source },
+            ) if observed_source == &self.source => {}
+            _ => {
+                return Err(TransitionTransactionError::InvalidStateChange {
+                    from: self.state,
+                    to: TransitionTransactionState::CleanupPending,
+                });
+            }
+        }
+
+        self.state = TransitionTransactionState::CleanupPending;
+        self.remote_version = proof.remote_version;
+        self.bump_revision()?;
+        self.validate()?;
+        Ok(self.fence())
+    }
+
+    pub fn encode(&self) -> Result<Vec<u8>> {
+        self.validate()?;
+        let transaction_bytes = serde_json::to_vec(self)?;
+        let content_sha256 = hex_sha256(&transaction_bytes, ToOwned::to_owned);
+        let persisted = PersistedTransitionTransaction {
+            schema: TRANSITION_TRANSACTION_SCHEMA.to_string(),
+            content_sha256,
+            transaction: self.clone(),
+        };
+        let encoded = serde_json::to_vec(&persisted)?;
+        if encoded.len() > MAX_TRANSITION_TRANSACTION_SIZE {
+            return Err(TransitionTransactionError::Corrupt("encoded transaction exceeds maximum size"));
+        }
+        Ok(encoded)
+    }
+
+    pub fn decode(expected_transaction_id: Uuid, data: &[u8]) -> Result<Self> {
+        if data.len() > MAX_TRANSITION_TRANSACTION_SIZE {
+            return Err(TransitionTransactionError::Corrupt("encoded transaction exceeds maximum size"));
+        }
+        let persisted: PersistedTransitionTransaction = serde_json::from_slice(data)?;
+        if persisted.schema != TRANSITION_TRANSACTION_SCHEMA {
+            return Err(TransitionTransactionError::UnsupportedSchema(persisted.schema));
+        }
+        if !is_sha256_checksum(&persisted.content_sha256) {
+            return Err(TransitionTransactionError::Corrupt("content checksum is not a sha256 checksum"));
+        }
+        let transaction_bytes = serde_json::to_vec(&persisted.transaction)?;
+        let actual_checksum = hex_sha256(&transaction_bytes, ToOwned::to_owned);
+        if persisted.content_sha256 != actual_checksum {
+            return Err(TransitionTransactionError::ChecksumMismatch);
+        }
+        if persisted.transaction.transaction_id != expected_transaction_id {
+            return Err(TransitionTransactionError::Corrupt("transaction_id does not match record key"));
+        }
+        persisted.transaction.validate()?;
+        Ok(persisted.transaction)
+    }
+
+    fn check_fence(&self, fence: TransitionTransactionFence) -> Result<()> {
+        if self.revision != fence.revision || self.owner_epoch != fence.owner_epoch {
+            return Err(TransitionTransactionError::Fenced);
+        }
+        Ok(())
+    }
+
+    fn validate_cleanup_proof(&self, proof: &TransitionCleanupProof) -> Result<()> {
+        if proof.transaction_id != self.transaction_id
+            || proof.write_id != self.write_id
+            || proof.remote_object != self.remote_object
+            || proof.backend_fingerprint != self.backend_fingerprint
+        {
+            return Err(TransitionTransactionError::Fenced);
+        }
+        proof.remote_version.validate()?;
+        if proof.remote_version.is_unknown() {
+            return Err(TransitionTransactionError::Corrupt("cleanup proof requires known remote version"));
+        }
+        if state_requires_known_remote_version(self.state) && proof.remote_version != self.remote_version {
+            return Err(TransitionTransactionError::Fenced);
+        }
+        Ok(())
+    }
+
+    fn bump_revision(&mut self) -> Result<()> {
+        self.revision = self
+            .revision
+            .checked_add(1)
+            .ok_or(TransitionTransactionError::Corrupt("revision overflow"))?;
+        Ok(())
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct TransitionCleanupProof {
+    pub transaction_id: Uuid,
+    pub write_id: Uuid,
+    pub remote_object: String,
+    pub remote_version: TransitionRemoteVersion,
+    pub backend_fingerprint: String,
+    pub decision: TransitionCleanupDecision,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
+pub enum TransitionCleanupDecision {
+    UploadAbortedBeforeLocalCommit,
+    RemoteVersionRecoveredAfterCancellation,
+    SourceReconciledUnchanged { observed_source: TransitionSourceIdentity },
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct PersistedTransitionTransaction {
+    schema: String,
+    content_sha256: String,
+    transaction: TransitionTransaction,
+}
+
+pub fn canonical_transition_remote_object(
+    deployment_id: Uuid,
+    source_bucket: &str,
+    transaction_id: Uuid,
+    write_id: Uuid,
+) -> Result<String> {
+    if deployment_id.is_nil() {
+        return Err(TransitionTransactionError::Corrupt("deployment_id is nil"));
+    }
+    if source_bucket.is_empty() {
+        return Err(TransitionTransactionError::Corrupt("source bucket is empty"));
+    }
+    if transaction_id.is_nil() {
+        return Err(TransitionTransactionError::Corrupt("transaction_id is nil"));
+    }
+    if write_id.is_nil() {
+        return Err(TransitionTransactionError::Corrupt("write_id is nil"));
+    }
+
+    let bucket_scope = format!("{deployment_id}:{source_bucket}");
+    let bucket_hash = hex_sha256(bucket_scope.as_bytes(), ToOwned::to_owned);
+    let transaction_key = transaction_id.simple().to_string();
+    let write_key = write_id.simple().to_string();
+    Ok(format!(
+        "{}/{}/{}/{}/{}/{}",
+        TRANSITION_TRANSACTION_PREFIX,
+        &bucket_hash[..16],
+        &transaction_key[..2],
+        &transaction_key[2..4],
+        transaction_key,
+        write_key
+    ))
+}
+
+fn state_change_allowed(from: TransitionTransactionState, to: TransitionTransactionState) -> bool {
+    matches!(
+        (from, to),
+        (TransitionTransactionState::UploadStarted, TransitionTransactionState::Uploaded)
+            | (
+                TransitionTransactionState::UploadStarted,
+                TransitionTransactionState::UploadOutcomeUnknown
+            )
+            | (TransitionTransactionState::UploadStarted, TransitionTransactionState::AbortedNoRemote)
+            | (TransitionTransactionState::UploadOutcomeUnknown, TransitionTransactionState::Uploaded)
+            | (TransitionTransactionState::Uploaded, TransitionTransactionState::LocalCommitStarted)
+            | (TransitionTransactionState::LocalCommitStarted, TransitionTransactionState::Committed)
+    )
+}
+
+fn state_requires_known_remote_version(state: TransitionTransactionState) -> bool {
+    matches!(
+        state,
+        TransitionTransactionState::Uploaded
+            | TransitionTransactionState::LocalCommitStarted
+            | TransitionTransactionState::Committed
+            | TransitionTransactionState::CleanupPending
+    )
+}
+
+#[cfg(test)]
+mod tests {
+    use std::collections::HashMap;
+
+    use super::*;
+
+    const BACKEND_FINGERPRINT: &str = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef";
+
+    #[derive(Default)]
+    struct MemoryTransactionStore {
+        records: HashMap<Uuid, Vec<u8>>,
+    }
+
+    impl MemoryTransactionStore {
+        fn create(&mut self, transaction: &TransitionTransaction) -> Result<()> {
+            if self.records.contains_key(&transaction.transaction_id) {
+                return Err(TransitionTransactionError::AlreadyExists);
+            }
+            self.records.insert(transaction.transaction_id, transaction.encode()?);
+            Ok(())
+        }
+
+        fn load(&self, transaction_id: Uuid) -> Result<TransitionTransaction> {
+            let data = self
+                .records
+                .get(&transaction_id)
+                .ok_or(TransitionTransactionError::NotFound)?;
+            TransitionTransaction::decode(transaction_id, data)
+        }
+
+        fn compare_save(
+            &mut self,
+            transaction_id: Uuid,
+            fence: TransitionTransactionFence,
+            next: TransitionTransactionState,
+        ) -> Result<TransitionTransactionFence> {
+            let mut transaction = self.load(transaction_id)?;
+            let remote_version = (next == TransitionTransactionState::Uploaded)
+                .then(|| TransitionRemoteVersion::versioned(Uuid::new_v4().to_string()));
+            let next_fence = transaction.advance(fence, next, remote_version)?;
+            self.records.insert(transaction_id, transaction.encode()?);
+            Ok(next_fence)
+        }
+
+        fn compare_delete(&mut self, transaction_id: Uuid, fence: TransitionTransactionFence) -> Result<()> {
+            let transaction = self.load(transaction_id)?;
+            if transaction.fence() != fence {
+                return Err(TransitionTransactionError::Fenced);
+            }
+            self.records.remove(&transaction_id);
+            Ok(())
+        }
+    }
+
+    fn source_identity(version_mode: TransitionSourceVersionMode) -> TransitionSourceIdentity {
+        TransitionSourceIdentity {
+            bucket: "source-bucket".to_string(),
+            object: "object/key".to_string(),
+            version_id: (version_mode == TransitionSourceVersionMode::Versioned).then(Uuid::new_v4),
+            data_dir: Uuid::new_v4(),
+            mod_time_unix_nanos: 1_770_000_000_000_000_000,
+            size: 42,
+            etag: "etag".to_string(),
+            version_mode,
+        }
+    }
+
+    fn new_transaction() -> TransitionTransaction {
+        TransitionTransaction::new(TransitionTransactionInit {
+            deployment_id: Uuid::new_v4(),
+            transaction_id: Uuid::new_v4(),
+            owner_epoch: Uuid::new_v4(),
+            write_id: Uuid::new_v4(),
+            source: source_identity(TransitionSourceVersionMode::Versioned),
+            tier_name: "warm-tier".to_string(),
+            backend_fingerprint: BACKEND_FINGERPRINT.to_string(),
+            not_after_unix_nanos: 1_780_000_000_000_000_000,
+        })
+        .expect("valid transaction should be created")
+    }
+
+    fn upload(transaction: &mut TransitionTransaction) -> TransitionTransactionFence {
+        let fence = transaction.fence();
+        transaction
+            .advance(
+                fence,
+                TransitionTransactionState::Uploaded,
+                Some(TransitionRemoteVersion::versioned(Uuid::new_v4().to_string())),
+            )
+            .expect("upload state change should succeed")
+    }
+
+    fn cleanup_proof(transaction: &TransitionTransaction, decision: TransitionCleanupDecision) -> TransitionCleanupProof {
+        TransitionCleanupProof {
+            transaction_id: transaction.transaction_id,
+            write_id: transaction.write_id,
+            remote_object: transaction.remote_object.clone(),
+            remote_version: transaction.remote_version.clone(),
+            backend_fingerprint: transaction.backend_fingerprint.clone(),
+            decision,
+        }
+    }
+
+    #[test]
+    fn remote_version_distinguishes_unknown_unversioned_and_versioned() {
+        assert_eq!(TransitionRemoteVersion::known_from_put_response("").tier_delete_version_id(), None);
+        assert_eq!(
+            TransitionRemoteVersion::known_from_put_response(Uuid::nil().to_string()).tier_delete_version_id(),
+            None
+        );
+
+        let version_id = Uuid::new_v4().to_string();
+        assert_eq!(
+            TransitionRemoteVersion::known_from_put_response(version_id.clone()).tier_delete_version_id(),
+            Some(version_id.as_str())
+        );
+        assert!(TransitionRemoteVersion::unknown().is_unknown());
+    }
+
+    #[test]
+    fn transaction_round_trip_rejects_schema_checksum_unknown_fields_and_wrong_key() {
+        let transaction = new_transaction();
+        let encoded = transaction.encode().expect("transaction should encode");
+        let decoded = TransitionTransaction::decode(transaction.transaction_id, &encoded).expect("transaction should decode");
+        assert_eq!(decoded, transaction);
+
+        let mut wrong_schema: serde_json::Value = serde_json::from_slice(&encoded).expect("encoded transaction should be json");
+        wrong_schema["schema"] = serde_json::Value::String("rustfs-transition-transaction-v2".to_string());
+        let wrong_schema_bytes = serde_json::to_vec(&wrong_schema).expect("wrong schema transaction should encode");
+        assert!(matches!(
+            TransitionTransaction::decode(transaction.transaction_id, &wrong_schema_bytes),
+            Err(TransitionTransactionError::UnsupportedSchema(_))
+        ));
+
+        let mut bad_checksum: serde_json::Value = serde_json::from_slice(&encoded).expect("encoded transaction should be json");
+        bad_checksum["content_sha256"] =
+            serde_json::Value::String("ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff".to_string());
+        let bad_checksum_bytes = serde_json::to_vec(&bad_checksum).expect("bad checksum transaction should encode");
+        assert!(matches!(
+            TransitionTransaction::decode(transaction.transaction_id, &bad_checksum_bytes),
+            Err(TransitionTransactionError::ChecksumMismatch)
+        ));
+
+        let mut unknown_field: serde_json::Value = serde_json::from_slice(&encoded).expect("encoded transaction should be json");
+        unknown_field["unexpected"] = serde_json::Value::Bool(true);
+        let unknown_field_bytes = serde_json::to_vec(&unknown_field).expect("unknown field transaction should encode");
+        assert!(matches!(
+            TransitionTransaction::decode(transaction.transaction_id, &unknown_field_bytes),
+            Err(TransitionTransactionError::Json(_))
+        ));
+
+        let mut nested_unknown_field: serde_json::Value =
+            serde_json::from_slice(&encoded).expect("encoded transaction should be json");
+        nested_unknown_field["transaction"]["remote_version"]["unexpected"] = serde_json::Value::Bool(true);
+        let nested_unknown_field_bytes =
+            serde_json::to_vec(&nested_unknown_field).expect("nested unknown field transaction should encode");
+        assert!(matches!(
+            TransitionTransaction::decode(transaction.transaction_id, &nested_unknown_field_bytes),
+            Err(TransitionTransactionError::Json(_))
+        ));
+
+        assert!(matches!(
+            TransitionTransaction::decode(Uuid::new_v4(), &encoded),
+            Err(TransitionTransactionError::Corrupt("transaction_id does not match record key"))
+        ));
+    }
+
+    #[test]
+    fn invalid_source_and_identity_boundaries_fail_closed() {
+        assert!(matches!(
+            TransitionTransaction::new(TransitionTransactionInit {
+                deployment_id: Uuid::nil(),
+                transaction_id: Uuid::new_v4(),
+                owner_epoch: Uuid::new_v4(),
+                write_id: Uuid::new_v4(),
+                source: source_identity(TransitionSourceVersionMode::Versioned),
+                tier_name: "warm-tier".to_string(),
+                backend_fingerprint: BACKEND_FINGERPRINT.to_string(),
+                not_after_unix_nanos: 1,
+            }),
+            Err(TransitionTransactionError::Corrupt("deployment_id is nil"))
+        ));
+
+        let mut source = source_identity(TransitionSourceVersionMode::Unversioned);
+        source.version_id = Some(Uuid::nil());
+        assert!(matches!(
+            TransitionTransaction::new(TransitionTransactionInit {
+                deployment_id: Uuid::new_v4(),
+                transaction_id: Uuid::new_v4(),
+                owner_epoch: Uuid::new_v4(),
+                write_id: Uuid::new_v4(),
+                source,
+                tier_name: "warm-tier".to_string(),
+                backend_fingerprint: BACKEND_FINGERPRINT.to_string(),
+                not_after_unix_nanos: 1,
+            }),
+            Err(TransitionTransactionError::Corrupt("source version_id is nil"))
+        ));
+
+        let mut source = source_identity(TransitionSourceVersionMode::VersionSuspended);
+        source.version_id = Some(Uuid::new_v4());
+        assert!(matches!(
+            TransitionTransaction::new(TransitionTransactionInit {
+                deployment_id: Uuid::new_v4(),
+                transaction_id: Uuid::new_v4(),
+                owner_epoch: Uuid::new_v4(),
+                write_id: Uuid::new_v4(),
+                source,
+                tier_name: "warm-tier".to_string(),
+                backend_fingerprint: BACKEND_FINGERPRINT.to_string(),
+                not_after_unix_nanos: 1,
+            }),
+            Err(TransitionTransactionError::Corrupt("non-versioned source mode must not carry version_id"))
+        ));
+
+        assert!(
+            TransitionTransaction::new(TransitionTransactionInit {
+                deployment_id: Uuid::new_v4(),
+                transaction_id: Uuid::new_v4(),
+                owner_epoch: Uuid::new_v4(),
+                write_id: Uuid::new_v4(),
+                source: source_identity(TransitionSourceVersionMode::VersionSuspended),
+                tier_name: "warm-tier".to_string(),
+                backend_fingerprint: BACKEND_FINGERPRINT.to_string(),
+                not_after_unix_nanos: 1,
+            })
+            .is_ok()
+        );
+    }
+
+    #[test]
+    fn state_machine_requires_known_remote_version_and_rejects_commit_skip() {
+        let mut transaction = new_transaction();
+        let fence = transaction.fence();
+        assert!(matches!(
+            transaction.advance(fence, TransitionTransactionState::Committed, None),
+            Err(TransitionTransactionError::InvalidStateChange {
+                from: TransitionTransactionState::UploadStarted,
+                to: TransitionTransactionState::Committed,
+            })
+        ));
+
+        assert!(matches!(
+            transaction.advance(fence, TransitionTransactionState::Uploaded, Some(TransitionRemoteVersion::unknown())),
+            Err(TransitionTransactionError::Corrupt("uploaded state requires known remote version"))
+        ));
+
+        let uploaded_fence = upload(&mut transaction);
+        let local_commit_fence = transaction
+            .advance(uploaded_fence, TransitionTransactionState::LocalCommitStarted, None)
+            .expect("local commit start should succeed");
+        let committed_fence = transaction
+            .advance(local_commit_fence, TransitionTransactionState::Committed, None)
+            .expect("commit should succeed");
+        assert_eq!(committed_fence.revision, 4);
+        assert_eq!(transaction.state, TransitionTransactionState::Committed);
+    }
+
+    #[test]
+    fn cleanup_pending_requires_exact_proof_and_state_specific_decision() {
+        let mut transaction = new_transaction();
+        let uploaded_fence = upload(&mut transaction);
+
+        let mut wrong_remote = cleanup_proof(&transaction, TransitionCleanupDecision::UploadAbortedBeforeLocalCommit);
+        wrong_remote.remote_object.push_str("-different");
+        assert!(matches!(
+            transaction.mark_cleanup_pending(uploaded_fence, wrong_remote),
+            Err(TransitionTransactionError::Fenced)
+        ));
+
+        let proof = cleanup_proof(&transaction, TransitionCleanupDecision::UploadAbortedBeforeLocalCommit);
+        let cleanup_fence = transaction
+            .mark_cleanup_pending(uploaded_fence, proof)
+            .expect("uploaded transaction should enter cleanup pending");
+        assert_eq!(cleanup_fence.revision, 3);
+        assert_eq!(transaction.state, TransitionTransactionState::CleanupPending);
+
+        let mut local_commit_transaction = new_transaction();
+        let uploaded_fence = upload(&mut local_commit_transaction);
+        let local_commit_fence = local_commit_transaction
+            .advance(uploaded_fence, TransitionTransactionState::LocalCommitStarted, None)
+            .expect("local commit start should succeed");
+        let mut changed_source = local_commit_transaction.source.clone();
+        changed_source.etag = "different-etag".to_string();
+        let proof = cleanup_proof(
+            &local_commit_transaction,
+            TransitionCleanupDecision::SourceReconciledUnchanged {
+                observed_source: changed_source,
+            },
+        );
+        assert!(matches!(
+            local_commit_transaction.mark_cleanup_pending(local_commit_fence, proof),
+            Err(TransitionTransactionError::InvalidStateChange {
+                from: TransitionTransactionState::LocalCommitStarted,
+                to: TransitionTransactionState::CleanupPending,
+            })
+        ));
+    }
+
+    #[test]
+    fn upload_outcome_unknown_can_only_cleanup_after_version_recovery() {
+        let mut transaction = new_transaction();
+        let unknown_fence = transaction
+            .advance(transaction.fence(), TransitionTransactionState::UploadOutcomeUnknown, None)
+            .expect("upload outcome unknown should be recorded");
+
+        let mut proof = cleanup_proof(&transaction, TransitionCleanupDecision::RemoteVersionRecoveredAfterCancellation);
+        proof.remote_version = TransitionRemoteVersion::unknown();
+        assert!(matches!(
+            transaction.mark_cleanup_pending(unknown_fence, proof),
+            Err(TransitionTransactionError::Corrupt("cleanup proof requires known remote version"))
+        ));
+
+        let proof = TransitionCleanupProof {
+            remote_version: TransitionRemoteVersion::versioned(Uuid::new_v4().to_string()),
+            decision: TransitionCleanupDecision::RemoteVersionRecoveredAfterCancellation,
+            ..cleanup_proof(&transaction, TransitionCleanupDecision::RemoteVersionRecoveredAfterCancellation)
+        };
+        let cleanup_fence = transaction
+            .mark_cleanup_pending(unknown_fence, proof)
+            .expect("recovered remote version should allow cleanup pending");
+        assert_eq!(cleanup_fence.revision, 3);
+        assert_eq!(transaction.state, TransitionTransactionState::CleanupPending);
+    }
+
+    #[test]
+    fn stale_fence_blocks_second_actor_and_compare_delete() {
+        let mut store = MemoryTransactionStore::default();
+        let transaction = new_transaction();
+        let transaction_id = transaction.transaction_id;
+        let first_actor_fence = transaction.fence();
+        let second_actor_fence = transaction.fence();
+        store.create(&transaction).expect("initial create should succeed");
+
+        let uploaded_fence = store
+            .compare_save(transaction_id, first_actor_fence, TransitionTransactionState::Uploaded)
+            .expect("first actor should advance upload");
+        assert!(matches!(
+            store.compare_save(transaction_id, second_actor_fence, TransitionTransactionState::LocalCommitStarted),
+            Err(TransitionTransactionError::Fenced)
+        ));
+
+        let local_commit_fence = store
+            .compare_save(transaction_id, uploaded_fence, TransitionTransactionState::LocalCommitStarted)
+            .expect("fresh fence should advance local commit");
+        assert!(matches!(
+            store.compare_delete(transaction_id, uploaded_fence),
+            Err(TransitionTransactionError::Fenced)
+        ));
+        store
+            .compare_delete(transaction_id, local_commit_fence)
+            .expect("fresh fence should delete transaction");
+        assert!(matches!(store.load(transaction_id), Err(TransitionTransactionError::NotFound)));
+    }
+
+    #[test]
+    fn decode_rejects_oversized_record_before_parsing() {
+        let transaction = new_transaction();
+        let oversized = vec![b' '; MAX_TRANSITION_TRANSACTION_SIZE + 1];
+        assert!(matches!(
+            TransitionTransaction::decode(transaction.transaction_id, &oversized),
+            Err(TransitionTransactionError::Corrupt("encoded transaction exceeds maximum size"))
+        ));
+    }
+}