fix(epoxy): use vbare

Author: MasterPtato

Cargo.lock

@@ -35,6 +35,7 @@ tracing.workspace = true
 universaldb.workspace = true
 url.workspace = true
 uuid.workspace = true
+vbare.workspace = true
 
 [dev-dependencies]
 gas.workspace = true

engine/packages/epoxy/Cargo.toml

@@ -1,17 +1,9 @@
 use anyhow::Result;
-use epoxy_protocol::protocol;
+use epoxy_protocol::{PROTOCOL_VERSION, protocol, versioned};
 use serde::{Deserialize, Serialize};
 use universaldb::prelude::*;
 use universaldb::tuple::Versionstamp;
-
-#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Hash)]
-pub struct CommittedValue {
-	// NOTE: An empty value may exist for cached entries to denote the value was not found on any datacenter
-	// and cached as such.
-	pub value: Vec<u8>,
-	pub version: u64,
-	pub mutable: bool,
-}
+use vbare::OwnedVersionedData;
 
 /// In-flight accepted proposal state stored under `kv/{key}/accepted`.
 ///
@@ -44,14 +36,15 @@ impl KvValueKey {
 }
 
 impl FormalKey for KvValueKey {
-	type Value = CommittedValue;
+	type Value = protocol::CommittedValue;
 
 	fn deserialize(&self, raw: &[u8]) -> Result<Self::Value> {
-		serde_bare::from_slice(raw).map_err(Into::into)
+		versioned::CommittedValue::deserialize_with_embedded_version(raw)
 	}
 
 	fn serialize(&self, value: Self::Value) -> Result<Vec<u8>> {
-		serde_bare::to_vec(&value).map_err(Into::into)
+		versioned::CommittedValue::wrap_latest(value)
+			.serialize_with_embedded_version(PROTOCOL_VERSION)
 	}
 }
 
@@ -260,14 +253,14 @@ impl KvOptimisticCacheKey {
 }
 
 impl FormalKey for KvOptimisticCacheKey {
-	type Value = CommittedValue;
+	type Value = protocol::CachedValue;
 
 	fn deserialize(&self, raw: &[u8]) -> Result<Self::Value> {
-		serde_bare::from_slice(raw).map_err(Into::into)
+		versioned::CachedValue::deserialize_with_embedded_version(raw)
 	}
 
 	fn serialize(&self, value: Self::Value) -> Result<Vec<u8>> {
-		serde_bare::to_vec(&value).map_err(Into::into)
+		versioned::CachedValue::wrap_latest(value).serialize_with_embedded_version(PROTOCOL_VERSION)
 	}
 }
 

engine/packages/epoxy/src/keys/keys.rs

@@ -5,8 +5,8 @@ pub mod keys;
 pub mod replica;
 
 pub use self::keys::{
-	ChangelogKey, CommittedValue, KvAcceptedKey, KvAcceptedValue, KvBallotKey,
-	KvOptimisticCacheKey, KvValueKey, LegacyCommittedValueKey,
+	ChangelogKey, KvAcceptedKey, KvAcceptedValue, KvBallotKey, KvOptimisticCacheKey, KvValueKey,
+	LegacyCommittedValueKey,
 };
 pub use self::replica::ConfigKey;
 

engine/packages/epoxy/src/keys/mod.rs

@@ -1,11 +1,9 @@
 use anyhow::Result;
-use epoxy_protocol::protocol::ReplicaId;
+use epoxy_protocol::protocol::{CachedValue, CommittedValue, ReplicaId};
 use gas::prelude::*;
 use universaldb::utils::{FormalKey, IsolationLevel::Serializable};
 
-use crate::keys::{
-	self, CommittedValue, KvOptimisticCacheKey, KvValueKey, LegacyCommittedValueKey,
-};
+use crate::keys::{self, KvOptimisticCacheKey, KvValueKey, LegacyCommittedValueKey};
 
 #[derive(Debug)]
 pub struct Input {
@@ -26,7 +24,7 @@ pub async fn epoxy_kv_get_local(
 #[derive(Debug)]
 pub(crate) struct LocalValueRead {
 	pub value: Option<CommittedValue>,
-	pub cache_value: Option<CommittedValue>,
+	pub cache_value: Option<CachedValue>,
 }
 
 /// Reads a committed value from the local replica with dual-read fallback.
@@ -95,18 +93,12 @@ pub(crate) async fn read_local_value(
 				if let Some(value) = cache_value {
 					let cache_value = cache_key.deserialize(&value)?;
 
-					// Special case with empty values. These are inserted in kv_get_optimistic with `save_empty`
-					if cache_value.value.is_empty() {
+					if let Some(value) = cache_value.value {
 						return Ok(LocalValueRead {
 							value: None,
-							cache_value: None,
+							cache_value: Some(cache_key.deserialize(&value)?),
 						});
 					}
-
-					return Ok(LocalValueRead {
-						value: None,
-						cache_value: Some(cache_key.deserialize(&value)?),
-					});
 				}
 
 				Ok(LocalValueRead {

engine/packages/epoxy/src/ops/kv/get_local.rs

@@ -1,14 +1,10 @@
 use anyhow::*;
-use epoxy_protocol::protocol::{self, ReplicaId};
+use epoxy_protocol::protocol::{self, CachedValue, CommittedValue, ReplicaId};
 use gas::prelude::*;
 use rivet_api_builder::ApiCtx;
 use universaldb::prelude::*;
 
-use crate::{
-	http_client,
-	keys::{self, CommittedValue},
-	utils,
-};
+use crate::{http_client, keys, utils};
 
 use super::get_local::read_local_value;
 
@@ -68,10 +64,12 @@ pub async fn epoxy_kv_get_optimistic(ctx: &OperationCtx, input: &Input) -> Resul
 		input.caching_behavior == protocol::CachingBehavior::Optimistic,
 	)
 	.await?;
-	if let Some(value) = local_read.value.or(local_read.cache_value) {
-		return Ok(Output {
-			value: Some(value.value),
-		});
+	if let Some(value) = local_read
+		.value
+		.map(|v| v.value)
+		.or(local_read.cache_value.and_then(|v| v.value))
+	{
+		return Ok(Output { value: Some(value) });
 	}
 
 	// Request fanout to other datacenters, return first datacenter with any non-none value
@@ -169,7 +167,13 @@ async fn cache_fanout_value(
 					}
 				}
 
-				tx.write(&cache_key, value_to_cache.clone())?;
+				tx.write(
+					&cache_key,
+					CachedValue {
+						value: Some(value_to_cache.value.clone()),
+						version: value_to_cache.version,
+					},
+				)?;
 				Ok(())
 			}
 		})
@@ -189,11 +193,9 @@ async fn cache_empty_value(ctx: &OperationCtx, replica_id: ReplicaId, key: &[u8]
 
 			tx.write(
 				&cache_key,
-				CommittedValue {
-					value: Vec::new(),
+				CachedValue {
+					value: None,
 					version: 0,
-					// TODO: What should this be set to?
-					mutable: true,
 				},
 			)?;
 			Ok(())

engine/packages/epoxy/src/ops/kv/get_optimistic.rs

@@ -1,5 +1,5 @@
 use anyhow::{Context, Result, bail};
-use epoxy_protocol::protocol::{self, ReplicaId};
+use epoxy_protocol::protocol::{self, CommittedValue, ReplicaId};
 use futures_util::{StreamExt, stream::FuturesUnordered};
 use gas::prelude::*;
 use rand::Rng;
@@ -8,9 +8,7 @@ use serde::{Deserialize, Serialize};
 use std::time::{Duration, Instant};
 
 use crate::{
-	http_client,
-	keys::CommittedValue,
-	metrics,
+	http_client, metrics,
 	replica::{
 		ballot::{self, Ballot, BallotSelection},
 		commit_kv::{self, CommitKvOutcome},

engine/packages/epoxy/src/ops/propose.rs

@@ -1,10 +1,10 @@
 use anyhow::{Context, Result};
-use epoxy_protocol::protocol;
+use epoxy_protocol::protocol::{self, CommittedValue};
 use std::cmp::Ordering;
 use universaldb::Transaction;
 use universaldb::utils::{FormalKey, IsolationLevel::Serializable};
 
-use crate::keys::{self, CommittedValue, KvBallotKey, KvValueKey, LegacyCommittedValueKey};
+use crate::keys::{self, KvBallotKey, KvValueKey, LegacyCommittedValueKey};
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
 pub struct Ballot {

engine/packages/epoxy/src/replica/ballot.rs

@@ -1,5 +1,5 @@
 use anyhow::{Context, Result, bail};
-use epoxy_protocol::protocol;
+use epoxy_protocol::protocol::{self, CommittedValue};
 use futures_util::TryStreamExt;
 use universaldb::{
 	KeySelector, RangeOption, Transaction,
@@ -10,8 +10,7 @@ use universaldb::{
 };
 
 use crate::keys::{
-	self, ChangelogKey, CommittedValue, KvAcceptedKey, KvBallotKey, KvOptimisticCacheKey,
-	KvValueKey,
+	self, ChangelogKey, KvAcceptedKey, KvBallotKey, KvOptimisticCacheKey, KvValueKey,
 };
 use crate::metrics;
 

engine/packages/epoxy/src/replica/changelog.rs

@@ -1,9 +1,9 @@
 use anyhow::Result;
-use epoxy_protocol::protocol;
+use epoxy_protocol::protocol::{self, CommittedValue};
 use universaldb::{Transaction, utils::IsolationLevel::Serializable};
 
 use crate::{
-	keys::{self, CommittedValue, KvAcceptedKey, KvBallotKey, KvOptimisticCacheKey, KvValueKey},
+	keys::{self, KvAcceptedKey, KvBallotKey, KvOptimisticCacheKey, KvValueKey},
 	replica::ballot::Ballot,
 };