fix(pegboard): stabilize actor2 validation and lifecycle

NathanFlurry · NathanFlurry · commit ffd92027c63b · 2026-04-27T02:35:16.000-07:00
diff --git a/engine/packages/pegboard/src/ops/actor/create.rs b/engine/packages/pegboard/src/ops/actor/create.rs
@@ -63,6 +63,7 @@ pub async fn pegboard_actor_create(ctx: &OperationCtx, input: &Input) -> Result<
 			name: input.name.clone(),
 			pool_name: input.runner_name_selector.clone(),
 			key: input.key.clone(),
+			crash_policy: input.crash_policy,
 			namespace_id: input.namespace_id,
 			input: input.input.clone(),
 			from_v1: false,
diff --git a/engine/packages/pegboard/src/ops/actor/util.rs b/engine/packages/pegboard/src/ops/actor/util.rs
@@ -1,6 +1,6 @@
 use gas::db::WorkflowData;
 use gas::prelude::*;
-use rivet_types::actors::{Actor, CrashPolicy};
+use rivet_types::actors::Actor;
 use std::collections::{HashMap, HashSet};
 
 use crate::workflows::actor::FailureReason as WorkflowFailureReason;
@@ -212,7 +212,7 @@ pub async fn build_actors_from_workflows(
 				namespace_id: s.namespace_id,
 				datacenter: dc_name.to_string(),
 				runner_name_selector: s.pool_name,
-				crash_policy: CrashPolicy::Sleep,
+				crash_policy: s.crash_policy,
 
 				create_ts: s.create_ts,
 				start_ts: s.start_ts,
diff --git a/engine/packages/pegboard/src/workflows/actor/mod.rs b/engine/packages/pegboard/src/workflows/actor/mod.rs
@@ -230,6 +230,7 @@ pub async fn pegboard_actor(ctx: &mut WorkflowCtx, input: &Input) -> Result<()>
 					name: input.name.clone(),
 					pool_name: input.runner_name_selector.clone(),
 					key: input.key.clone(),
+					crash_policy: input.crash_policy,
 					namespace_id: input.namespace_id,
 					input: input.input.clone(),
 					from_v1: true,
@@ -865,6 +866,7 @@ pub async fn pegboard_actor(ctx: &mut WorkflowCtx, input: &Input) -> Result<()>
 			name: input.name.clone(),
 			pool_name: input.runner_name_selector.clone(),
 			key: input.key.clone(),
+			crash_policy: input.crash_policy,
 			namespace_id: input.namespace_id,
 			input: input.input.clone(),
 			from_v1: true,
diff --git a/engine/packages/pegboard/src/workflows/actor2/mod.rs b/engine/packages/pegboard/src/workflows/actor2/mod.rs
@@ -1,7 +1,9 @@
 use futures_util::FutureExt;
 use gas::prelude::*;
+use rivet_data::converted::ActorNameKeyData;
 use rivet_data::converted::ActorByKeyKeyData;
 use rivet_envoy_protocol as protocol;
+use rivet_types::actors::CrashPolicy;
 use universaldb::prelude::*;
 
 use crate::errors;
@@ -14,6 +16,9 @@ use runtime::{StoppedResult, Transition};
 
 /// Batch size of how many events to ack.
 const EVENT_ACK_BATCH_SIZE: i64 = 250;
+pub const SQLITE_SCHEMA_VERSION_V1: u32 = 1;
+pub const SQLITE_SCHEMA_VERSION_V2: u32 = 2;
+const MAX_INPUT_SIZE: usize = util::size::mebibytes(4) as usize;
 
 // NOTE: Assumes input is validated.
 #[derive(Clone, Debug, Serialize, Deserialize, Hash)]
@@ -22,6 +27,7 @@ pub struct Input {
 	pub name: String,
 	pub pool_name: String,
 	pub key: Option<String>,
+	pub crash_policy: CrashPolicy,
 
 	pub namespace_id: Id,
 
@@ -36,6 +42,8 @@ pub struct State {
 	pub name: String,
 	pub pool_name: String,
 	pub key: Option<String>,
+	#[serde(default)]
+	pub crash_policy: CrashPolicy,
 	pub namespace_id: Id,
 
 	pub acquired_slot: bool,
@@ -68,6 +76,7 @@ impl State {
 		name: String,
 		pool_name: String,
 		key: Option<String>,
+		crash_policy: CrashPolicy,
 		namespace_id: Id,
 		create_ts: i64,
 	) -> Self {
@@ -76,6 +85,7 @@ impl State {
 			name,
 			pool_name,
 			key,
+			crash_policy,
 			namespace_id,
 
 			acquired_slot: false,
@@ -111,11 +121,30 @@ pub async fn pegboard_actor2(ctx: &mut WorkflowCtx, input: &Input) -> Result<()>
 	//    we added to indexes before Epoxy validation, actors could appear in lists with duplicate
 	//    key (since reservation wasn't confirmed yet).
 
+	let validation_res = ctx
+		.activity(ValidateInput {
+			name: input.name.clone(),
+			key: input.key.clone(),
+			namespace_id: input.namespace_id,
+			input: input.input.clone(),
+		})
+		.await?;
+
+	if let Err(error) = validation_res {
+		ctx.msg(Failed { error })
+			.topic(("actor_id", input.actor_id))
+			.send()
+			.await?;
+
+		return Ok(());
+	}
+
 	ctx.activity(InitStateAndUdbInput {
 		actor_id: input.actor_id,
 		name: input.name.clone(),
 		pool_name: input.pool_name.clone(),
 		key: input.key.clone(),
+		crash_policy: input.crash_policy,
 		namespace_id: input.namespace_id,
 		create_ts: ctx.create_ts(),
 		from_v1: input.from_v1,
@@ -224,13 +253,63 @@ pub async fn pegboard_actor2(ctx: &mut WorkflowCtx, input: &Input) -> Result<()>
 	destroy(ctx, input).await
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize, Hash)]
+pub struct ValidateInput {
+	pub namespace_id: Id,
+	pub name: String,
+	pub key: Option<String>,
+	pub input: Option<String>,
+}
+
+#[activity(Validate)]
+pub async fn validate(
+	ctx: &ActivityCtx,
+	input: &ValidateInput,
+) -> Result<std::result::Result<(), errors::Actor>> {
+	let ns_res = ctx
+		.op(namespace::ops::get_global::Input {
+			namespace_ids: vec![input.namespace_id],
+		})
+		.await?;
+
+	if ns_res.is_empty() {
+		return Ok(Err(errors::Actor::NamespaceNotFound));
+	};
+
+	if input
+		.input
+		.as_ref()
+		.map(|x| x.len() > MAX_INPUT_SIZE)
+		.unwrap_or_default()
+	{
+		return Ok(Err(errors::Actor::InputTooLarge {
+			max_size: MAX_INPUT_SIZE,
+		}));
+	}
+
+	if let Some(k) = &input.key {
+		if k.is_empty() {
+			return Ok(Err(errors::Actor::EmptyKey));
+		}
+		if k.len() > 1024 {
+			return Ok(Err(errors::Actor::KeyTooLarge {
+				max_size: 1024,
+				key_preview: util::safe_slice(k, 0, 1024).to_string(),
+			}));
+		}
+	}
+
+	Ok(Ok(()))
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, Hash)]
 pub struct InitStateAndUdbInput {
 	pub actor_id: Id,
 	pub name: String,
 	pub key: Option<String>,
 	pub namespace_id: Id,
 	pub pool_name: String,
+	pub crash_policy: CrashPolicy,
 	pub create_ts: i64,
 	pub from_v1: bool,
 }
@@ -244,6 +323,7 @@ pub async fn insert_state_and_db(ctx: &ActivityCtx, input: &InitStateAndUdbInput
 		input.name.clone(),
 		input.pool_name.clone(),
 		input.key.clone(),
+		input.crash_policy,
 		input.namespace_id,
 		input.create_ts,
 	));
@@ -343,6 +423,17 @@ pub async fn populate_indexes(ctx: &ActivityCtx, input: &PopulateIndexesInput) -
 					ctx.workflow_id(),
 				)?;
 
+				// Write the name into the namespace actor names list with empty metadata if it does not already exist.
+				let name_key = crate::keys::ns::ActorNameKey::new(namespace_id, name.clone());
+				if !tx.exists(&name_key, Serializable).await? {
+					tx.write(
+						&name_key,
+						ActorNameKeyData {
+							metadata: serde_json::Map::new(),
+						},
+					)?;
+				}
+
 				// NOTE: keys::ns::ActorByKeyKey is written in actor_keys.rs when reserved by epoxy
 
 				Ok(())
diff --git a/engine/packages/pegboard/src/workflows/actor2/runtime.rs b/engine/packages/pegboard/src/workflows/actor2/runtime.rs
@@ -4,6 +4,7 @@ use futures_util::TryStreamExt;
 use gas::prelude::*;
 use rand::prelude::SliceRandom;
 use rivet_envoy_protocol::{self as protocol, PROTOCOL_VERSION, versioned};
+use rivet_types::actors::CrashPolicy;
 use rivet_types::runner_configs::RunnerConfigKind;
 use universaldb::prelude::*;
 use universalpubsub::PublishOpts;
@@ -299,6 +300,8 @@ pub async fn allocate(ctx: &ActivityCtx, input: &AllocateInput) -> Result<Alloca
 	state.acquired_slot = acquired_slot;
 	state.error = error;
 	state.envoy_last_command_idx = 0;
+	state.sleep_ts = None;
+	state.reschedule_ts = None;
 
 	Ok(AllocateOutput {
 		allocation,
@@ -608,7 +611,11 @@ pub async fn handle_stopped(
 				code: protocol::StopCode::Error,
 				..
 			}
-			| StoppedVariant::Lost { .. } => Decision::Sleep,
+			| StoppedVariant::Lost { .. } => match input.crash_policy {
+				CrashPolicy::Restart => Decision::Backoff,
+				CrashPolicy::Sleep => Decision::Sleep,
+				CrashPolicy::Destroy => Decision::Destroy,
+			},
 		},
 	};
 
@@ -617,14 +624,20 @@ pub async fn handle_stopped(
 			let allocate_res = ctx.activity(AllocateInput {}).await?;
 
 			if let Some(allocation) = allocate_res.allocation {
-				start_new_generation(ctx, input, state, allocation).await?;
-
-				// Transition to allocating
-				state.transition = Transition::Allocating {
-					destroy_after_start: false,
-					lost_timeout_ts: allocate_res.now
-						+ ctx.config().pegboard().actor_allocation_threshold(),
+				state.transition = match &allocation {
+					Allocation::Serverless => Transition::Allocating {
+						destroy_after_start: false,
+						lost_timeout_ts: allocate_res.now
+							+ ctx.config().pegboard().actor_allocation_threshold(),
+					},
+					Allocation::Serverful { .. } => Transition::Starting {
+						destroy_after_start: false,
+						lost_timeout_ts: allocate_res.now
+							+ ctx.config().pegboard().actor_start_threshold(),
+					},
 				};
+
+				start_new_generation(ctx, input, state, allocation).await?;
 			} else {
 				// Transition to retry backoff
 				state.transition = Transition::Reallocating {
@@ -658,13 +671,20 @@ pub async fn handle_stopped(
 				let allocate_res = ctx.activity(AllocateInput {}).await?;
 
 				if let Some(allocation) = allocate_res.allocation {
-					start_new_generation(ctx, input, state, allocation).await?;
-
-					state.transition = Transition::Allocating {
-						destroy_after_start: false,
-						lost_timeout_ts: allocate_res.now
-							+ ctx.config().pegboard().actor_allocation_threshold(),
+					state.transition = match &allocation {
+						Allocation::Serverless => Transition::Allocating {
+							destroy_after_start: false,
+							lost_timeout_ts: allocate_res.now
+								+ ctx.config().pegboard().actor_allocation_threshold(),
+						},
+						Allocation::Serverful { .. } => Transition::Starting {
+							destroy_after_start: false,
+							lost_timeout_ts: allocate_res.now
+								+ ctx.config().pegboard().actor_start_threshold(),
+						},
 					};
+
+					start_new_generation(ctx, input, state, allocation).await?;
 				} else {
 					state.transition = Transition::Reallocating {
 						since_ts: allocate_res.now,
