From 302dd42d38040735bea94bdb96728b1260f6bd71 Mon Sep 17 00:00:00 2001 From: houseme Date: Mon, 20 Jul 2026 22:10:11 +0800 Subject: [PATCH] fix(ilm): add transition transaction foundation (#5065) Co-authored-by: heihutu --- crates/ecstore/src/bucket/lifecycle/mod.rs | 1 + .../lifecycle/transition_transaction.rs | 925 ++++++++++++++++++ 2 files changed, 926 insertions(+) create mode 100644 crates/ecstore/src/bucket/lifecycle/transition_transaction.rs diff --git a/crates/ecstore/src/bucket/lifecycle/mod.rs b/crates/ecstore/src/bucket/lifecycle/mod.rs index ffd3f3801..311b63006 100644 --- a/crates/ecstore/src/bucket/lifecycle/mod.rs +++ b/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; diff --git a/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs b/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs new file mode 100644 index 000000000..00c684363 --- /dev/null +++ b/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 = std::result::Result; + +#[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, +} + +impl TransitionRemoteVersion { + pub fn known_from_put_response(version_id: impl Into) -> 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) -> 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, + 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 { + 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, + ) -> Result { + 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 { + 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> { + 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 { + 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 { + 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>, + } + + 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 { + 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 { + 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")) + )); + } +}