Fuzzing for K2VItem Crdt (#1438)

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

fuzz/Cargo.toml

@@ -14,7 +14,7 @@ libfuzzer-sys = { workspace = true }
 garage_db.workspace = true
 garage_table.workspace = true
 garage_util.workspace = true
-garage_model = { workspace = true, default-features = false, features = ["arbitrary"] }
+garage_model = { workspace = true, default-features = false, features = ["arbitrary", "k2v"] }
 
 [[bin]]
 name = "version_crdt"
@@ -64,3 +64,10 @@ path = "fuzz_targets/bucket_alias_crdt.rs"
 test = false
 doc = false
 bench = false
+
+[[bin]]
+name = "k2v_item_crdt"
+path = "fuzz_targets/k2v_item_crdt.rs"
+test = false
+doc = false
+bench = false

fuzz/fuzz_targets/k2v_item_crdt.rs

@@ -0,0 +1,36 @@
+#![no_main]
+
+use std::collections::BTreeMap;
+
+use garage_fuzz::check_crdt_laws;
+use garage_model::k2v::item_table::{DvvsEntry, DvvsValue, K2VItem};
+use libfuzzer_sys::fuzz_target;
+
+// Timestamps are encoded as `(ts << 32) | shift` so that items built with different
+// shifts (0, 1, 2) have disjoint timestamp spaces that still interleave in the sorted merge.
+fn make(raw: BTreeMap<u64, (u32, BTreeMap<u32, DvvsValue>)>, shift: u32) -> K2VItem {
+	let shift = shift as u64;
+	let items = raw
+		.into_iter()
+		.map(|(node, (t_discard, values))| {
+			let entry = DvvsEntry::from_raw(
+				(t_discard as u64) << 32 | shift,
+				values
+					.into_iter()
+					.map(|(ts, v)| ((ts as u64) << 32 | shift, v))
+					.collect(),
+			);
+			(node, entry)
+		})
+		.collect();
+	K2VItem::with_raw_items(items)
+}
+
+fuzz_target!(|inputs: (
+	BTreeMap<u64, (u32, BTreeMap<u32, DvvsValue>)>,
+	BTreeMap<u64, (u32, BTreeMap<u32, DvvsValue>)>,
+	BTreeMap<u64, (u32, BTreeMap<u32, DvvsValue>)>,
+)| {
+	let (a, b, c) = inputs;
+	check_crdt_laws(make(a, 0), make(b, 1), make(c, 2));
+});

src/model/k2v/item_table.rs

@@ -45,6 +45,7 @@ mod v08 {
 	}
 
 	#[derive(PartialEq, Eq, Clone, Debug, Serialize, Deserialize)]
+	#[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))]
 	pub enum DvvsValue {
 		Value(#[serde(with = "serde_bytes")] Vec<u8>),
 		Deleted,
@@ -131,9 +132,26 @@ impl K2VItem {
 			ent.discard();
 		}
 	}
+
+	pub fn with_raw_items(items: BTreeMap<K2VNodeId, DvvsEntry>) -> Self {
+		let mut item = K2VItem {
+			partition: K2VItemPartition {
+				bucket_id: [0u8; 32].into(),
+				partition_key: String::new(),
+			},
+			sort_key: String::new(),
+			items,
+		};
+		item.discard();
+		item
+	}
 }
 
 impl DvvsEntry {
+	pub fn from_raw(t_discard: u64, values: Vec<(u64, DvvsValue)>) -> Self {
+		DvvsEntry { t_discard, values }
+	}
+
 	fn max_time(&self) -> u64 {
 		self.values
 			.iter()
@@ -162,15 +180,26 @@ impl Crdt for K2VItem {
 
 impl Crdt for DvvsEntry {
 	fn merge(&mut self, other: &Self) {
-		self.t_discard = std::cmp::max(self.t_discard, other.t_discard);
-		self.discard();
-
-		let t_max = self.max_time();
-		for (vt, vv) in other.values.iter() {
-			if *vt > t_max {
-				self.values.push((*vt, vv.clone()));
+		let mut slf = std::mem::take(&mut self.values).into_iter().peekable();
+		let mut otr = other.values.iter().peekable();
+		while let (Some((slf_t, _)), Some((otr_t, _))) = (slf.peek(), otr.peek()) {
+			match slf_t.cmp(otr_t) {
+				std::cmp::Ordering::Less => {
+					self.values.push(slf.next().unwrap());
+				}
+				std::cmp::Ordering::Equal => {
+					self.values.push(slf.next().unwrap());
+					otr.next();
+				}
+				std::cmp::Ordering::Greater => {
+					self.values.push(otr.next().unwrap().clone());
+				}
 			}
 		}
+		self.values.extend(slf);
+		self.values.extend(otr.cloned());
+		self.t_discard = std::cmp::max(self.t_discard, other.t_discard);
+		self.discard();
 	}
 }