feat: implement new actor ids

Author: MasterPtato

Cargo.lock

@@ -20,7 +20,7 @@ export function connectToManager() {
 
 		let message = {
 			init: {
-				runner_id: process.env.RIVET_RUNNER_ID
+				access_token: process.env.RIVET_ACCESS_TOKEN
 			}
 		};
 		let buffer = Buffer.from(JSON.stringify(message));

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

@@ -3,7 +3,7 @@ extern crate proc_macro;
 use std::iter::FromIterator;
 
 use proc_macro::TokenStream;
-use proc_macro2::{Literal, TokenStream as TokenStream2, TokenTree};
+use proc_macro2::{Literal, Spacing, TokenStream as TokenStream2, TokenTree};
 use proc_macro_error::{emit_warning, proc_macro_error};
 use quote::{format_ident, quote, ToTokens};
 use syn::{
@@ -454,6 +454,22 @@ impl Parse for Endpoint {
 			while let Some((tt, next)) = rest.token_tree() {
 				match &tt {
 					TokenTree::Punct(punct) if punct.as_char() == ':' => {
+						// Check for path separator (::)
+						if punct.spacing() == Spacing::Joint {
+							if let Some((tt2, next)) = next.token_tree() {
+								match &tt2 {
+									TokenTree::Punct(punct) if punct.as_char() == ':' => {
+										tts.push(tt);
+										tts.push(tt2);
+										rest = next;
+
+										continue;
+									}
+									_ => {}
+								}
+							}
+						}
+
 						return Ok((tts.into_iter().collect::<TokenStream2>(), next));
 					}
 					_ => {

packages/common/api-helper/macros/src/lib.rs

@@ -535,6 +535,7 @@ impl DatabaseDebug for DatabaseCrdbNats {
 				tags,
 				NULL AS workflow_id,
 				create_ts,
+				silence_ts,
 				body,
 				ack_ts
 			FROM db_workflow.tagged_signals

packages/common/chirp-workflow/core/src/db/crdb_nats/debug.rs

@@ -926,7 +926,6 @@ impl TuplePack for ByNameAndTagSubspaceKey {
 			&self.k,
 			&self.v,
 		);
-		tracing::info!(?t, "---------------------");
 		t.pack(w, tuple_depth)
 	}
 }

packages/common/chirp-workflow/core/src/db/fdb_sqlite_nats/keys/workflow.rs

@@ -2795,7 +2795,6 @@ impl Database for DatabaseFdbSqliteNats {
 				// TODO: Add config parameter in either fdb or sqlite to toggle this per wf
 				let delete_instead_of_forget =
 					workflow_name == "pegboard_client" || workflow_name == "pegboard_actor";
-				// let delete_instead_of_forget = false;
 
 				if delete_instead_of_forget {
 					sql_execute!(

packages/common/chirp-workflow/core/src/db/fdb_sqlite_nats/mod.rs

@@ -848,3 +848,12 @@ pub struct Edge {
 	#[serde(default)]
 	pub redirect_logs_dir: Option<PathBuf>,
 }
+
+impl Edge {
+	/// u16 used in the `Id` type for actors
+	pub fn datacenter_label(&self) -> u16 {
+		// Read first 2 bytes
+		let bytes = self.datacenter_id.as_bytes();
+		u16::from_be_bytes([bytes[0], bytes[1]])
+	}
+}

packages/common/config/src/config/server/rivet/mod.rs

@@ -0,0 +1,4 @@
+// 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
+
+pub const ID: u8 = 0x40;

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

@@ -51,6 +51,7 @@ pub const RUNNERS_BY_REMAINING_SLOTS: usize = 49;
 pub const REMAINING_SLOTS: usize = 50;
 pub const TOTAL_SLOTS: usize = 51;
 pub const IMAGE_ID: usize = 52;
+pub const ACTOR2: usize = 53;
 
 // Directories with fdbrs must use string paths instead of tuples
 pub mod dir {
@@ -114,6 +115,7 @@ pub fn key_from_str(key: &str) -> Option<usize> {
 		"remaining_slots" => Some(REMAINING_SLOTS),
 		"total_slots" => Some(TOTAL_SLOTS),
 		"image_id" => Some(IMAGE_ID),
+		"actor2" => Some(ACTOR2),
 		_ => None,
 	}
 }

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

@@ -11,11 +11,12 @@ use foundationdb::{
 	self as fdb,
 	future::FdbValue,
 	options::DatabaseOption,
-	tuple::{self, PackResult, TuplePack, TupleUnpack},
+	tuple::{self, PackResult, PackError, TuplePack, TupleUnpack},
 	KeySelector, RangeOption,
 };
 
 pub mod keys;
+pub mod codes;
 mod metrics;
 
 /// Makes the code blatantly obvious if its using a snapshot read.
@@ -189,6 +190,39 @@ pub fn end_of_key_range(key: &[u8]) -> Vec<u8> {
 	end_key
 }
 
+// Copied from foundationdb crate
+#[inline]
+pub fn parse_bytes(input: &[u8], num: usize) -> PackResult<(&[u8], &[u8])> {
+    if input.len() < num {
+        Err(PackError::MissingBytes)
+    } else {
+        Ok((&input[num..], &input[..num]))
+    }
+}
+
+// Copied from foundationdb crate
+#[inline]
+pub fn parse_byte(input: &[u8]) -> PackResult<(&[u8], u8)> {
+    if input.is_empty() {
+        Err(PackError::MissingBytes)
+    } else {
+        Ok((&input[1..], input[0]))
+    }
+}
+
+// Copied from foundationdb crate
+pub 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),
+        })
+    }
+}
+
 pub mod prelude {
 	pub use std::result::Result::Ok;