From 8bbf7e98c9bc85333b5afe560500c1f11d522aad Mon Sep 17 00:00:00 2001 From: trinity-1686a Date: Sat, 7 Feb 2026 13:26:56 +0100 Subject: [PATCH] add internals for supporting transactional updates --- src/model/admin_token_table.rs | 1 + src/model/bucket_alias_table.rs | 1 + src/model/bucket_table.rs | 1 + src/model/index_counter.rs | 1 + src/model/k2v/item_table.rs | 1 + src/model/key_table.rs | 1 + src/model/s3/block_ref_table.rs | 1 + src/model/s3/mpu_table.rs | 1 + src/model/s3/object_table.rs | 52 ++++++++++++++++++++++++ src/model/s3/version_table.rs | 1 + src/table/schema.rs | 8 ++++ src/table/table.rs | 71 +++++++++++++++++++++++++++++++++ src/util/error.rs | 3 ++ src/util/keyed_mutex.rs | 41 +++++++++++++++++++ src/util/lib.rs | 1 + 15 files changed, 185 insertions(+) create mode 100644 src/util/keyed_mutex.rs diff --git a/src/model/admin_token_table.rs b/src/model/admin_token_table.rs index 0af8ec78..4946842b 100644 --- a/src/model/admin_token_table.rs +++ b/src/model/admin_token_table.rs @@ -171,6 +171,7 @@ impl TableSchema for AdminApiTokenTable { type S = String; type E = AdminApiToken; type Filter = KeyFilter; + type Precondition = (); fn matches_filter(entry: &Self::E, filter: &Self::Filter) -> bool { match filter { diff --git a/src/model/bucket_alias_table.rs b/src/model/bucket_alias_table.rs index 276d0d1c..30ea593c 100644 --- a/src/model/bucket_alias_table.rs +++ b/src/model/bucket_alias_table.rs @@ -61,6 +61,7 @@ impl TableSchema for BucketAliasTable { type S = String; type E = BucketAlias; type Filter = DeletedFilter; + type Precondition = (); fn matches_filter(entry: &Self::E, filter: &Self::Filter) -> bool { filter.apply(entry.is_deleted()) diff --git a/src/model/bucket_table.rs b/src/model/bucket_table.rs index 7317c36f..7809f5da 100644 --- a/src/model/bucket_table.rs +++ b/src/model/bucket_table.rs @@ -371,6 +371,7 @@ impl TableSchema for BucketTable { type S = Uuid; type E = Bucket; type Filter = DeletedFilter; + type Precondition = (); fn matches_filter(entry: &Self::E, filter: &Self::Filter) -> bool { filter.apply(entry.is_deleted()) diff --git a/src/model/index_counter.rs b/src/model/index_counter.rs index 50abdec3..bb583e91 100644 --- a/src/model/index_counter.rs +++ b/src/model/index_counter.rs @@ -145,6 +145,7 @@ impl TableSchema for CounterTable { type S = T::CS; type E = CounterEntry; type Filter = (DeletedFilter, Vec); + type Precondition = (); fn matches_filter(entry: &Self::E, filter: &Self::Filter) -> bool { if filter.0 == DeletedFilter::Any { diff --git a/src/model/k2v/item_table.rs b/src/model/k2v/item_table.rs index 9e3ba5a5..4e92c5f6 100644 --- a/src/model/k2v/item_table.rs +++ b/src/model/k2v/item_table.rs @@ -219,6 +219,7 @@ impl TableSchema for K2VItemTable { type S = String; type E = K2VItem; type Filter = ItemFilter; + type Precondition = (); fn updated( &self, diff --git a/src/model/key_table.rs b/src/model/key_table.rs index ded9832d..addc6ac7 100644 --- a/src/model/key_table.rs +++ b/src/model/key_table.rs @@ -255,6 +255,7 @@ impl TableSchema for KeyTable { type S = String; type E = Key; type Filter = KeyFilter; + type Precondition = (); fn matches_filter(entry: &Self::E, filter: &Self::Filter) -> bool { match filter { diff --git a/src/model/s3/block_ref_table.rs b/src/model/s3/block_ref_table.rs index 67db109e..4da920d0 100644 --- a/src/model/s3/block_ref_table.rs +++ b/src/model/s3/block_ref_table.rs @@ -65,6 +65,7 @@ impl TableSchema for BlockRefTable { type S = Uuid; type E = BlockRef; type Filter = DeletedFilter; + type Precondition = (); fn updated( &self, diff --git a/src/model/s3/mpu_table.rs b/src/model/s3/mpu_table.rs index 135e1888..6353dd8e 100644 --- a/src/model/s3/mpu_table.rs +++ b/src/model/s3/mpu_table.rs @@ -179,6 +179,7 @@ impl TableSchema for MultipartUploadTable { type S = EmptyKey; type E = MultipartUpload; type Filter = DeletedFilter; + type Precondition = (); fn updated( &self, diff --git a/src/model/s3/object_table.rs b/src/model/s3/object_table.rs index ebee67f8..303487bf 100644 --- a/src/model/s3/object_table.rs +++ b/src/model/s3/object_table.rs @@ -769,6 +769,16 @@ pub enum ObjectFilter { IsUploading { check_multipart: Option }, } +#[derive(Clone, Debug, Serialize, Deserialize)] +pub enum ObjectPrecondition { + /// Match if the object doesn't exist (or is a tombstone), and this "new" + /// version is newer than the local copy + IsAbsent, + /// Match if the object's last version has a matching etag, and this "new" + /// version is newer than the local copy + HasEtag(String), +} + impl TableSchema for ObjectTable { const TABLE_NAME: &'static str = "object"; @@ -776,6 +786,7 @@ impl TableSchema for ObjectTable { type S = String; type E = Object; type Filter = ObjectFilter; + type Precondition = ObjectPrecondition; fn updated( &self, @@ -872,6 +883,47 @@ impl TableSchema for ObjectTable { .any(|v| v.is_uploading(*check_multipart)), } } + + fn matches_condition(local_entry: Option<&Self::E>, new_entry: &Self::E, condition: &Self::Precondition) -> bool { + let Some(last_new_version) = new_entry.versions().into_iter().rev().find(|version| version.is_complete()) else { + // transactional update must be made with a complete version + return false; + }; + + let last_stored_version = local_entry.and_then(|local_entry| local_entry.versions() + .into_iter() + .rev() + .find(|version| version.is_complete())); + + if last_stored_version.map_or(false, |last_stored_version| last_stored_version.cmp_key() > last_new_version.cmp_key()) { + // our update is older than the newest complete version, we can discard it + return false + } + + match condition { + ObjectPrecondition::IsAbsent => { + let Some(last_stored_version) = last_stored_version else { + // no version stored, the object doesn't exist + return true + }; + !last_stored_version.is_data() + } + ObjectPrecondition::HasEtag(etag) => { + let Some(last_stored_version) = last_stored_version else { + // no version stored, the object doesn't exist + return false + }; + match &last_stored_version.state { + ObjectVersionState::Complete(ObjectVersionData::Inline(meta, _)) + | ObjectVersionState::Complete(ObjectVersionData::FirstBlock(meta, _)) => { + &meta.etag == etag + }, + // last version was a tombstone + _ => false + } + } + } + } } impl CountedItem for Object { diff --git a/src/model/s3/version_table.rs b/src/model/s3/version_table.rs index 45be5af8..485c7298 100644 --- a/src/model/s3/version_table.rs +++ b/src/model/s3/version_table.rs @@ -205,6 +205,7 @@ impl TableSchema for VersionTable { type S = EmptyKey; type E = Version; type Filter = DeletedFilter; + type Precondition = (); fn updated( &self, diff --git a/src/table/schema.rs b/src/table/schema.rs index fc1a465e..999ee2c5 100644 --- a/src/table/schema.rs +++ b/src/table/schema.rs @@ -85,6 +85,9 @@ pub trait TableSchema: Send + Sync + 'static { /// (e.g. filter out deleted entries) type Filter: Clone + Serialize + for<'de> Deserialize<'de> + Send + Sync + 'static; + /// A precondition that should be checked before some update operation + type Precondition: Clone + Serialize + for<'de> Deserialize<'de> + Send + Sync + 'static; + /// Actions triggered by data changing in a table. If such actions /// include updates to the local database that should be applied /// atomically with the item update itself, a db transaction is @@ -100,4 +103,9 @@ pub trait TableSchema: Send + Sync + 'static { } fn matches_filter(entry: &Self::E, filter: &Self::Filter) -> bool; + + fn matches_condition(local_entry: Option<&Self::E>, new_entry: &Self::E, condition: &Self::Precondition) -> bool { + let _ = (local_entry, new_entry, condition); + false + } } diff --git a/src/table/table.rs b/src/table/table.rs index 8ddd8378..4f8719de 100644 --- a/src/table/table.rs +++ b/src/table/table.rs @@ -18,6 +18,7 @@ use garage_util::data::*; use garage_util::error::Error; use garage_util::metrics::RecordDuration; use garage_util::migrate::Migrate; +use garage_util::keyed_mutex::KeyedMutex; use garage_rpc::rpc_helper::QuorumSetResultTracker; use garage_rpc::system::System; @@ -40,6 +41,7 @@ pub struct Table { pub syncer: Arc>, gc: Arc>, endpoint: Arc, Self>>, + keyed_mutex: KeyedMutex<(Hash, Vec)>, } #[derive(Serialize, Deserialize)] @@ -59,6 +61,19 @@ pub(crate) enum TableRpc { }, Update(Vec>), + CompareUpdate(CompareUpdate) +} + +#[derive(Serialize, Deserialize, Clone)] +pub(crate) struct CompareUpdate { + condition: Precondition, + /// Nodes that were not contacted yet, and should be + node_ids: Vec, + /// number of success still required, when this reach zero, enough nodes + /// agree that we can persist the transaction + success_required: usize, + /// Actual value to store + value: Arc, } impl Rpc for TableRpc { @@ -89,6 +104,7 @@ impl Table { gc, syncer, endpoint, + keyed_mutex: KeyedMutex::new(), }); table.endpoint.set_handler(table.clone()); @@ -491,6 +507,57 @@ impl Table { bytes.map(|b| self.data.decode_entry(&b)).transpose() } + async fn handle_compare_update(self: &Arc, update: &CompareUpdate) -> Result<(), Error> { + let mut update = update.clone(); + + let new_entry = self.data.decode_entry(update.value.as_slice())?; + let mutex_handle = self.keyed_mutex.lock((new_entry.partition_key().hash(), new_entry.sort_key().sort_key().to_vec())); + let local_value = self.get_local(new_entry.partition_key(), new_entry.sort_key())?; + if F::matches_condition(local_value.as_ref(), &new_entry, &update.condition) { + update.success_required -= 1; + } else { + // there is no point in maintaining a lock, the condition didn't hold for us + // it might still hold for enough nodes that this is a valid update though + drop(mutex_handle) + } + if update.success_required == 0 { + self.data.update_entry(update.value.as_slice())?; + let this = self.clone(); + tokio::spawn(async move { + this.system + .rpc_helper() + .try_call_many( + &this.endpoint, + &update.node_ids, + TableRpc::::Update(vec![update.value]), + RequestStrategy::with_priority(PRIO_NORMAL), + ).await + }); + return Ok(()) + } else { + // the node that called us thought this could succeed, + // the only way we don't do a single loop iteration + // is the condition was false for us, so set that as default + // exit condition + let mut last_error = Error::PreconditionFailed; + while update.node_ids.len() >= update.success_required { + let next_node = update.node_ids.pop().unwrap(/* node_ids >= success_required > 0, pop always succeed*/); + match self.system.rpc_helper().call(&self.endpoint, next_node, TableRpc::::CompareUpdate(update.clone()), RequestStrategy::with_priority(PRIO_NORMAL)).await { + Ok(_) => { + self.data.update_entry(update.value.as_slice())?; + }, + Err(Error::PreconditionFailed) => { + return Err(Error::PreconditionFailed); + } + Err(e) => { + last_error = e; + }, + } + } + return Err(last_error) + } + } + // =============== UTILITY FUNCTION FOR CLIENT OPERATIONS =============== async fn repair_on_read(&self, who: &[Uuid], what: F::E) -> Result<(), Error> { @@ -539,6 +606,10 @@ impl EndpointHandler> for Table self.data.update_many(pairs)?; Ok(TableRpc::Ok) } + TableRpc::CompareUpdate(compare_update) => { + self.handle_compare_update(compare_update).await?; + Ok(TableRpc::Ok) + } m => Err(Error::unexpected_rpc_message(m)), } } diff --git a/src/util/error.rs b/src/util/error.rs index 043d0dce..26377ab9 100644 --- a/src/util/error.rs +++ b/src/util/error.rs @@ -71,6 +71,9 @@ pub enum Error { #[error("{0}")] Message(String), + + #[error("Precondition failed")] + PreconditionFailed, } impl Error { diff --git a/src/util/keyed_mutex.rs b/src/util/keyed_mutex.rs new file mode 100644 index 00000000..824f1b89 --- /dev/null +++ b/src/util/keyed_mutex.rs @@ -0,0 +1,41 @@ +use std::collections::HashSet; +use std::hash::Hash; + +use tokio::sync::watch::Sender as WatchSender; + +pub struct KeyedMutex { + state: WatchSender>, +} + +impl KeyedMutex { + pub fn new() -> Self { + KeyedMutex { + state: WatchSender::new(HashSet::new()), + } + } + + pub async fn lock(&self, key: K) -> LockGuard<'_, K> { + let mut receiver = self.state.subscribe(); + loop { + if self.state.send_if_modified(|set| set.insert(key.clone())) { + return LockGuard { + lock: self, + key, + } + } + // this can't error because we still hold a sender + let _ = receiver.wait_for(|set| !set.contains(&key)).await; + } + } +} + +pub struct LockGuard<'a, K: Hash + Eq > { + lock: &'a KeyedMutex, + key: K, +} + +impl<'a, K: Hash + Eq> Drop for LockGuard<'a, K> { + fn drop(&mut self) { + self.lock.state.send_modify(|set| assert!(set.remove(&self.key), "unlocked mutex that wasn't locked")) + } +} diff --git a/src/util/lib.rs b/src/util/lib.rs index 8b035ff0..81c673b4 100644 --- a/src/util/lib.rs +++ b/src/util/lib.rs @@ -10,6 +10,7 @@ pub mod data; pub mod encode; pub mod error; pub mod forwarded_headers; +pub mod keyed_mutex; pub mod metrics; pub mod migrate; pub mod persister;