feat: add actor kv to runners

Author: MasterPtato

Cargo.toml

@@ -76,14 +76,6 @@ default-features = false
 git = "https://github.com/rivet-gg/serde_array_query"
 rev = "b9f8bfa"
 
-[workspace.dependencies.deno_core]
-git = "https://github.com/rivet-gg/deno_core"
-rev = "8a313913fa73d58f4f9532565b0084e723bc34ad"
-
-[workspace.dependencies.deno_runtime]
-git = "https://github.com/rivet-gg/deno"
-rev = "a6903d67063e07b82836399f63c7a0fa5be8bf56"
-
 [workspace.dependencies.api-helper]
 path = "packages/common/api-helper/build"
 

examples/system-test-actor/src/managerClient.ts

@@ -1,6 +1,7 @@
 import * as net from "net";
 import * as fs from "fs";
 import { setInterval, clearInterval } from "timers";
+import * as util from "util";
 
 export function connectToManager() {
 	const socketPath = process.env.RIVET_MANAGER_SOCKET_PATH;
@@ -31,8 +32,9 @@ export function connectToManager() {
 
 	client.on("data", (data) => {
 		const packets = decodeFrames(data);
-		packets.forEach((packet) => {
-			console.log("Received packet from manager:", packet);
+
+		for (let packet of packets) {
+			console.log("Received packet from manager:", util.inspect(packet, { depth: null }));
 
 			if (packet.start_actor) {
 				const response = {
@@ -45,6 +47,41 @@ export function connectToManager() {
 					},
 				};
 				client.write(encodeFrame(response));
+
+				const kvMessage = {
+					kv: {
+						actor_id: packet.start_actor.actor_id,
+						generation: packet.start_actor.generation,
+						request_id: 1,
+						data: {
+							put: {
+								keys: [
+									[[1, 2, 3], [4, 5, 6]],
+								],
+								values: [
+									[11, 12, 13, 14, 15, 16]
+								],
+							}
+						}
+					}
+				};
+				client.write(encodeFrame(kvMessage));
+
+				const kvMessage2 = {
+					kv: {
+						actor_id: packet.start_actor.actor_id,
+						generation: packet.start_actor.generation,
+						request_id: 2,
+						data: {
+							get: {
+								keys: [
+									[[1, 2, 3], [4, 5, 6]]
+								],
+							}
+						}
+					}
+				};
+				client.write(encodeFrame(kvMessage2));
 			} else if (packet.signal_actor) {
 				const response = {
 					actor_state_update: {
@@ -59,7 +96,7 @@ export function connectToManager() {
 				};
 				client.write(encodeFrame(response));
 			}
-		});
+		}
 	});
 
 	client.on("error", (error) => {
@@ -98,7 +135,7 @@ function decodeFrames(buffer: Buffer): any[] {
 		offset += 4;
 
 		if (buffer.length - offset < payloadLength) break; // Incomplete frame data
-		const json = buffer.slice(offset, offset + payloadLength).toString();
+		const json = buffer.subarray(offset, offset + payloadLength).toString();
 		packets.push(JSON.parse(json));
 		offset += payloadLength;
 	}

packages/common/fdb-util/src/codes.rs

@@ -1,3 +1,8 @@
+// === Copied from foundationdbrs ===
+pub const NIL: u8 = 0x00;
+pub const NESTED: u8 = 0x05;
+pub const ESCAPE: u8 = 0xff;
+
 // FDB defines a range (0x40-0x4f) of user type codes for use with its tuple encoding system.
 // https://github.com/apple/foundationdb/blob/main/design/tuple.md#user-type-codes
 

packages/edge/infra/client/actor-kv/Cargo.toml

@@ -7,12 +7,12 @@ license = "Apache-2.0"
 
 [dependencies]
 anyhow.workspace = true
-deno_core.workspace = true
 fdb-util.workspace = true
 foundationdb.workspace = true
 futures-util = { version = "0.3" }
 indexmap = { version = "2.0" }
 prost = "0.13.3"
+rivet-util-id.workspace = true
 serde = { version = "1.0.195", features = ["derive"] }
 serde_json = "1.0.111"
 tokio-tungstenite = "0.23.1"

packages/edge/infra/client/actor-kv/src/entry.rs

@@ -1,7 +1,7 @@
 use anyhow::*;
 use foundationdb as fdb;
 use prost::Message;
-use serde::Serialize;
+use serde::{Deserialize, Serialize};
 
 use crate::{key::Key, metadata::Metadata};
 
@@ -49,7 +49,7 @@ impl EntryBuilder {
 }
 
 /// Represents a Rivet KV value.
-#[derive(Serialize)]
+#[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct Entry {
 	pub metadata: Metadata,
 	pub value: Vec<u8>,

packages/edge/infra/client/actor-kv/src/key.rs

@@ -1,18 +1,11 @@
-use deno_core::JsBuffer;
 use foundationdb::tuple::{
-	Bytes, PackError, PackResult, TupleDepth, TuplePack, TupleUnpack, VersionstampOffset,
+	Bytes, PackResult, TupleDepth, TuplePack, TupleUnpack, VersionstampOffset,
 };
-use serde::Deserialize;
+use serde::{Serialize, Deserialize};
 
 // TODO: Custom deser impl that uses arrays instead of objects?
-#[derive(Clone, Deserialize)]
-#[serde(rename_all = "camelCase")]
-pub enum Key {
-	/// Contains references to v8-owned buffers. Requires no copies.
-	JsInKey(Vec<JsBuffer>),
-	/// Cant use `ToJsBuffer` because of its API, so it gets converted to ToJsBuffer in the KV ext.
-	JsOutKey(Vec<Vec<u8>>),
-}
+#[derive(Clone, Serialize, Deserialize)]
+pub struct Key(Vec<Vec<u8>>);
 
 impl std::fmt::Debug for Key {
 	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
@@ -22,49 +15,24 @@ impl std::fmt::Debug for Key {
 
 impl PartialEq for Key {
 	fn eq(&self, other: &Self) -> bool {
-		match (self, other) {
-			(Key::JsInKey(a), Key::JsInKey(b)) => a
-				.iter()
-				.map(|x| x.as_ref())
-				.eq(b.iter().map(|x| x.as_ref())),
-			(Key::JsOutKey(a), Key::JsOutKey(b)) => a == b,
-			(Key::JsInKey(a), Key::JsOutKey(b)) => a.iter().map(|x| x.as_ref()).eq(b.iter()),
-			(Key::JsOutKey(a), Key::JsInKey(b)) => a.iter().eq(b.iter().map(|x| x.as_ref())),
-		}
+		self.0 == other.0
 	}
 }
 
 impl Eq for Key {}
 
 impl std::hash::Hash for Key {
 	fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
-		match self {
-			Key::JsInKey(js_in_key) => {
-				for buffer in js_in_key {
-					state.write(buffer.as_ref());
-				}
-			}
-			Key::JsOutKey(out_key) => {
-				for buffer in out_key {
-					state.write(buffer);
-				}
-			}
+		for buffer in &self.0 {
+			state.write(buffer);
 		}
 	}
 }
 
 impl Key {
 	pub fn len(&self) -> usize {
-		match self {
-			Key::JsInKey(js_in_key) => {
-				// Arbitrary 4 accounting for nesting overhead
-				js_in_key.iter().fold(0, |acc, x| acc + x.len()) + 4 * js_in_key.len()
-			}
-			Key::JsOutKey(out_key) => {
-				// Arbitrary 4 accounting for nesting overhead
-				out_key.iter().fold(0, |acc, x| acc + x.len()) + 4 * out_key.len()
-			}
-		}
+		// Arbitrary 4 accounting for nesting overhead
+		self.0.iter().fold(0, |acc, x| acc + x.len()) + 4 * self.0.len()
 	}
 }
 
@@ -74,30 +42,25 @@ impl TuplePack for Key {
 		w: &mut W,
 		tuple_depth: TupleDepth,
 	) -> std::io::Result<VersionstampOffset> {
-		match self {
-			Key::JsInKey(tuple) => {
-				let mut offset = VersionstampOffset::None { size: 0 };
+		let mut offset = VersionstampOffset::None { size: 0 };
 
-				w.write_all(&[NESTED])?;
-				offset += 1;
+		w.write_all(&[fdb_util::codes::NESTED])?;
+		offset += 1;
 
-				for v in tuple.iter() {
-					offset += v.as_ref().pack(w, tuple_depth.increment())?;
-				}
+		for v in self.0.iter() {
+			offset += v.pack(w, tuple_depth.increment())?;
+		}
 
-				w.write_all(&[NIL])?;
-				offset += 1;
+		w.write_all(&[fdb_util::codes::NIL])?;
+		offset += 1;
 
-				Ok(offset)
-			}
-			Key::JsOutKey(_) => unreachable!("should not be packing out keys"),
-		}
+		Ok(offset)
 	}
 }
 
 impl<'de> TupleUnpack<'de> for Key {
 	fn unpack(mut input: &[u8], tuple_depth: TupleDepth) -> PackResult<(&[u8], Self)> {
-		input = parse_code(input, NESTED)?;
+		input = fdb_util::parse_code(input, fdb_util::codes::NESTED)?;
 
 		let mut vec = Vec::new();
 		while !is_end_of_tuple(input, true) {
@@ -106,15 +69,21 @@ impl<'de> TupleUnpack<'de> for Key {
 			vec.push(v.into_owned());
 		}
 
-		input = parse_code(input, NIL)?;
+		input = fdb_util::parse_code(input, fdb_util::codes::NIL)?;
 
-		Ok((input, Key::JsOutKey(vec)))
+		Ok((input, Key(vec)))
 	}
 }
 
-/// Same as Key::JsInKey except when packing, it leaves off the NIL byte to allow for an open range.
-#[derive(Deserialize)]
-pub struct ListKey(Vec<JsBuffer>);
+/// Same as Key: except when packing, it leaves off the NIL byte to allow for an open range.
+#[derive(Clone, Serialize, Deserialize)]
+pub struct ListKey(Vec<Vec<u8>>);
+
+impl std::fmt::Debug for ListKey {
+	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+		write!(f, "ListKey({})", self.len())
+	}
+}
 
 impl TuplePack for ListKey {
 	fn pack<W: std::io::Write>(
@@ -124,11 +93,11 @@ impl TuplePack for ListKey {
 	) -> std::io::Result<VersionstampOffset> {
 		let mut offset = VersionstampOffset::None { size: 0 };
 
-		w.write_all(&[NESTED])?;
+		w.write_all(&[fdb_util::codes::NESTED])?;
 		offset += 1;
 
-		for v in self.0.iter() {
-			offset += v.as_ref().pack(w, tuple_depth.increment())?;
+		for v in &self.0 {
+			offset += v.pack(w, tuple_depth.increment())?;
 		}
 
 		// No ending NIL byte compared to `Key::pack`
@@ -144,37 +113,11 @@ impl ListKey {
 	}
 }
 
-// === Copied from foundationdbrs ===
-const NIL: u8 = 0x00;
-const NESTED: u8 = 0x05;
-const ESCAPE: u8 = 0xff;
-
-#[inline]
-fn parse_byte(input: &[u8]) -> PackResult<(&[u8], u8)> {
-	if input.is_empty() {
-		Err(PackError::MissingBytes)
-	} else {
-		Ok((&input[1..], input[0]))
-	}
-}
-
-fn parse_code(input: &[u8], expected: u8) -> PackResult<&[u8]> {
-	let (input, found) = parse_byte(input)?;
-	if found == expected {
-		Ok(input)
-	} else {
-		Err(PackError::BadCode {
-			found,
-			expected: Some(expected),
-		})
-	}
-}
-
 fn is_end_of_tuple(input: &[u8], nested: bool) -> bool {
 	match input.first() {
 		None => true,
 		_ if !nested => false,
-		Some(&NIL) => Some(&ESCAPE) != input.get(1),
+		Some(&fdb_util::codes::NIL) => Some(&fdb_util::codes::ESCAPE) != input.get(1),
 		_ => false,
 	}
 }