From 2d7c0a6087e1d70e739774c0b9a34e9573a223fe Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Arma=C3=ABl=20Gu=C3=A9neau?= Date: Wed, 13 May 2026 14:37:05 +0200 Subject: [PATCH 1/8] table: also apply repair-on-read the first time a value is set --- src/table/table.rs | 64 +++++++++++++++++++++++++++++++--------------- 1 file changed, 43 insertions(+), 21 deletions(-) diff --git a/src/table/table.rs b/src/table/table.rs index 8ddd8378..3fc0d5ee 100644 --- a/src/table/table.rs +++ b/src/table/table.rs @@ -326,25 +326,34 @@ impl Table { let mut ret = None; let mut not_all_same = false; - for resp in resps { - if let TableRpc::ReadEntryResponse(value) = resp { - if let Some(v_bytes) = value { - let v = self.data.decode_entry(v_bytes.as_slice())?; - ret = match ret { - None => Some(v), - Some(mut x) => { - if x != v { - not_all_same = true; - x.merge(&v); + { + let mut vals_nb = 0; + for resp in &resps { + if let TableRpc::ReadEntryResponse(value) = resp { + if let Some(v_bytes) = value { + vals_nb += 1; + let v = self.data.decode_entry(v_bytes.as_slice())?; + ret = match ret { + None => Some(v), + Some(mut x) => { + if x != v { + not_all_same = true; + x.merge(&v); + } + Some(x) } - Some(x) } } + } else { + return Err(Error::Message("Invalid return value to read".to_string())); } - } else { - return Err(Error::Message("Invalid return value to read".to_string())); + } + // Only some nodes store this value; we must propagate it during repair + if vals_nb < resps.len() { + not_all_same = true; } } + if let Some(ret_entry) = &ret { if not_all_same { let self2 = self.clone(); @@ -421,11 +430,26 @@ impl Table { let mut ret: BTreeMap, F::E> = BTreeMap::new(); let mut to_repair = BTreeSet::new(); - for resp in resps { - if let TableRpc::Update(entries) = resp { - for entry_bytes in entries.iter() { - let entry = self.data.decode_entry(entry_bytes.as_slice())?; - let entry_key = self.data.tree_key(entry.partition_key(), entry.sort_key()); + { + let mut all_entries: BTreeMap, Vec> = BTreeMap::new(); + for resp in &resps { + if let TableRpc::Update(entries) = resp { + for entry_bytes in entries.iter() { + let entry = self.data.decode_entry(entry_bytes.as_slice())?; + let entry_key = self.data.tree_key(entry.partition_key(), entry.sort_key()); + all_entries.entry(entry_key).or_default().push(entry); + } + } else { + return Err(Error::unexpected_rpc_message(resp)); + } + } + for (entry_key, entries) in all_entries { + // Only some nodes store this entry; we must propagate it during repair + if entries.len() < resps.len() { + to_repair.insert(entry_key.clone()); + } + // Merge all entries for this key together + for entry in entries { match ret.get_mut(&entry_key) { Some(e) => { if *e != entry { @@ -434,12 +458,10 @@ impl Table { } } None => { - ret.insert(entry_key, entry); + ret.insert(entry_key.clone(), entry); } } } - } else { - return Err(Error::unexpected_rpc_message(resp)); } } From 5500f1c412f66987309df7d9e4456c3c1510b5dd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Arma=C3=ABl=20Gu=C3=A9neau?= Date: Wed, 13 May 2026 14:38:03 +0200 Subject: [PATCH 2/8] table: disable asynchronous repair-on-read; add get_*_monotonic functions for synchronous repair-on-read --- src/table/table.rs | 81 +++++++++++++++++++++++++++++++++------------- 1 file changed, 59 insertions(+), 22 deletions(-) diff --git a/src/table/table.rs b/src/table/table.rs index 3fc0d5ee..0a767fe2 100644 --- a/src/table/table.rs +++ b/src/table/table.rs @@ -293,7 +293,26 @@ impl Table { let span = tracer.start(format!("{} get", F::TABLE_NAME)); let res = self - .get_internal(partition_key, sort_key) + .get_internal(partition_key, sort_key, false) + .bound_record_duration(&self.data.metrics.get_request_duration) + .with_context(Context::current_with_span(span)) + .await?; + + self.data.metrics.get_request_counter.add(1); + + Ok(res) + } + + pub async fn get_monotonic( + self: &Arc, + partition_key: &F::P, + sort_key: &F::S, + ) -> Result, Error> { + let tracer = opentelemetry::global::tracer("garage_table"); + let span = tracer.start(format!("{} get_monotonic", F::TABLE_NAME)); + + let res = self + .get_internal(partition_key, sort_key, true) .bound_record_duration(&self.data.metrics.get_request_duration) .with_context(Context::current_with_span(span)) .await?; @@ -307,6 +326,7 @@ impl Table { self: &Arc, partition_key: &F::P, sort_key: &F::S, + monotonic_read: bool, ) -> Result, Error> { let hash = partition_key.hash(); let who = self.data.replication.read_nodes(&hash)?; @@ -355,14 +375,8 @@ impl Table { } if let Some(ret_entry) = &ret { - if not_all_same { - let self2 = self.clone(); - let ent2 = ret_entry.clone(); - tokio::spawn(async move { - if let Err(e) = self2.repair_on_read(&who[..], ent2).await { - warn!("Error doing repair on read: {}", e); - } - }); + if monotonic_read && not_all_same { + self.repair_on_read(&who, ret_entry.clone()).await?; } } @@ -387,6 +401,36 @@ impl Table { filter, limit, enumeration_order, + false, + ) + .bound_record_duration(&self.data.metrics.get_request_duration) + .with_context(Context::current_with_span(span)) + .await?; + + self.data.metrics.get_request_counter.add(1); + + Ok(res) + } + + pub async fn get_range_monotonic( + self: &Arc, + partition_key: &F::P, + begin_sort_key: Option, + filter: Option, + limit: usize, + enumeration_order: EnumerationOrder, + ) -> Result, Error> { + let tracer = opentelemetry::global::tracer("garage_table"); + let span = tracer.start(format!("{} get_range_monotonic", F::TABLE_NAME)); + + let res = self + .get_range_internal( + partition_key, + begin_sort_key, + filter, + limit, + enumeration_order, + true, ) .bound_record_duration(&self.data.metrics.get_request_duration) .with_context(Context::current_with_span(span)) @@ -404,6 +448,7 @@ impl Table { filter: Option, limit: usize, enumeration_order: EnumerationOrder, + monotonic_read: bool, ) -> Result, Error> { let hash = partition_key.hash(); let who = self.data.replication.read_nodes(&hash)?; @@ -465,19 +510,11 @@ impl Table { } } - if !to_repair.is_empty() { - let self2 = self.clone(); - let to_repair = to_repair - .into_iter() - .map(|k| ret.get(&k).unwrap().clone()) - .collect::>(); - tokio::spawn(async move { - for v in to_repair { - if let Err(e) = self2.repair_on_read(&who[..], v).await { - warn!("Error doing repair on read: {}", e); - } - } - }); + if monotonic_read && !to_repair.is_empty() { + let to_repair = to_repair.into_iter().map(|k| ret.get(&k).unwrap().clone()); + for v in to_repair { + self.repair_on_read(&who, v).await?; + } } // At this point, the `ret` btreemap might contain more than `limit` From 3e25914210d2044d24785eba90f7e5d18a42a47f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Arma=C3=ABl=20Gu=C3=A9neau?= Date: Wed, 13 May 2026 14:39:27 +0200 Subject: [PATCH 3/8] K2V: provide monotonic reads by default, with a flag to opt-out This performs synchronous repair-on-read for K2V reads: - uses the new get_*_monotonic operations in the table module - implements repair-on-read for the K2V-specific poll operations --- src/api/k2v/batch.rs | 20 ++++++----- src/api/k2v/error.rs | 8 ++++- src/api/k2v/item.rs | 48 +++++++++++++++++++------- src/model/k2v/rpc.rs | 82 +++++++++++++++++++++++++++++++++++--------- src/table/table.rs | 2 +- 5 files changed, 121 insertions(+), 39 deletions(-) diff --git a/src/api/k2v/batch.rs b/src/api/k2v/batch.rs index 5f38cce1..db0f4a80 100644 --- a/src/api/k2v/batch.rs +++ b/src/api/k2v/batch.rs @@ -10,7 +10,7 @@ use garage_api_common::helpers::*; use crate::api_server::{ReqBody, ResBody}; use crate::error::*; -use crate::item::parse_causality_token; +use crate::item::{is_monotonic_read, parse_causality_token}; use crate::range::read_range; pub async fn handle_insert_batch( @@ -47,12 +47,13 @@ pub async fn handle_read_batch( ctx: ReqCtx, req: Request, ) -> Result, Error> { + let monotonic_read = is_monotonic_read(&req)?; let queries = req.into_body().json::>().await?; let resp_results = futures::future::join_all( queries .into_iter() - .map(|q| handle_read_batch_query(&ctx, q)), + .map(|q| handle_read_batch_query(&ctx, q, monotonic_read)), ) .await; @@ -67,6 +68,7 @@ pub async fn handle_read_batch( async fn handle_read_batch_query( ctx: &ReqCtx, query: ReadBatchQuery, + monotonic_read: bool, ) -> Result { let ReqCtx { garage, bucket_id, .. @@ -90,12 +92,12 @@ async fn handle_read_batch_query( .start .as_ref() .ok_or_bad_request("start should be specified if single_item is set")?; - let item = garage - .k2v - .item_table - .get(&partition, sk) - .await? - .filter(|e| K2VItemTable::matches_filter(e, &filter)); + let item = if monotonic_read { + garage.k2v.item_table.get_monotonic(&partition, sk).await? + } else { + garage.k2v.item_table.get(&partition, sk).await? + } + .filter(|e| K2VItemTable::matches_filter(e, &filter)); match item { Some(i) => (vec![ReadBatchResponseItem::from(i)], false, None), None => (vec![], false, None), @@ -260,6 +262,7 @@ pub(crate) async fn handle_poll_range( let ReqCtx { garage, bucket_id, .. } = ctx; + let monotonic_read = is_monotonic_read(&req)?; use garage_model::k2v::sub::PollRange; let query = req.into_body().json::().await?; @@ -281,6 +284,7 @@ pub(crate) async fn handle_poll_range( }, query.seen_marker, timeout_msec, + monotonic_read, ) .await .map_err(pass_helper_error)?; diff --git a/src/api/k2v/error.rs b/src/api/k2v/error.rs index 797eb868..f8913c43 100644 --- a/src/api/k2v/error.rs +++ b/src/api/k2v/error.rs @@ -44,6 +44,10 @@ pub enum Error { #[error("Invalid causality token")] InvalidCausalityToken, + /// Invalid parameter for x-garage-non-monotonic-read + #[error("Invalid X-Garage-Non-Monotonic-Read value: {0}")] + InvalidNonMonotonicRead(String), + /// The client asked for an invalid return format (invalid Accept header) #[error("Not acceptable: {0}")] NotAcceptable(String), @@ -85,6 +89,7 @@ impl Error { Error::InvalidBase64(_) => "InvalidBase64", Error::InvalidUtf8Str(_) => "InvalidUtf8String", Error::InvalidCausalityToken => "CausalityToken", + Error::InvalidNonMonotonicRead(_) => "InvalidNonMonotonicRead", Error::InvalidDigest(_) => "InvalidDigest", } } @@ -101,7 +106,8 @@ impl ApiError for Error { | Error::InvalidBase64(_) | Error::InvalidUtf8Str(_) | Error::InvalidDigest(_) - | Error::InvalidCausalityToken => StatusCode::BAD_REQUEST, + | Error::InvalidCausalityToken + | Error::InvalidNonMonotonicRead(_) => StatusCode::BAD_REQUEST, } } diff --git a/src/api/k2v/item.rs b/src/api/k2v/item.rs index 2482d7cd..7916bbc7 100644 --- a/src/api/k2v/item.rs +++ b/src/api/k2v/item.rs @@ -12,6 +12,7 @@ use crate::api_server::{ReqBody, ResBody}; use crate::error::*; pub const X_GARAGE_CAUSALITY_TOKEN: &str = "X-Garage-Causality-Token"; +pub const X_GARAGE_NON_MONOTONIC_READ: &str = "X-Garage-Non-Monotonic-Read"; pub enum ReturnFormat { Json, @@ -23,6 +24,23 @@ pub(crate) fn parse_causality_token(s: &str) -> Result { CausalContext::parse(s).ok_or(Error::InvalidCausalityToken) } +pub(crate) fn is_monotonic_read(req: &Request) -> Result { + let v_opt = req + .headers() + .get(X_GARAGE_NON_MONOTONIC_READ) + .map(|s| s.to_str()) + .transpose()?; + + // The header is set to 'true' if the read may be *non*-monotonic, + // and this function returns if we must do a *monotonic* read; hence + // the boolean negation. + match v_opt { + Some("true") => Ok(false), + Some("false") | None => Ok(true), + Some(s) => Err(Error::InvalidNonMonotonicRead(s.to_string())), + } +} + impl ReturnFormat { pub fn from(req: &Request) -> Result { let accept = match req.headers().get(header::ACCEPT) { @@ -108,21 +126,23 @@ pub async fn handle_read_item( let ReqCtx { garage, bucket_id, .. } = &ctx; - + let monotonic_read = is_monotonic_read(req)?; let format = ReturnFormat::from(req)?; + let partition_key = K2VItemPartition { + bucket_id: *bucket_id, + partition_key: partition_key.to_string(), + }; - let item = garage - .k2v - .item_table - .get( - &K2VItemPartition { - bucket_id: *bucket_id, - partition_key: partition_key.to_string(), - }, - sort_key, - ) - .await? - .ok_or(Error::NoSuchKey)?; + let item = if monotonic_read { + garage + .k2v + .item_table + .get_monotonic(&partition_key, sort_key) + .await? + } else { + garage.k2v.item_table.get(&partition_key, sort_key).await? + } + .ok_or(Error::NoSuchKey)?; format.make_response(&item) } @@ -214,6 +234,7 @@ pub async fn handle_poll_item( let ReqCtx { garage, bucket_id, .. } = &ctx; + let monotonic_read = is_monotonic_read(req)?; let format = ReturnFormat::from(req)?; let causal_context = @@ -230,6 +251,7 @@ pub async fn handle_poll_item( sort_key, causal_context, timeout_msec, + monotonic_read, ) .await?; diff --git a/src/model/k2v/rpc.rs b/src/model/k2v/rpc.rs index 8fcf8309..84782adc 100644 --- a/src/model/k2v/rpc.rs +++ b/src/model/k2v/rpc.rs @@ -5,7 +5,7 @@ //! node does not process the entry directly, as this would //! mean the vector clock gets much larger than needed). -use std::collections::{BTreeMap, HashMap}; +use std::collections::{BTreeMap, BTreeSet, HashMap}; use std::convert::TryInto; use std::sync::{Arc, Mutex, MutexGuard}; use std::time::{Duration, Instant}; @@ -210,6 +210,7 @@ impl K2VRpcHandler { sort_key: String, causal_context: CausalContext, timeout_msec: u64, + monotonic_read: bool, ) -> Result, Error> { let poll_key = PollKey { partition: K2VItemPartition { @@ -244,17 +245,35 @@ impl K2VRpcHandler { }; let mut resp: Option = None; - for v in resps { - match v { - K2VRpc::PollItemResponse(Some(x)) => { - if let Some(y) = &mut resp { - y.merge(&x); - } else { - resp = Some(x); + let mut not_all_same = false; + { + let mut vals_nb = 0; + let resps_nb = resps.len(); + for v in resps { + match v { + K2VRpc::PollItemResponse(Some(x)) => { + vals_nb += 1; + if let Some(y) = &mut resp { + if *y != x { + not_all_same = true; + y.merge(&x); + } + } else { + resp = Some(x); + } } + K2VRpc::PollItemResponse(None) => (), + v => return Err(Error::unexpected_rpc_message(v)), } - K2VRpc::PollItemResponse(None) => (), - v => return Err(Error::unexpected_rpc_message(v)), + } + if vals_nb < resps_nb { + not_all_same = true; + } + } + + if let Some(v) = &resp { + if monotonic_read && not_all_same { + self.item_table.repair_on_read(&nodes, v.clone()).await?; } } @@ -266,6 +285,7 @@ impl K2VRpcHandler { range: PollRange, seen_str: Option, timeout_msec: u64, + monotonic_read: bool, ) -> Result, String)>, HelperError> { let has_seen_marker = seen_str.is_some(); @@ -343,21 +363,51 @@ impl K2VRpcHandler { // Take all returned items into account to produce the response. let mut new_items = BTreeMap::::new(); - for v in resps { - if let K2VRpc::PollRangeResponse(node, items) = v { - seen.mark_seen_node_items(node, items.iter()); + let mut to_repair = BTreeSet::new(); + { + let mut all_items: BTreeMap<_, Vec<_>> = BTreeMap::new(); + let resps_nb = resps.len(); + for v in resps { + if let K2VRpc::PollRangeResponse(node, items) = v { + seen.mark_seen_node_items(node, items.iter()); + for item in items.into_iter() { + all_items + .entry(item.sort_key.clone()) + .or_default() + .push(item); + } + } else { + return Err(Error::unexpected_rpc_message(v).into()); + } + } + for (item_key, items) in all_items { + // Only some nodes store this item; we must propage it during repair + if items.len() < resps_nb { + to_repair.insert(item_key.clone()); + } + // Merge all items for this key together for item in items.into_iter() { match new_items.get_mut(&item.sort_key) { Some(ent) => { - ent.merge(&item); + if *ent != item { + ent.merge(&item); + to_repair.insert(item.sort_key.clone()); + } } None => { new_items.insert(item.sort_key.clone(), item); } } } - } else { - return Err(Error::unexpected_rpc_message(v).into()); + } + } + + if monotonic_read && !to_repair.is_empty() { + let to_repair = to_repair + .into_iter() + .map(|k| new_items.get(&k).unwrap().clone()); + for v in to_repair { + self.item_table.repair_on_read(&nodes, v).await? } } diff --git a/src/table/table.rs b/src/table/table.rs index 0a767fe2..446f7d6e 100644 --- a/src/table/table.rs +++ b/src/table/table.rs @@ -552,7 +552,7 @@ impl Table { // =============== UTILITY FUNCTION FOR CLIENT OPERATIONS =============== - async fn repair_on_read(&self, who: &[Uuid], what: F::E) -> Result<(), Error> { + pub async fn repair_on_read(&self, who: &[Uuid], what: F::E) -> Result<(), Error> { let what_enc = Arc::new(ByteBuf::from(what.encode()?)); self.system .rpc_helper() From d0f89068c639db4934e3fef406966256c70a52f5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Arma=C3=ABl=20Gu=C3=A9neau?= Date: Wed, 13 May 2026 18:23:32 +0200 Subject: [PATCH 4/8] repair_on_read: send multiple items to update in a single RPC --- src/model/k2v/rpc.rs | 11 +++++------ src/table/table.rs | 20 ++++++++++++-------- 2 files changed, 17 insertions(+), 14 deletions(-) diff --git a/src/model/k2v/rpc.rs b/src/model/k2v/rpc.rs index 84782adc..abdf80ef 100644 --- a/src/model/k2v/rpc.rs +++ b/src/model/k2v/rpc.rs @@ -273,7 +273,7 @@ impl K2VRpcHandler { if let Some(v) = &resp { if monotonic_read && not_all_same { - self.item_table.repair_on_read(&nodes, v.clone()).await?; + self.item_table.repair_on_read(&nodes, &[&v]).await?; } } @@ -403,12 +403,11 @@ impl K2VRpcHandler { } if monotonic_read && !to_repair.is_empty() { - let to_repair = to_repair + let to_repair: Vec<_> = to_repair .into_iter() - .map(|k| new_items.get(&k).unwrap().clone()); - for v in to_repair { - self.item_table.repair_on_read(&nodes, v).await? - } + .map(|k| new_items.get(&k).unwrap()) + .collect(); + self.item_table.repair_on_read(&nodes, &to_repair).await? } if new_items.is_empty() && has_seen_marker { diff --git a/src/table/table.rs b/src/table/table.rs index 446f7d6e..d6f4018a 100644 --- a/src/table/table.rs +++ b/src/table/table.rs @@ -376,7 +376,7 @@ impl Table { if let Some(ret_entry) = &ret { if monotonic_read && not_all_same { - self.repair_on_read(&who, ret_entry.clone()).await?; + self.repair_on_read(&who, &[&ret_entry]).await?; } } @@ -511,10 +511,11 @@ impl Table { } if monotonic_read && !to_repair.is_empty() { - let to_repair = to_repair.into_iter().map(|k| ret.get(&k).unwrap().clone()); - for v in to_repair { - self.repair_on_read(&who, v).await?; - } + let to_repair: Vec<_> = to_repair + .into_iter() + .map(|k| ret.get(&k).unwrap()) + .collect(); + self.repair_on_read(&who, &to_repair).await?; } // At this point, the `ret` btreemap might contain more than `limit` @@ -552,14 +553,17 @@ impl Table { // =============== UTILITY FUNCTION FOR CLIENT OPERATIONS =============== - pub async fn repair_on_read(&self, who: &[Uuid], what: F::E) -> Result<(), Error> { - let what_enc = Arc::new(ByteBuf::from(what.encode()?)); + pub async fn repair_on_read(&self, who: &[Uuid], what: &[&F::E]) -> Result<(), Error> { + let what_enc = what + .iter() + .map(|v| Ok(Arc::new(ByteBuf::from(v.encode()?)))) + .collect::, Error>>()?; self.system .rpc_helper() .try_call_many( &self.endpoint, who, - TableRpc::::Update(vec![what_enc]), + TableRpc::::Update(what_enc), RequestStrategy::with_priority(PRIO_NORMAL).with_quorum(who.len()), ) .await?; From 555e0826a28d67dbfd79971dc006504c8ca72dd5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Arma=C3=ABl=20Gu=C3=A9neau?= Date: Wed, 13 May 2026 18:36:45 +0200 Subject: [PATCH 5/8] define a separate enum for the monotonic/non-monotonic read flag --- src/api/k2v/batch.rs | 12 +++++++----- src/api/k2v/item.rs | 30 ++++++++++++++++-------------- src/model/k2v/rpc.rs | 14 ++++++++++---- 3 files changed, 33 insertions(+), 23 deletions(-) diff --git a/src/api/k2v/batch.rs b/src/api/k2v/batch.rs index db0f4a80..8a2d1196 100644 --- a/src/api/k2v/batch.rs +++ b/src/api/k2v/batch.rs @@ -5,6 +5,7 @@ use serde::{Deserialize, Serialize}; use garage_table::{EnumerationOrder, TableSchema}; use garage_model::k2v::item_table::*; +use garage_model::k2v::rpc::K2VMonotonicRead; use garage_api_common::helpers::*; @@ -68,7 +69,7 @@ pub async fn handle_read_batch( async fn handle_read_batch_query( ctx: &ReqCtx, query: ReadBatchQuery, - monotonic_read: bool, + monotonic_read: K2VMonotonicRead, ) -> Result { let ReqCtx { garage, bucket_id, .. @@ -92,10 +93,11 @@ async fn handle_read_batch_query( .start .as_ref() .ok_or_bad_request("start should be specified if single_item is set")?; - let item = if monotonic_read { - garage.k2v.item_table.get_monotonic(&partition, sk).await? - } else { - garage.k2v.item_table.get(&partition, sk).await? + let item = match monotonic_read { + K2VMonotonicRead::Monotonic => { + garage.k2v.item_table.get_monotonic(&partition, sk).await? + } + K2VMonotonicRead::NonMonotonic => garage.k2v.item_table.get(&partition, sk).await?, } .filter(|e| K2VItemTable::matches_filter(e, &filter)); match item { diff --git a/src/api/k2v/item.rs b/src/api/k2v/item.rs index 7916bbc7..8b5ad453 100644 --- a/src/api/k2v/item.rs +++ b/src/api/k2v/item.rs @@ -5,6 +5,7 @@ use hyper::{Request, Response, StatusCode}; use garage_model::k2v::causality::*; use garage_model::k2v::item_table::*; +use garage_model::k2v::rpc::K2VMonotonicRead; use garage_api_common::helpers::*; @@ -24,19 +25,17 @@ pub(crate) fn parse_causality_token(s: &str) -> Result { CausalContext::parse(s).ok_or(Error::InvalidCausalityToken) } -pub(crate) fn is_monotonic_read(req: &Request) -> Result { +pub(crate) fn is_monotonic_read(req: &Request) -> Result { let v_opt = req .headers() .get(X_GARAGE_NON_MONOTONIC_READ) .map(|s| s.to_str()) .transpose()?; - // The header is set to 'true' if the read may be *non*-monotonic, - // and this function returns if we must do a *monotonic* read; hence - // the boolean negation. match v_opt { - Some("true") => Ok(false), - Some("false") | None => Ok(true), + Some("true") => Ok(K2VMonotonicRead::NonMonotonic), + // Reads are monotonic by default + Some("false") | None => Ok(K2VMonotonicRead::Monotonic), Some(s) => Err(Error::InvalidNonMonotonicRead(s.to_string())), } } @@ -133,14 +132,17 @@ pub async fn handle_read_item( partition_key: partition_key.to_string(), }; - let item = if monotonic_read { - garage - .k2v - .item_table - .get_monotonic(&partition_key, sort_key) - .await? - } else { - garage.k2v.item_table.get(&partition_key, sort_key).await? + let item = match monotonic_read { + K2VMonotonicRead::Monotonic => { + garage + .k2v + .item_table + .get_monotonic(&partition_key, sort_key) + .await? + } + K2VMonotonicRead::NonMonotonic => { + garage.k2v.item_table.get(&partition_key, sort_key).await? + } } .ok_or(Error::NoSuchKey)?; diff --git a/src/model/k2v/rpc.rs b/src/model/k2v/rpc.rs index abdf80ef..09931abd 100644 --- a/src/model/k2v/rpc.rs +++ b/src/model/k2v/rpc.rs @@ -38,6 +38,12 @@ const POLL_RANGE_EXTRA_DELAY: Duration = Duration::from_millis(200); const TIMESTAMP_KEY: &[u8] = b"timestamp"; +#[derive(Debug, Clone, Copy, PartialEq)] +pub enum K2VMonotonicRead { + Monotonic, + NonMonotonic, +} + /// RPC messages for K2V #[derive(Debug, Serialize, Deserialize)] enum K2VRpc { @@ -210,7 +216,7 @@ impl K2VRpcHandler { sort_key: String, causal_context: CausalContext, timeout_msec: u64, - monotonic_read: bool, + monotonic_read: K2VMonotonicRead, ) -> Result, Error> { let poll_key = PollKey { partition: K2VItemPartition { @@ -272,7 +278,7 @@ impl K2VRpcHandler { } if let Some(v) = &resp { - if monotonic_read && not_all_same { + if monotonic_read == K2VMonotonicRead::Monotonic && not_all_same { self.item_table.repair_on_read(&nodes, &[&v]).await?; } } @@ -285,7 +291,7 @@ impl K2VRpcHandler { range: PollRange, seen_str: Option, timeout_msec: u64, - monotonic_read: bool, + monotonic_read: K2VMonotonicRead, ) -> Result, String)>, HelperError> { let has_seen_marker = seen_str.is_some(); @@ -402,7 +408,7 @@ impl K2VRpcHandler { } } - if monotonic_read && !to_repair.is_empty() { + if monotonic_read == K2VMonotonicRead::Monotonic && !to_repair.is_empty() { let to_repair: Vec<_> = to_repair .into_iter() .map(|k| new_items.get(&k).unwrap()) From a159c1c483d9305f0f42455eb212c5f6f44d34ad Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Arma=C3=ABl=20Gu=C3=A9neau?= Date: Wed, 13 May 2026 18:52:18 +0200 Subject: [PATCH 6/8] add missing repair-on-read for k2v range reads --- src/api/k2v/batch.rs | 2 ++ src/api/k2v/index.rs | 2 ++ src/api/k2v/range.rs | 36 +++++++++++++++++++++++++++--------- 3 files changed, 31 insertions(+), 9 deletions(-) diff --git a/src/api/k2v/batch.rs b/src/api/k2v/batch.rs index 8a2d1196..fffcaf83 100644 --- a/src/api/k2v/batch.rs +++ b/src/api/k2v/batch.rs @@ -114,6 +114,7 @@ async fn handle_read_batch_query( query.limit, Some(filter), EnumerationOrder::from_reverse(query.reverse), + monotonic_read, ) .await?; @@ -222,6 +223,7 @@ async fn handle_delete_batch_query( None, Some(filter), EnumerationOrder::Forward, + K2VMonotonicRead::NonMonotonic, ) .await?; assert!(!more); diff --git a/src/api/k2v/index.rs b/src/api/k2v/index.rs index 5188c32f..bb3a628d 100644 --- a/src/api/k2v/index.rs +++ b/src/api/k2v/index.rs @@ -4,6 +4,7 @@ use serde::Serialize; use garage_table::util::*; use garage_model::k2v::item_table::{BYTES, CONFLICTS, ENTRIES, VALUES}; +use garage_model::k2v::rpc::K2VMonotonicRead; use garage_api_common::helpers::*; @@ -40,6 +41,7 @@ pub async fn handle_read_index( limit, Some((DeletedFilter::NotDeleted, node_id_vec)), EnumerationOrder::from_reverse(reverse), + K2VMonotonicRead::NonMonotonic, ) .await?; diff --git a/src/api/k2v/range.rs b/src/api/k2v/range.rs index dc1bdaac..bb851041 100644 --- a/src/api/k2v/range.rs +++ b/src/api/k2v/range.rs @@ -4,6 +4,8 @@ use std::sync::Arc; +use garage_model::k2v::rpc::K2VMonotonicRead; + use garage_table::replication::TableShardedReplication; use garage_table::*; @@ -23,6 +25,7 @@ pub(crate) async fn read_range( limit: Option, filter: Option, enumeration_order: EnumerationOrder, + monotonic_read: K2VMonotonicRead, ) -> Result<(Vec, bool, Option), Error> where F: TableSchema + 'static, @@ -53,15 +56,30 @@ where 1000, limit.map(|x| x as usize).unwrap_or(usize::MAX - 10) - entries.len() + 2, ); - let get_ret = table - .get_range( - partition_key, - start.clone(), - filter.clone(), - n_get, - enumeration_order, - ) - .await?; + let get_ret = match monotonic_read { + K2VMonotonicRead::Monotonic => { + table + .get_range_monotonic( + partition_key, + start.clone(), + filter.clone(), + n_get, + enumeration_order, + ) + .await? + } + K2VMonotonicRead::NonMonotonic => { + table + .get_range( + partition_key, + start.clone(), + filter.clone(), + n_get, + enumeration_order, + ) + .await? + } + }; let get_ret_len = get_ret.len(); From 3c7990027a86f2c4abc181bbee3fee61f3f1199d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Arma=C3=ABl=20Gu=C3=A9neau?= Date: Wed, 13 May 2026 19:14:31 +0200 Subject: [PATCH 7/8] K2V: update docs to add read monotonicity guarantees & flags --- doc/drafts/k2v-spec.md | 47 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 47 insertions(+) diff --git a/doc/drafts/k2v-spec.md b/doc/drafts/k2v-spec.md index b16628e2..32be024d 100644 --- a/doc/drafts/k2v-spec.md +++ b/doc/drafts/k2v-spec.md @@ -166,6 +166,25 @@ that map to zeroes. Note that we need to filter out values from nodes that are no longer part of the cluster layout, as when nodes are removed they won't necessarily have had the time to set their counters to zero. +### Consistency guarantees + +K2V provides the following consistency guarantees: + +**Read after Write**. After a write has been acknowledged (the request returned +successfully), a subsequent read is guaranteed to contain the value that was +written. + +**Monotonic Reads**. Two sequencial reads will return values in an order that is +consistent with the order in which they are written (e.g. by concurrent writes). +For example, consider a scenario where a value is set initially set to 0 and a +request writing 1 is performed. Doing two subsequent reads concurrently with the +write is guaranteed to return either `0`, `0` or `0`,`1` or `1`,`1`, but not +`1`,`0`. + +It is also possible to perform non-monotonic reads (allowing this last +behavior), which are slightly faster than monotonic reads. This is done by +passing a dedicated flag to read operations (see the endpoints documentation). + ## Important details **THIS SECTION CONTAINS A FEW WARNINGS ON THE K2V API WHICH ARE IMPORTANT @@ -210,6 +229,12 @@ Query parameters: |------------|---------------|----------------------------------| | `sort_key` | **mandatory** | The sort key of the item to read | +Headers: + +| name | default value | meaning | +|-------------------------------|---------------|------------------------------------------| +| `X-Garage-Non-Monotonic-Read` | `false` | Whether to allow for non-monotonic reads | + Returns the item with specified partition key and sort key. Values can be returned in either of two ways: @@ -325,6 +350,12 @@ Query parameters: The timeout can be set to any number of seconds, with a maximum of 600 seconds (10 minutes). +Headers: + +| name | default value | meaning | +|-------------------------------|---------------|------------------------------------------| +| `X-Garage-Non-Monotonic-Read` | `false` | Whether to allow for non-monotonic reads | + **InsertItem: `PUT //?sort_key=`** @@ -521,6 +552,14 @@ HTTP/1.1 204 NO CONTENT Batch read of triplets in a bucket. +Headers: + +| name | default value | meaning | +|-------------------------------|---------------|------------------------------------------| +| `X-Garage-Non-Monotonic-Read` | `false` | Whether to allow for non-monotonic reads | + +Body: + The request body is a JSON list of searches, that each specify a range of items to get (to get single items, set `singleItem` to `true`). A search is a JSON struct with the following fields: @@ -711,6 +750,14 @@ HTTP/1.1 200 OK Polls a range of items for changes. +Headers: + +| name | default value | meaning | +|-------------------------------|---------------|------------------------------------------| +| `X-Garage-Non-Monotonic-Read` | `false` | Whether to allow for non-monotonic reads | + +Body: + The query body is a JSON object consisting of the following fields: | name | default value | meaning | From 4f9faeb282ed647f493da6041cd78971620bb52e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Arma=C3=ABl=20Gu=C3=A9neau?= Date: Wed, 13 May 2026 20:32:24 +0200 Subject: [PATCH 8/8] minor fixes --- doc/drafts/k2v-spec.md | 2 +- src/model/k2v/rpc.rs | 6 +++--- src/table/table.rs | 2 +- 3 files changed, 5 insertions(+), 5 deletions(-) diff --git a/doc/drafts/k2v-spec.md b/doc/drafts/k2v-spec.md index 32be024d..2db49a6b 100644 --- a/doc/drafts/k2v-spec.md +++ b/doc/drafts/k2v-spec.md @@ -174,7 +174,7 @@ K2V provides the following consistency guarantees: successfully), a subsequent read is guaranteed to contain the value that was written. -**Monotonic Reads**. Two sequencial reads will return values in an order that is +**Monotonic Reads**. Two sequential reads will return values in an order that is consistent with the order in which they are written (e.g. by concurrent writes). For example, consider a scenario where a value is set initially set to 0 and a request writing 1 is performed. Doing two subsequent reads concurrently with the diff --git a/src/model/k2v/rpc.rs b/src/model/k2v/rpc.rs index 09931abd..7d2bede3 100644 --- a/src/model/k2v/rpc.rs +++ b/src/model/k2v/rpc.rs @@ -279,7 +279,7 @@ impl K2VRpcHandler { if let Some(v) = &resp { if monotonic_read == K2VMonotonicRead::Monotonic && not_all_same { - self.item_table.repair_on_read(&nodes, &[&v]).await?; + self.item_table.repair_on_read(&nodes, &[v]).await?; } } @@ -387,7 +387,7 @@ impl K2VRpcHandler { } } for (item_key, items) in all_items { - // Only some nodes store this item; we must propage it during repair + // Only some nodes store this item; we must propagate it during repair if items.len() < resps_nb { to_repair.insert(item_key.clone()); } @@ -413,7 +413,7 @@ impl K2VRpcHandler { .into_iter() .map(|k| new_items.get(&k).unwrap()) .collect(); - self.item_table.repair_on_read(&nodes, &to_repair).await? + self.item_table.repair_on_read(&nodes, &to_repair).await?; } if new_items.is_empty() && has_seen_marker { diff --git a/src/table/table.rs b/src/table/table.rs index d6f4018a..5b723e7f 100644 --- a/src/table/table.rs +++ b/src/table/table.rs @@ -376,7 +376,7 @@ impl Table { if let Some(ret_entry) = &ret { if monotonic_read && not_all_same { - self.repair_on_read(&who, &[&ret_entry]).await?; + self.repair_on_read(&who, &[ret_entry]).await?; } }