add internals for supporting transactional updates

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 {

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())

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())

src/model/index_counter.rs

@@ -145,6 +145,7 @@ impl<T: CountedItem> TableSchema for CounterTable<T> {
 	type S = T::CS;
 	type E = CounterEntry<T>;
 	type Filter = (DeletedFilter, Vec<Uuid>);
+        type Precondition = ();
 
 	fn matches_filter(entry: &Self::E, filter: &Self::Filter) -> bool {
 		if filter.0 == DeletedFilter::Any {

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,

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 {

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,

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,

src/model/s3/object_table.rs

@@ -769,6 +769,16 @@ pub enum ObjectFilter {
 	IsUploading { check_multipart: Option<bool> },
 }
 
+#[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 {

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,

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
+        }
 }

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<F: TableSchema, R: TableReplication> {
 	pub syncer: Arc<TableSyncer<F, R>>,
 	gc: Arc<TableGc<F, R>>,
 	endpoint: Arc<Endpoint<TableRpc<F>, Self>>,
+        keyed_mutex: KeyedMutex<(Hash, Vec<u8>)>,
 }
 
 #[derive(Serialize, Deserialize)]
@@ -59,6 +61,19 @@ pub(crate) enum TableRpc<F: TableSchema> {
 	},
 
 	Update(Vec<Arc<ByteBuf>>),
+        CompareUpdate(CompareUpdate<F::Precondition>)
+}
+
+#[derive(Serialize, Deserialize, Clone)]
+pub(crate) struct CompareUpdate<Precondition> {
+    condition: Precondition,
+    /// Nodes that were not contacted yet, and should be
+    node_ids: Vec<Uuid>,
+    /// 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<ByteBuf>,
 }
 
 impl<F: TableSchema> Rpc for TableRpc<F> {
@@ -89,6 +104,7 @@ impl<F: TableSchema, R: TableReplication> Table<F, R> {
 			gc,
 			syncer,
 			endpoint,
+                        keyed_mutex: KeyedMutex::new(),
 		});
 
 		table.endpoint.set_handler(table.clone());
@@ -491,6 +507,57 @@ impl<F: TableSchema, R: TableReplication> Table<F, R> {
 		bytes.map(|b| self.data.decode_entry(&b)).transpose()
 	}
 
+        async fn handle_compare_update(self: &Arc<Self>, update: &CompareUpdate<F::Precondition>) -> 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::<F>::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::<F>::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<F: TableSchema, R: TableReplication> EndpointHandler<TableRpc<F>> 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)),
 		}
 	}

src/util/error.rs

@@ -71,6 +71,9 @@ pub enum Error {
 
 	#[error("{0}")]
 	Message(String),
+
+        #[error("Precondition failed")]
+        PreconditionFailed,
 }
 
 impl Error {

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<K> {
+    state: WatchSender<HashSet<K>>,
+}
+
+impl<K: Hash + Eq + Clone> KeyedMutex<K> {
+    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<K>,
+    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"))
+    }
+}

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;