Merge pull request 'K2V: provide monotonic reads by default, with a flag to opt-out' (#1452) from Armael/garage:read_repair into main-v2

Reviewed-on: https://git.deuxfleurs.fr/Deuxfleurs/garage/pulls/1452
Reviewed-by: Alex <lx@deuxfleurs.fr>

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 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
+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 /<bucket>/<partition key>?sort_key=<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                                                                                |

src/api/k2v/batch.rs

@@ -5,12 +5,13 @@ 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::*;
 
 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 +48,13 @@ pub async fn handle_read_batch(
 	ctx: ReqCtx,
 	req: Request<ReqBody>,
 ) -> Result<Response<ResBody>, Error> {
+	let monotonic_read = is_monotonic_read(&req)?;
 	let queries = req.into_body().json::<Vec<ReadBatchQuery>>().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 +69,7 @@ pub async fn handle_read_batch(
 async fn handle_read_batch_query(
 	ctx: &ReqCtx,
 	query: ReadBatchQuery,
+	monotonic_read: K2VMonotonicRead,
 ) -> Result<ReadBatchResponse, Error> {
 	let ReqCtx {
 		garage, bucket_id, ..
@@ -90,12 +93,13 @@ 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 = 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 {
 			Some(i) => (vec![ReadBatchResponseItem::from(i)], false, None),
 			None => (vec![], false, None),
@@ -110,6 +114,7 @@ async fn handle_read_batch_query(
 			query.limit,
 			Some(filter),
 			EnumerationOrder::from_reverse(query.reverse),
+			monotonic_read,
 		)
 		.await?;
 
@@ -218,6 +223,7 @@ async fn handle_delete_batch_query(
 			None,
 			Some(filter),
 			EnumerationOrder::Forward,
+			K2VMonotonicRead::NonMonotonic,
 		)
 		.await?;
 		assert!(!more);
@@ -260,6 +266,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::<PollRangeQuery>().await?;
@@ -281,6 +288,7 @@ pub(crate) async fn handle_poll_range(
 			},
 			query.seen_marker,
 			timeout_msec,
+			monotonic_read,
 		)
 		.await
 		.map_err(pass_helper_error)?;

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

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?;
 

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::*;
 
@@ -12,6 +13,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 +25,21 @@ pub(crate) fn parse_causality_token(s: &str) -> Result<CausalContext, Error> {
 	CausalContext::parse(s).ok_or(Error::InvalidCausalityToken)
 }
 
+pub(crate) fn is_monotonic_read(req: &Request<ReqBody>) -> Result<K2VMonotonicRead, Error> {
+	let v_opt = req
+		.headers()
+		.get(X_GARAGE_NON_MONOTONIC_READ)
+		.map(|s| s.to_str())
+		.transpose()?;
+
+	match v_opt {
+		Some("true") => Ok(K2VMonotonicRead::NonMonotonic),
+		// Reads are monotonic by default
+		Some("false") | None => Ok(K2VMonotonicRead::Monotonic),
+		Some(s) => Err(Error::InvalidNonMonotonicRead(s.to_string())),
+	}
+}
+
 impl ReturnFormat {
 	pub fn from(req: &Request<ReqBody>) -> Result<Self, Error> {
 		let accept = match req.headers().get(header::ACCEPT) {
@@ -108,21 +125,26 @@ 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 = 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)?;
 
 	format.make_response(&item)
 }
@@ -214,6 +236,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 +253,7 @@ pub async fn handle_poll_item(
 			sort_key,
 			causal_context,
 			timeout_msec,
+			monotonic_read,
 		)
 		.await?;
 

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<F>(
 	limit: Option<u64>,
 	filter: Option<F::Filter>,
 	enumeration_order: EnumerationOrder,
+	monotonic_read: K2VMonotonicRead,
 ) -> Result<(Vec<F::E>, bool, Option<String>), Error>
 where
 	F: TableSchema<S = String> + '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();
 

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};
@@ -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,6 +216,7 @@ impl K2VRpcHandler {
 		sort_key: String,
 		causal_context: CausalContext,
 		timeout_msec: u64,
+		monotonic_read: K2VMonotonicRead,
 	) -> Result<Option<K2VItem>, Error> {
 		let poll_key = PollKey {
 			partition: K2VItemPartition {
@@ -244,17 +251,35 @@ impl K2VRpcHandler {
 		};
 
 		let mut resp: Option<K2VItem> = 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 == K2VMonotonicRead::Monotonic && not_all_same {
+				self.item_table.repair_on_read(&nodes, &[v]).await?;
 			}
 		}
 
@@ -266,6 +291,7 @@ impl K2VRpcHandler {
 		range: PollRange,
 		seen_str: Option<String>,
 		timeout_msec: u64,
+		monotonic_read: K2VMonotonicRead,
 	) -> Result<Option<(BTreeMap<String, K2VItem>, String)>, HelperError> {
 		let has_seen_marker = seen_str.is_some();
 
@@ -343,24 +369,53 @@ impl K2VRpcHandler {
 
 		// Take all returned items into account to produce the response.
 		let mut new_items = BTreeMap::<String, K2VItem>::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 propagate 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 == K2VMonotonicRead::Monotonic && !to_repair.is_empty() {
+			let to_repair: Vec<_> = to_repair
+				.into_iter()
+				.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 {
 			Ok(None)
 		} else {

src/table/table.rs

@@ -293,7 +293,26 @@ impl<F: TableSchema, R: TableReplication> Table<F, R> {
 		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<Self>,
+		partition_key: &F::P,
+		sort_key: &F::S,
+	) -> Result<Option<F::E>, 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<F: TableSchema, R: TableReplication> Table<F, R> {
 		self: &Arc<Self>,
 		partition_key: &F::P,
 		sort_key: &F::S,
+		monotonic_read: bool,
 	) -> Result<Option<F::E>, Error> {
 		let hash = partition_key.hash();
 		let who = self.data.replication.read_nodes(&hash)?;
@@ -326,34 +346,37 @@ impl<F: TableSchema, R: TableReplication> Table<F, R> {
 
 		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();
-				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]).await?;
 			}
 		}
 
@@ -378,6 +401,36 @@ impl<F: TableSchema, R: TableReplication> Table<F, R> {
 				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<Self>,
+		partition_key: &F::P,
+		begin_sort_key: Option<F::S>,
+		filter: Option<F::Filter>,
+		limit: usize,
+		enumeration_order: EnumerationOrder,
+	) -> Result<Vec<F::E>, 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))
@@ -395,6 +448,7 @@ impl<F: TableSchema, R: TableReplication> Table<F, R> {
 		filter: Option<F::Filter>,
 		limit: usize,
 		enumeration_order: EnumerationOrder,
+		monotonic_read: bool,
 	) -> Result<Vec<F::E>, Error> {
 		let hash = partition_key.hash();
 		let who = self.data.replication.read_nodes(&hash)?;
@@ -421,11 +475,26 @@ impl<F: TableSchema, R: TableReplication> Table<F, R> {
 
 		let mut ret: BTreeMap<Vec<u8>, 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<u8>, Vec<F::E>> = 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,28 +503,19 @@ impl<F: TableSchema, R: TableReplication> Table<F, R> {
 							}
 						}
 						None => {
-							ret.insert(entry_key, entry);
+							ret.insert(entry_key.clone(), entry);
 						}
 					}
 				}
-			} else {
-				return Err(Error::unexpected_rpc_message(resp));
 			}
 		}
 
-		if !to_repair.is_empty() {
-			let self2 = self.clone();
-			let to_repair = to_repair
+		if monotonic_read && !to_repair.is_empty() {
+			let to_repair: Vec<_> = to_repair
 				.into_iter()
-				.map(|k| ret.get(&k).unwrap().clone())
-				.collect::<Vec<_>>();
-			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);
-					}
-				}
-			});
+				.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`
@@ -493,14 +553,17 @@ impl<F: TableSchema, R: TableReplication> Table<F, R> {
 
 	// =============== UTILITY FUNCTION FOR CLIENT OPERATIONS ===============
 
-	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::<Result<Vec<_>, Error>>()?;
 		self.system
 			.rpc_helper()
 			.try_call_many(
 				&self.endpoint,
 				who,
-				TableRpc::<F>::Update(vec![what_enc]),
+				TableRpc::<F>::Update(what_enc),
 				RequestStrategy::with_priority(PRIO_NORMAL).with_quorum(who.len()),
 			)
 			.await?;