fix: fix logs for new actors

Author: MasterPtato

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

@@ -48,6 +48,8 @@ export function connectToManager() {
 				};
 				client.write(encodeFrame(response));
 
+				console.log(`actor_${packet.start_actor.actor_id}`, 'fweh');
+
 				const kvMessage = {
 					kv: {
 						actor_id: packet.start_actor.actor_id,

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

@@ -11,12 +11,12 @@ use foundationdb::{
 	self as fdb,
 	future::FdbValue,
 	options::DatabaseOption,
-	tuple::{self, PackResult, PackError, TuplePack, TupleUnpack},
+	tuple::{self, PackError, PackResult, TuplePack, TupleUnpack},
 	KeySelector, RangeOption,
 };
 
-pub mod keys;
 pub mod codes;
+pub mod keys;
 mod metrics;
 
 /// Makes the code blatantly obvious if its using a snapshot read.
@@ -193,34 +193,34 @@ pub fn end_of_key_range(key: &[u8]) -> Vec<u8> {
 // 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]))
-    }
+	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]))
-    }
+	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),
-        })
-    }
+	let (input, found) = parse_byte(input)?;
+	if found == expected {
+		Ok(input)
+	} else {
+		Err(PackError::BadCode {
+			found,
+			expected: Some(expected),
+		})
+	}
 }
 
 pub mod prelude {

packages/common/util/core/src/lib.rs

@@ -1,7 +1,7 @@
+pub use id::Id;
 use rand::Rng;
-pub use rivet_util_macros as macros;
 pub use rivet_util_id as id;
-pub use id::Id;
+pub use rivet_util_macros as macros;
 use tokio::time::{Duration, Instant};
 
 pub mod billing;

packages/common/util/id/src/lib.rs

@@ -242,7 +242,7 @@ impl TuplePack for Id {
 		let mut size = 1;
 
 		w.write_all(&[fdb_util::codes::ID])?;
-		
+
 		// IMPORTANT: While the normal bytes representation of a v0 ID doesn't include the version, we write
 		// it here so that we can unpack without a terminating NIL.
 		if let Id::V0(_) = self {

packages/core/api/actor/src/route/logs.rs

@@ -167,6 +167,11 @@ pub async fn get_logs(
 		.iter()
 		.map(|x| x.stream_type as i32)
 		.collect::<Vec<_>>();
+	let mut foreigns = logs_res
+		.entries
+		.iter()
+		.map(|x| x.foreign)
+		.collect::<Vec<_>>();
 	let mut actor_indices = logs_res
 		.entries
 		.iter()
@@ -177,6 +182,7 @@ pub async fn get_logs(
 	lines.reverse();
 	timestamps.reverse();
 	streams.reverse();
+	foreigns.reverse();
 	actor_indices.reverse();
 
 	let watch_nts = logs_res.entries.first().map_or(before_nts, |x| x.ts);
@@ -185,6 +191,7 @@ pub async fn get_logs(
 		lines,
 		timestamps,
 		streams,
+		foreigns,
 		actor_indices,
 		watch: WatchResponse::new_as_model(watch_nts),
 	})

packages/core/services/cluster/src/ops/datacenter/get_for_label.rs

@@ -1,7 +1,7 @@
 use chirp_workflow::prelude::*;
 
-use crate::types::Datacenter;
 use crate::ops::datacenter::get::DatacenterRow;
+use crate::types::Datacenter;
 
 #[derive(Debug)]
 pub struct Input {
@@ -14,7 +14,10 @@ pub struct Output {
 }
 
 #[operation]
-pub async fn cluster_datacenter_get_for_label(ctx: &OperationCtx, input: &Input) -> GlobalResult<Output> {
+pub async fn cluster_datacenter_get_for_label(
+	ctx: &OperationCtx,
+	input: &Input,
+) -> GlobalResult<Output> {
 	let datacenters = ctx
 		.cache()
 		.fetch_all_json("cluster.datacenters_get_for_label", input.labels.clone(), {

packages/core/services/cluster/src/ops/datacenter/mod.rs

@@ -1,9 +1,9 @@
 pub mod get;
+pub mod get_for_label;
 pub mod list;
 pub mod location_get;
 pub mod resolve_for_name_id;
 pub mod server_discovery;
 pub mod server_spec_get;
 pub mod tls_get;
 pub mod topology_get;
-pub mod get_for_label;

packages/edge/api/actor/src/route/actors.rs

@@ -451,14 +451,13 @@ pub async fn destroy(
 
 	if let Some(WorkflowError::WorkflowNotFound) = res.as_workflow_error() {
 		// Try old actors
-		ctx
-			.signal(pegboard::workflows::actor::Destroy {
-				override_kill_timeout_ms: query.override_kill_timeout,
-			})
-			.to_workflow::<pegboard::workflows::actor::Workflow>()
-			.tag("actor_id", actor_id)
-			.send()
-			.await?;
+		ctx.signal(pegboard::workflows::actor::Destroy {
+			override_kill_timeout_ms: query.override_kill_timeout,
+		})
+		.to_workflow::<pegboard::workflows::actor::Workflow>()
+		.tag("actor_id", actor_id)
+		.send()
+		.await?;
 
 		old_sub.next().await?;
 	} else {
@@ -514,14 +513,13 @@ pub async fn upgrade(
 
 	if let Some(WorkflowError::WorkflowNotFound) = res.as_workflow_error() {
 		// Try old actors
-		ctx
-			.signal(pegboard::workflows::actor::Upgrade {
-				image_id: build.build_id,
-			})
-			.to_workflow::<pegboard::workflows::actor::Workflow>()
-			.tag("actor_id", actor_id)
-			.send()
-			.await?;
+		ctx.signal(pegboard::workflows::actor::Upgrade {
+			image_id: build.build_id,
+		})
+		.to_workflow::<pegboard::workflows::actor::Workflow>()
+		.tag("actor_id", actor_id)
+		.send()
+		.await?;
 	} else {
 		res?;
 	}
@@ -635,14 +633,13 @@ pub async fn upgrade_all(
 
 					if let Some(WorkflowError::WorkflowNotFound) = res.as_workflow_error() {
 						// Try old actors
-						ctx
-							.signal(pegboard::workflows::actor::Upgrade {
-								image_id: build.build_id,
-							})
-							.to_workflow::<pegboard::workflows::actor::Workflow>()
-							.tag("actor_id", actor.actor_id)
-							.send()
-							.await?;
+						ctx.signal(pegboard::workflows::actor::Upgrade {
+							image_id: build.build_id,
+						})
+						.to_workflow::<pegboard::workflows::actor::Workflow>()
+						.tag("actor_id", actor.actor_id)
+						.send()
+						.await?;
 					} else {
 						res?;
 					}

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

@@ -1,7 +1,7 @@
 use foundationdb::tuple::{
 	Bytes, PackResult, TupleDepth, TuplePack, TupleUnpack, VersionstampOffset,
 };
-use serde::{Serialize, Deserialize};
+use serde::{Deserialize, Serialize};
 
 // TODO: Custom deser impl that uses arrays instead of objects?
 #[derive(Clone, Serialize, Deserialize)]

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

@@ -59,7 +59,7 @@ impl ActorKv {
 		if let Some(subspace) = &*guard {
 			return Ok(subspace.clone());
 		}
-		
+
 		tracing::info!(actor_id=?self.actor_id, "initializing actor KV");
 
 		let root = fdb::directory::DirectoryLayer::default();