From 747c01e54a95e3515e07f7406f2ca1aa1879df29 Mon Sep 17 00:00:00 2001 From: Nathan Flurry Date: Thu, 30 Jul 2026 13:14:40 -0700 Subject: [PATCH] fix(actors): surface unregistered actor startup errors --- Cargo.lock | 3 + engine/packages/guard/Cargo.toml | 1 + .../guard/src/routing/pegboard_gateway/mod.rs | 79 +++++++++++++++++++ engine/packages/pegboard/src/errors.rs | 7 ++ engine/sdks/rust/envoy-client/Cargo.toml | 1 + engine/sdks/rust/envoy-client/src/actor.rs | 19 ++++- engine/sdks/rust/envoy-protocol/Cargo.toml | 1 + engine/sdks/rust/envoy-protocol/src/util.rs | 45 +++++++++++ 8 files changed, 155 insertions(+), 1 deletion(-) diff --git a/Cargo.lock b/Cargo.lock index 12ad23aba3..950f29136f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5489,6 +5489,7 @@ dependencies = [ "parking_lot", "rand 0.8.5", "rivet-envoy-protocol", + "rivet-error", "rivet-metrics", "rivet-util-serde", "rustls", @@ -5520,6 +5521,7 @@ dependencies = [ "rivet-vbare-compiler", "serde", "serde_bare", + "serde_json", "utoipa", "uuid", "vbare", @@ -5582,6 +5584,7 @@ dependencies = [ "rivet-cache", "rivet-config", "rivet-data", + "rivet-envoy-protocol", "rivet-error", "rivet-guard-core", "rivet-logs", diff --git a/engine/packages/guard/Cargo.toml b/engine/packages/guard/Cargo.toml index f0f212d988..bd18b95763 100644 --- a/engine/packages/guard/Cargo.toml +++ b/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 diff --git a/engine/packages/guard/src/routing/pegboard_gateway/mod.rs b/engine/packages/guard/src/routing/pegboard_gateway/mod.rs index e16ec4fdfb..b72b0c205f 100644 --- a/engine/packages/guard/src/routing/pegboard_gateway/mod.rs +++ b/engine/packages/guard/src/routing/pegboard_gateway/mod.rs @@ -514,6 +514,11 @@ async fn handle_actor_v2( 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 @@ async fn handle_actor_v2( Ok(RoutingOutput::CustomServe(std::sync::Arc::new(gateway))) } +async fn actor_not_registered_name(ctx: &StandaloneCtx, actor_id: Id) -> Result> { + 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 { + 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 @@ async fn check_runner_pool_error_loop( 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); + } +} diff --git a/engine/packages/pegboard/src/errors.rs b/engine/packages/pegboard/src/errors.rs index 13e21b55cb..c0b6753eb3 100644 --- a/engine/packages/pegboard/src/errors.rs +++ b/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.", diff --git a/engine/sdks/rust/envoy-client/Cargo.toml b/engine/sdks/rust/envoy-client/Cargo.toml index 57c28b36a8..14d7212049 100644 --- a/engine/sdks/rust/envoy-client/Cargo.toml +++ b/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 } diff --git a/engine/sdks/rust/envoy-client/src/actor.rs b/engine/sdks/rust/envoy-client/src/actor.rs index 4a5bfc0581..944c86fc6c 100644 --- a/engine/sdks/rust/envoy-client/src/actor.rs +++ b/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::()) + 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( diff --git a/engine/sdks/rust/envoy-protocol/Cargo.toml b/engine/sdks/rust/envoy-protocol/Cargo.toml index 4e3a222a97..82735fb900 100644 --- a/engine/sdks/rust/envoy-protocol/Cargo.toml +++ b/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 diff --git a/engine/sdks/rust/envoy-protocol/src/util.rs b/engine/sdks/rust/envoy-protocol/src/util.rs index 9a4034963e..fc8866da61 100644 --- a/engine/sdks/rust/envoy-protocol/src/util.rs +++ b/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, +} + +pub fn encode_actor_error(error: &ActorErrorEnvelope) -> Result { + Ok(format!( + "{ACTOR_ERROR_PREFIX}{}", + serde_json::to_string(error)? + )) +} + +pub fn decode_actor_error(message: &str) -> Option { + 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::().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); + } +}