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