wip: fix whatever is going on with envoys

Author: NathanFlurry

engine/packages/pegboard-envoy/src/conn.rs

@@ -147,7 +147,7 @@ pub async fn handle_init(
 				// Read existing data
 				let (create_ts_entry, old_last_ping_ts_entry, version_entry) = tokio::try_join!(
 					tx.read_opt(&create_ts_key, Serializable),
-					tx.read_opt(&create_ts_key, Serializable),
+					tx.read_opt(&last_ping_ts_key, Serializable),
 					tx.read_opt(&version_key, Serializable),
 				)?;
 
@@ -227,7 +227,7 @@ pub async fn handle_init(
 						envoy_key.clone(),
 					);
 
-					tx.add_conflict_key(&old_lb_key, ConflictRangeType::Read)?;
+					tx.delete(&old_lb_key);
 				}
 
 				// Insert into LB

engine/packages/pegboard-gateway2/src/lib.rs

@@ -832,7 +832,7 @@ impl PegboardGateway2 {
 			})
 			.await?
 		{
-			if actor.envoy_key.is_some() {
+			if actor.envoy_key.is_some() && actor.connectable {
 				tracing::debug!("actor became ready during hibernation");
 
 				return Ok(HibernationResult::Continue);

engine/packages/pegboard-kv-channel/src/lib.rs

@@ -434,9 +434,10 @@ async fn actor_request_task(
 	open_actors: Arc<Mutex<HashSet<String>>>,
 	mut rx: mpsc::Receiver<protocol::ToServerRequest>,
 ) {
-	// Cached actor resolution. Populated on first KV request, reused for all
-	// subsequent requests. Actor name is immutable so this never goes stale.
-	let mut cached_actor: Option<(Id, String)> = None;
+	// Cached actor resolution per actor id. A single KV channel connection can
+	// multiplex requests for many actors, so the cache must be keyed by actor id.
+	// Actor name is immutable so each cached entry does not go stale.
+	let mut cached_actors: HashMap<String, (Id, String)> = HashMap::new();
 
 	while let Some(req) = rx.recv().await {
 		let is_close = matches!(req.data, protocol::RequestData::ActorCloseRequest);
@@ -464,11 +465,11 @@ async fn actor_request_task(
 						)
 					}
 				} else {
-					// Lazy-resolve and cache.
-					if cached_actor.is_none() {
+					// Lazy-resolve and cache per actor id.
+					if !cached_actors.contains_key(&req.actor_id) {
 						match resolve_actor(&ctx, &req.actor_id, namespace_id).await {
 							Ok(v) => {
-								cached_actor = Some(v);
+								cached_actors.insert(req.actor_id.clone(), v);
 							}
 							Err(resp) => {
 								// Don't cache failures. Next request will retry.
@@ -480,7 +481,8 @@ async fn actor_request_task(
 							}
 						}
 					}
-					let (parsed_id, actor_name) = cached_actor.as_ref().unwrap();
+					let (parsed_id, actor_name) =
+						cached_actors.get(&req.actor_id).unwrap();
 
 					let recipient = actor_kv::Recipient {
 						actor_id: *parsed_id,
@@ -811,12 +813,21 @@ async fn resolve_actor(
 		)
 	})?;
 
+	// Resolve the actor from the general actor record rather than the
+	// runner-specific lookup. SQLite opens happen during actor startup
+	// before runner-specific fields are guaranteed to exist for v2 actors,
+	// so requiring RunnerIdKey here can race the database bootstrap path and
+	// reject perfectly valid opens with SQLITE_CANTOPEN.
 	let actor = ctx
-		.op(pegboard::ops::actor::get_for_runner::Input {
-			actor_id: parsed_id,
+		.op(pegboard::ops::actor::get::Input {
+			actor_ids: vec![parsed_id],
+			fetch_error: false,
 		})
 		.await
-		.map_err(|err| internal_error(&err))?;
+		.map_err(|err| internal_error(&err))?
+		.actors
+		.into_iter()
+		.next();
 
 	match actor {
 		Some(actor) => {

engine/packages/pegboard/src/ops/actor/create.rs

@@ -1,3 +1,5 @@
+use std::time::Duration;
+
 use anyhow::{Context, Result};
 use gas::prelude::*;
 use rivet_api_util::{Method, request_remote_datacenter};
@@ -47,7 +49,11 @@ pub async fn pegboard_actor_create(ctx: &OperationCtx, input: &Input) -> Result<
 		ctx.subscribe::<crate::workflows::actor2::DestroyStarted>(("actor_id", input.actor_id)),
 		ctx.op(crate::ops::runner_config::get::Input {
 			runners: vec![(input.namespace_id, input.runner_name_selector.clone())],
-			bypass_cache: false,
+			// This decides whether we dispatch the legacy runner workflow or the
+			// envoy-backed actor v2 workflow. The protocol version is updated
+			// during runner/envoy connect, so a cached read can lag behind and
+			// misclassify a freshly connected pool as legacy.
+			bypass_cache: true,
 		}),
 	)?;
 
@@ -75,6 +81,7 @@ pub async fn pegboard_actor_create(ctx: &OperationCtx, input: &Input) -> Result<
 
 		// Wait for actor creation to complete, fail, or be destroyed
 		tokio::select! {
+			biased;
 			res = create_sub2.next() => { res?; },
 			res = fail_sub2.next() => {
 				let msg = res?;
@@ -105,6 +112,14 @@ pub async fn pegboard_actor_create(ctx: &OperationCtx, input: &Input) -> Result<
 			}
 			res = destroy_sub2.next() => {
 				res?;
+				if let Ok(Ok(msg)) = tokio::time::timeout(
+					Duration::from_millis(50),
+					fail_sub2.next(),
+				)
+				.await
+				{
+					return Err(msg.error.clone().build());
+				}
 				return Err(crate::errors::Actor::DestroyedDuringCreation.build());
 			}
 		}
@@ -125,6 +140,7 @@ pub async fn pegboard_actor_create(ctx: &OperationCtx, input: &Input) -> Result<
 
 		// Wait for actor creation to complete, fail, or be destroyed
 		tokio::select! {
+			biased;
 			res = create_sub.next() => { res?; },
 			res = fail_sub.next() => {
 				let msg = res?;
@@ -155,6 +171,14 @@ pub async fn pegboard_actor_create(ctx: &OperationCtx, input: &Input) -> Result<
 			}
 			res = destroy_sub.next() => {
 				res?;
+				if let Ok(Ok(msg)) = tokio::time::timeout(
+					Duration::from_millis(50),
+					fail_sub.next(),
+				)
+				.await
+				{
+					return Err(msg.error.clone().build());
+				}
 				return Err(crate::errors::Actor::DestroyedDuringCreation.build());
 			}
 		}

engine/packages/pegboard/src/ops/actor/get_for_key.rs

@@ -1,7 +1,12 @@
 use anyhow::Result;
+use futures_util::TryStreamExt;
 use gas::prelude::*;
 use rivet_api_util::{Method, request_remote_datacenter};
 use rivet_types::actors::Actor;
+use universaldb::options::StreamingMode;
+use universaldb::utils::IsolationLevel::Serializable;
+
+use crate::keys;
 
 #[derive(Debug)]
 pub struct Input {
@@ -34,21 +39,67 @@ pub async fn pegboard_actor_get_for_key(ctx: &OperationCtx, input: &Input) -> Re
 
 	// Check if the actor is in the current datacenter
 	if reservation_id.label() == ctx.config().dc_label() {
-		// Local datacenter - get the actor directly
-		let actors_res = ctx
-			.op(crate::ops::actor::list_for_ns::Input {
-				namespace_id: input.namespace_id,
-				name: input.name.clone(),
-				key: Some(input.key.clone()),
-				include_destroyed: false,
-				created_before: None,
-				limit: 1,
-				fetch_error: input.fetch_error,
+		// Local datacenter - resolve with a serializable read so actors that are
+		// already tearing down do not leak through as a valid get/getOrCreate
+		// target.
+		let actors_with_wf_ids = ctx
+			.udb()?
+			.run(|tx| async move {
+				let tx = tx.with_subspace(keys::subspace());
+
+				let actor_subspace = keys::subspace().subspace(&keys::ns::ActorByKeyKey::subspace(
+					input.namespace_id,
+					input.name.clone(),
+					input.key.clone(),
+				));
+				let (start, end) = actor_subspace.range();
+
+				let mut stream = tx.get_ranges_keyvalues(
+					universaldb::RangeOption {
+						mode: StreamingMode::Iterator,
+						reverse: true,
+						..(start, end).into()
+					},
+					Serializable,
+				);
+
+				let mut results = Vec::new();
+
+				while let Some(entry) = stream.try_next().await? {
+					let (idx_key, data) = tx.read_entry::<keys::ns::ActorByKeyKey>(&entry)?;
+
+					if !data.is_destroyed {
+						results.push((idx_key.actor_id, data.workflow_id));
+						break;
+					}
+				}
+
+				Ok(results)
 			})
 			.await?;
 
+		let wfs = ctx
+			.get_workflows(
+				actors_with_wf_ids
+					.iter()
+					.map(|(_, workflow_id)| *workflow_id)
+					.collect(),
+			)
+			.await?;
+
+		let dc_name = ctx.config().dc_name()?.to_string();
+
+		let actors = super::util::build_actors_from_workflows(
+			ctx,
+			actors_with_wf_ids,
+			wfs,
+			&dc_name,
+			input.fetch_error,
+		)
+		.await?;
+
 		Ok(Output {
-			actor: actors_res.actors.into_iter().next(),
+			actor: actors.into_iter().next(),
 		})
 	} else {
 		// Remote datacenter - request from the correct datacenter

engine/packages/pegboard/src/workflows/actor2/mod.rs

@@ -672,22 +672,30 @@ async fn process_signal(
 							} = &mut state.transition
 							{
 								if *destroy_after_start {
-									// Transition to destroying
-									state.transition = Transition::Destroying {
-										envoy: runtime::EnvoyState::new(sig.envoy_key.clone()),
-										lost_timeout_ts: now
-											+ ctx.config().pegboard().actor_stop_threshold(),
+									state.transition = Transition::Running {
+										envoy: runtime::EnvoyState::new(
+											sig.envoy_key.clone(),
+										),
+										last_liveness_check_ts: now,
 									};
 
-									ctx.activity(runtime::InsertAndSendCommandsInput {
+									ctx.activity(runtime::SetConnectableInput {
+										envoy_key: sig.envoy_key.clone(),
 										generation: state.generation,
+									})
+									.await?;
+
+									ctx.msg(Ready {
 										envoy_key: sig.envoy_key.clone(),
-										commands: vec![protocol::Command::CommandStopActor(
-											protocol::CommandStopActor {
-												reason: protocol::StopActorReason::Destroy,
-											},
-										)],
 									})
+									.topic(("actor_id", input.actor_id))
+									.send()
+									.await?;
+
+									let workflow_id = ctx.workflow_id();
+									ctx.signal(Destroy {})
+										.to_workflow_id(workflow_id)
+										.send()
 									.await?;
 								} else {
 									// Transition to starting
@@ -804,10 +812,15 @@ async fn process_signal(
 					runtime::reschedule_actor(ctx, &input, state, metrics_workflow_id).await?;
 				}
 				Transition::SleepIntent {
-					rewake_after_stop, ..
+					lost_timeout_ts,
+					rewake_after_stop,
+					..
 				} => {
 					if !*rewake_after_stop {
 						*rewake_after_stop = true;
+						let now = ctx.activity(runtime::GetTsInput {}).await?;
+						*lost_timeout_ts =
+							now + ctx.config().pegboard().actor_stop_threshold();
 
 						tracing::debug!(
 							actor_id=?input.actor_id,
@@ -983,6 +996,8 @@ async fn process_signal(
 						lost_timeout_ts: now + ctx.config().pegboard().actor_stop_threshold(),
 					};
 
+					ctx.activity(runtime::SetNotConnectableInput {}).await?;
+
 					ctx.activity(runtime::InsertAndSendCommandsInput {
 						generation: state.generation,
 						envoy_key,
@@ -1004,6 +1019,8 @@ async fn process_signal(
 						lost_timeout_ts: now + ctx.config().pegboard().actor_stop_threshold(),
 					};
 
+					ctx.activity(runtime::SetNotConnectableInput {}).await?;
+
 					// Stop command was already sent
 				}
 				Transition::Allocating {