fix(actors): surface unregistered actor startup errors

engine/packages/guard/Cargo.toml

@@ -43,6 +43,7 @@ rivet-api-public.workspace = true
 rivet-cache.workspace = true
 rivet-config.workspace = true
 rivet-data.workspace = true
+rivet-envoy-protocol.workspace = true
 rivet-error.workspace = true
 rivet-types.workspace = true
 rivet-guard-core.workspace = true

engine/packages/guard/src/routing/pegboard_gateway/mod.rs

@@ -514,6 +514,11 @@
 				res = stopped_sub.next() => {
 					res?;
 
+					if let Some(actor_name) = actor_not_registered_name(ctx, actor_id).await? {
+						record_ready_wait("failed", wake_retries);
+						return Err(pegboard::errors::Actor::NotRegistered { actor_name }.build());
+					}
+
 					if wake_retries < 8 {
 						tracing::debug!(
 							?actor_id,
@@ -632,6 +637,44 @@
 	Ok(RoutingOutput::CustomServe(std::sync::Arc::new(gateway)))
 }
 
+async fn actor_not_registered_name(ctx: &StandaloneCtx, actor_id: Id) -> Result<Option<String>> {
+	let actor = ctx
+		.op(pegboard::ops::actor::get::Input {
+			actor_ids: vec![actor_id],
+			fetch_error: true,
+		})
+		.await?
+		.actors
+		.into_iter()
+		.next();
+
+	Ok(actor
+		.as_ref()
+		.and_then(|actor| not_registered_name_from_actor_error(actor.error.as_ref())))
+}
+
+fn not_registered_name_from_actor_error(
+	error: Option<&rivet_types::actor::ActorError>,
+) -> Option<String> {
+	let Some(rivet_types::actor::ActorError::Crashed {
+		message: Some(message),
+	}) = error
+	else {
+		return None;
+	};
+	let error = rivet_envoy_protocol::util::decode_actor_error(message)?;
+	if error.group != "actor" || error.code != "not_registered" {
+		return None;
+	}
+
+	error
+		.metadata
+		.as_ref()?
+		.get("actor_name")?
+		.as_str()
+		.map(ToOwned::to_owned)
+}
+
 async fn handle_actor_v1(
 	ctx: &StandaloneCtx,
 	shared_state: &SharedState,
@@ -922,3 +965,39 @@
 		tokio::time::sleep(RUNNER_POOL_ERROR_CHECK_INTERVAL).await;
 	}
 }
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	#[test]
+	fn extracts_not_registered_actor_name_from_envoy_start_error() {
+		let message =
+			rivet_envoy_protocol::util::encode_actor_error(
+				&rivet_envoy_protocol::util::ActorErrorEnvelope {
+					group: "actor".to_owned(),
+					code: "not_registered".to_owned(),
+					message: "Actor factory 'removed' is not registered.".to_owned(),
+					metadata: Some(serde_json::json!({ "actor_name": "removed" })),
+				},
+			)
+			.expect("encode actor error");
+		let error = rivet_types::actor::ActorError::Crashed {
+			message: Some(message),
+		};
+
+		assert_eq!(
+			not_registered_name_from_actor_error(Some(&error)).as_deref(),
+			Some("removed")
+		);
+	}
+
+	#[test]
+	fn ignores_unstructured_actor_crashes() {
+		let error = rivet_types::actor::ActorError::Crashed {
+			message: Some("boom".to_owned()),
+		};
+
+		assert_eq!(not_registered_name_from_actor_error(Some(&error)), None);
+	}
+}

engine/packages/pegboard/src/errors.rs

@@ -13,6 +13,13 @@ pub enum Actor {
 	#[error("namespace_not_found", "The namespace does not exist.")]
 	NamespaceNotFound,
 
+	#[error(
+		"not_registered",
+		"Actor factory is not registered.",
+		"Actor factory '{actor_name}' is not registered."
+	)]
+	NotRegistered { actor_name: String },
+
 	#[error(
 		"input_too_large",
 		"Actor input too large.",

engine/sdks/rust/envoy-client/Cargo.toml

@@ -21,6 +21,7 @@ hex.workspace = true
 js-sys = { version = "0.3", optional = true }
 rand.workspace = true
 rivet-envoy-protocol.workspace = true
+rivet-error.workspace = true
 rivet-metrics.workspace = true
 rivet-util-serde.workspace = true
 rustls = { workspace = true, optional = true }

engine/sdks/rust/envoy-client/src/actor.rs

@@ -4,6 +4,7 @@ use std::sync::Arc;
 
 use crate::async_counter::AsyncCounter;
 use rivet_envoy_protocol as protocol;
+use rivet_error::RivetError;
 use tokio::sync::mpsc;
 use tokio::sync::oneshot;
 use tokio::sync::oneshot::error::TryRecvError;
@@ -204,7 +205,7 @@ async fn actor_inner(
 			protocol::Event::EventActorStateUpdate(protocol::EventActorStateUpdate {
 				state: protocol::ActorState::ActorStateStopped(protocol::ActorStateStopped {
 					code: protocol::StopCode::Error,
-					message: Some(format!("{error:#}")),
+					message: Some(actor_start_error_message(&error)),
 				}),
 			}),
 		);
@@ -376,6 +377,22 @@ async fn actor_inner(
 	tracing::debug!("envoy actor stopped");
 }
 
+fn actor_start_error_message(error: &anyhow::Error) -> String {
+	let Some(error) = error
+		.chain()
+		.find_map(|cause| cause.downcast_ref::<RivetError>())
+	else {
+		return format!("{error:#}");
+	};
+	let envelope = protocol::util::ActorErrorEnvelope {
+		group: error.group().to_owned(),
+		code: error.code().to_owned(),
+		message: error.message().to_owned(),
+		metadata: error.metadata(),
+	};
+	protocol::util::encode_actor_error(&envelope).unwrap_or_else(|_| error.to_string())
+}
+
 fn send_event(ctx: &mut ActorContext, inner: protocol::Event) {
 	let checkpoint = increment_checkpoint(ctx);
 	let _ = crate::envoy::send_to_envoy_tx(

engine/sdks/rust/envoy-protocol/Cargo.toml

@@ -15,6 +15,7 @@ rand.workspace = true
 rivet-util-serde.workspace = true
 serde_bare.workspace = true
 serde.workspace = true
+serde_json.workspace = true
 utoipa.workspace = true
 vbare.workspace = true
 

engine/sdks/rust/envoy-protocol/src/util.rs

@@ -1,3 +1,28 @@
+use serde::{Deserialize, Serialize};
+
+const ACTOR_ERROR_PREFIX: &str = "rivet-error:";
+
+#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
+pub struct ActorErrorEnvelope {
+	pub group: String,
+	pub code: String,
+	pub message: String,
+	#[serde(default, skip_serializing_if = "Option::is_none")]
+	pub metadata: Option<serde_json::Value>,
+}
+
+pub fn encode_actor_error(error: &ActorErrorEnvelope) -> Result<String, serde_json::Error> {
+	Ok(format!(
+		"{ACTOR_ERROR_PREFIX}{}",
+		serde_json::to_string(error)?
+	))
+}
+
+pub fn decode_actor_error(message: &str) -> Option<ActorErrorEnvelope> {
+	let payload = message.strip_prefix(ACTOR_ERROR_PREFIX)?;
+	serde_json::from_str(payload).ok()
+}
+
 /// Generate a new 4-byte gateway ID from a random u32
 pub fn generate_gateway_id() -> crate::GatewayId {
 	rand::random::<u32>().to_le_bytes()
@@ -12,3 +37,23 @@ pub fn generate_request_id() -> crate::RequestId {
 pub fn id_to_string(gateway_id: &crate::GatewayId) -> String {
 	hex::encode(gateway_id)
 }
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	#[test]
+	fn actor_error_envelope_round_trips() {
+		let error = ActorErrorEnvelope {
+			group: "actor".to_owned(),
+			code: "not_registered".to_owned(),
+			message: "Actor factory 'removed' is not registered.".to_owned(),
+			metadata: Some(serde_json::json!({ "actor_name": "removed" })),
+		};
+
+		let encoded = encode_actor_error(&error).expect("encode actor error");
+
+		assert_eq!(decode_actor_error(&encoded), Some(error));
+		assert_eq!(decode_actor_error("ordinary crash message"), None);
+	}
+}