fix(pegboard): route scoped actor key reads by runner dc

Author: NathanFlurry

engine/CLAUDE.md

@@ -39,6 +39,7 @@ When changing a versioned VBARE schema, follow the existing migration pattern.
 - UniversalDB simulated latency for benchmarks comes from `UDB_SIMULATED_LATENCY_MS`, which `Database::txn(...)` reads once via `OnceLock`, so set it before process startup.
 - When adding fields to epoxy workflow state structs, mark them `#[serde(default)]` so Gasoline can replay older serialized state.
 - Epoxy integration tests that spin up `tests/common::TestCtx` must call `shutdown()` before returning.
+- Before issuing an Epoxy operation with scoped `target_replicas`, validate the local replica is in scope or forward to an in-scope datacenter first.
 
 ## Test snapshots
 
@@ -71,7 +72,12 @@ Use `test-snapshot-gen` to generate and load RocksDB snapshots of the full UDB K
 
 ## Pegboard Envoy
 
+- Write new actor-hosting engine tests under `engine/packages/engine/tests/envoy/`; do not add new legacy runner tests under `engine/packages/engine/tests/runner/`.
 - `PegboardEnvoyWs::new(...)` is constructed per websocket request, so shared sqlite dispatch state such as the `SqliteEngine` and `CompactionCoordinator` must live behind a process-wide `OnceCell` instead of per-connection fields.
 - Restored hibernatable WebSockets must rebuild runtime WebSocket handlers from callbacks and call `on_open`; pre-sleep NAPI callbacks are not reusable after actor wake.
 - `pegboard-envoy` SQLite websocket handlers must validate page numbers, page sizes, and duplicate dirty pages at the websocket trust boundary and return `SqliteErrorResponse` for unexpected failures instead of bubbling them through the shared connection task.
 - SQLite start-command schema dispatch should probe actor KV prefix `0x08` at startup instead of persisting a schema version in pegboard config or actor workflow state.
+
+## API routing
+
+- `api-public` owns cross-datacenter forwarding for external requests; `api-peer` handlers should be local datacenter operations and must not add forwarding requirements.

engine/packages/api-peer/src/actors/get_or_create.rs

@@ -106,17 +106,11 @@ pub async fn get_or_create(
 				}
 			}
 		}
-		// Make request to remote datacenter
 		pegboard::ops::actor::get_for_key::Output::Forward { dc_label } => {
-			rivet_api_util::request_remote_datacenter(
-				ctx.config(),
-				dc_label,
-				"/actors",
-				rivet_api_util::Method::PUT,
-				Some(&query),
-				Some(&body),
-			)
-			.await
+			Err(pegboard::errors::Actor::KeyReservedInDifferentDatacenter {
+				datacenter_label: dc_label,
+			}
+			.build())
 		}
 	}
 }

engine/packages/api-public/src/actors/utils.rs

@@ -139,32 +139,38 @@ pub async fn find_dc_for_actor_creation(
 	runner_name: &str,
 	dc_name: Option<&str>,
 ) -> Result<u16> {
-	let target_dc_label = if let Some(dc_name) = &dc_name {
+	let requested_dc_label = if let Some(dc_name) = &dc_name {
 		// Use user-configured DC
-		ctx.config()
+		Some(ctx.config()
 			.dc_for_name(dc_name)
 			.ok_or_else(|| rivet_api_util::errors::Datacenter::NotFound.build())?
-			.datacenter_label
+			.datacenter_label)
 	} else {
-		// Find the nearest DC with runners
-		let res = ctx
-			.op(
-				pegboard::ops::runner::list_runner_config_enabled_dcs::Input {
-					namespace_id,
-					runner_name: runner_name.into(),
-				},
-			)
-			.await?;
-		if let Some(dc_label) = res.dc_labels.into_iter().next() {
-			dc_label
-		} else {
-			return Err(pegboard::errors::Actor::NoRunnerConfigConfigured {
-				namespace: namespace_name.into(),
-				pool_name: runner_name.into(),
-			}
-			.build());
-		}
+		None
+	};
+
+	let res = ctx
+		.op(
+			pegboard::ops::runner::list_runner_config_enabled_dcs::Input {
+				namespace_id,
+				runner_name: runner_name.into(),
+			},
+		)
+		.await?;
+
+	let target_dc_label = if let Some(requested_dc_label) = requested_dc_label {
+		res.dc_labels
+			.into_iter()
+			.find(|dc_label| *dc_label == requested_dc_label)
+	} else {
+		res.dc_labels.into_iter().next()
 	};
 
-	Ok(target_dc_label)
+	target_dc_label.ok_or_else(|| {
+		pegboard::errors::Actor::NoRunnerConfigConfigured {
+			namespace: namespace_name.into(),
+			pool_name: runner_name.into(),
+		}
+		.build()
+	})
 }

engine/packages/engine/tests/envoy/actors_lifecycle.rs

@@ -359,6 +359,106 @@ fn envoy_actor_connectable_via_guard_websocket() {
 	});
 }
 
+#[test]
+fn query_get_or_create_from_dc_without_runner_forwards_to_runner_dc() {
+	common::run(common::TestOpts::new(2).with_timeout(45), |ctx| async move {
+		let (namespace, _, _envoy) =
+			common::setup_test_namespace_with_envoy(ctx.leader_dc()).await;
+		let wrong_dc = ctx.get_dc(2);
+
+		let client = reqwest::Client::new();
+		let response = client
+			.get(format!(
+				"http://127.0.0.1:{}/gateway/test-actor/ping",
+				wrong_dc.guard_port()
+			))
+			.query(&[
+				("rvt-namespace", namespace.as_str()),
+				("rvt-method", "getOrCreate"),
+				("rvt-runner", common::TEST_RUNNER_NAME),
+				("rvt-key", "geo-routed-key"),
+			])
+			.send()
+			.await
+			.expect("failed to send query gateway request");
+
+		assert_eq!(
+			response.status(),
+			reqwest::StatusCode::OK,
+			"query gateway should forward to the runner dc"
+		);
+
+		let body: serde_json::Value = response.json().await.expect("invalid ping response");
+		let actor_id = body["actorId"].as_str().expect("missing actor id");
+		assert_eq!(body["status"], "ok");
+		common::assert_actor_in_dc(actor_id, ctx.leader_dc().config.dc_label()).await;
+	});
+}
+
+#[test]
+fn public_get_or_create_with_unavailable_datacenter_returns_typed_error() {
+	common::run(common::TestOpts::new(2).with_timeout(45), |ctx| async move {
+		let (namespace, _, _envoy) =
+			common::setup_test_namespace_with_envoy(ctx.leader_dc()).await;
+		let wrong_dc = ctx.get_dc(2);
+
+		let request = common::api::public::build_actors_get_or_create_request(
+			ctx.leader_dc().guard_port(),
+			common::api::public::GetOrCreateQuery {
+				namespace: namespace.clone(),
+			},
+			common::api::public::GetOrCreateRequest {
+				datacenter: Some(wrong_dc.config.dc_name().unwrap().to_string()),
+				name: "test-actor".to_string(),
+				key: "public-explicit-wrong-dc-key".to_string(),
+				input: None,
+				runner_name_selector: common::TEST_RUNNER_NAME.to_string(),
+				crash_policy: rivet_types::actors::CrashPolicy::Sleep,
+			},
+		)
+		.await
+		.expect("failed to build request");
+		let response = request.send().await.expect("failed to send request");
+
+		assert_eq!(response.status(), reqwest::StatusCode::BAD_REQUEST);
+		let body: serde_json::Value = response.json().await.expect("invalid error response");
+		assert_eq!(body["group"], "actor");
+		assert_eq!(body["code"], "no_runner_config_configured");
+	});
+}
+
+#[test]
+fn public_create_with_unavailable_datacenter_returns_typed_error() {
+	common::run(common::TestOpts::new(2).with_timeout(45), |ctx| async move {
+		let (namespace, _, _envoy) =
+			common::setup_test_namespace_with_envoy(ctx.leader_dc()).await;
+		let wrong_dc = ctx.get_dc(2);
+
+		let request = common::api::public::build_actors_create_request(
+			ctx.leader_dc().guard_port(),
+			common::api_types::actors::create::CreateQuery {
+				namespace: namespace.clone(),
+			},
+			common::api_types::actors::create::CreateRequest {
+				datacenter: Some(wrong_dc.config.dc_name().unwrap().to_string()),
+				name: "test-actor".to_string(),
+				key: Some("public-create-explicit-wrong-dc-key".to_string()),
+				input: None,
+				runner_name_selector: common::TEST_RUNNER_NAME.to_string(),
+				crash_policy: rivet_types::actors::CrashPolicy::Sleep,
+			},
+		)
+		.await
+		.expect("failed to build request");
+		let response = request.send().await.expect("failed to send request");
+
+		assert_eq!(response.status(), reqwest::StatusCode::BAD_REQUEST);
+		let body: serde_json::Value = response.json().await.expect("invalid error response");
+		assert_eq!(body["group"], "actor");
+		assert_eq!(body["code"], "no_runner_config_configured");
+	});
+}
+
 #[test]
 fn envoy_websocket_actor_close_round_trip() {
 	common::run(common::TestOpts::new(1).with_timeout(20), |ctx| async move {
@@ -1129,4 +1229,3 @@ fn envoy_normal_pool_does_not_apply_legacy_runner_slot_capacity() {
 		}
 	});
 }
-

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

@@ -132,14 +132,20 @@ async fn resolve_query_get_or_create(
 	let namespace_id = resolve_namespace_id(ctx, namespace_name).await?;
 	let serialized_key = serialize_actor_key(key)?;
 
+	let target_dc_label =
+		resolve_query_target_dc_label(ctx, namespace_id, namespace_name, pool_name, region).await?;
+	if target_dc_label != ctx.config().dc_label() {
+		return Ok(ResolveQueryActorResult::Forward {
+			dc_label: target_dc_label,
+		});
+	}
+
 	if let Some(res) =
 		get_actor_for_key(ctx, namespace_id, name, &serialized_key, Some(pool_name)).await?
 	{
 		return Ok(res);
 	}
 
-	let target_dc_label =
-		resolve_query_target_dc_label(ctx, namespace_id, namespace_name, pool_name, region).await?;
 	let encoded_input = input.map(|input| STANDARD.encode(input));
 
 	if target_dc_label == ctx.config().dc_label() {
@@ -185,13 +191,15 @@ async fn resolve_query_target_dc_label(
 	runner_name_selector: &str,
 	region: Option<&str>,
 ) -> Result<u16> {
-	if let Some(region) = region {
-		return Ok(ctx
+	let requested_dc_label = if let Some(region) = region {
+		Some(ctx
 			.config()
 			.dc_for_name(region)
 			.ok_or_else(|| rivet_api_util::errors::Datacenter::NotFound.build())?
-			.datacenter_label);
-	}
+			.datacenter_label)
+	} else {
+		None
+	};
 
 	let res = ctx
 		.op(
@@ -202,8 +210,15 @@ async fn resolve_query_target_dc_label(
 		)
 		.await?;
 
-	// Return nearest enabled dc
-	if let Some(dc_label) = res.dc_labels.into_iter().next() {
+	let target_dc_label = if let Some(requested_dc_label) = requested_dc_label {
+		res.dc_labels
+			.into_iter()
+			.find(|dc_label| *dc_label == requested_dc_label)
+	} else {
+		res.dc_labels.into_iter().next()
+	};
+
+	if let Some(dc_label) = target_dc_label {
 		Ok(dc_label)
 	} else {
 		Err(pegboard::errors::Actor::NoRunnerConfigConfigured {

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

@@ -1,3 +1,4 @@
+use anyhow::ensure;
 use gas::prelude::*;
 use universaldb::utils::FormalKey;
 
@@ -33,6 +34,11 @@ pub async fn pegboard_actor_get_reservation_for_key(
 			)
 			.await?;
 
+		ensure!(
+			res.replicas.contains(&ctx.config().epoxy_replica_id()),
+			"get_reservation_for_key called outside the scoped runner replica set"
+		);
+
 		Some(res.replicas)
 	} else {
 		None