chore(core): update default region name to default

Author: NathanFlurry

out/errors/guard.invalid_regional_host.json

@@ -48,19 +48,20 @@ impl Default for Topology {
 		Topology {
 			datacenter_label: 1,
 			datacenters: vec![Datacenter {
-				name: "local".into(),
+				name: "default".into(),
 				datacenter_label: 1,
 				is_leader: true,
-				api_peer_url: Url::parse(&format!(
+				public_url: Url::parse(&format!(
 					"http://127.0.0.1:{}",
-					crate::defaults::ports::API_PEER
+					crate::defaults::ports::GUARD
 				))
 				.unwrap(),
-				guard_url: Url::parse(&format!(
+				api_peer_url: Url::parse(&format!(
 					"http://127.0.0.1:{}",
-					crate::defaults::ports::GUARD
+					crate::defaults::ports::API_PEER
 				))
 				.unwrap(),
+				valid_hosts: None,
 			}],
 		}
 	}
@@ -72,8 +73,28 @@ pub struct Datacenter {
 	pub name: String,
 	pub datacenter_label: u16,
 	pub is_leader: bool,
-	/// Url of the api-peer service
+	/// Public origin that can be used to connect to this region.
+	pub public_url: Url,
+	/// URL of the api-peer service
 	pub api_peer_url: Url,
-	/// Url of the peer's guard server
-	pub guard_url: Url,
+	/// List of hosts that are valid to connect to this region with. This is used in regional
+	/// endpoints to validate that incoming requests to this datacenter are going to a
+	/// region-specific domain.
+	///
+	/// IMPORTANT: Do not use a global origin that routes to multiple different regions. This will
+	/// cause unpredictable behavior when requests are expected to go to a specific region.
+	#[serde(default)]
+	pub valid_hosts: Option<Vec<String>>,
+}
+
+impl Datacenter {
+	pub fn is_valid_regional_host(&self, host: &str) -> bool {
+		if let Some(valid_hosts) = &self.valid_hosts {
+			// Check if host is in the valid_hosts list
+			valid_hosts.iter().any(|valid_host| valid_host == host)
+		} else {
+			// Check if host matches the origin of public_url
+			self.public_url.host_str() == Some(host)
+		}
+	}
 }

out/errors/guard.must_use_regional_host.json

@@ -48,8 +48,9 @@ impl TestDeps {
 				name: format!("dc-{dc_id}"),
 				datacenter_label: dc_id,
 				is_leader: dc_id == dc_ids[0], // First DC in list is leader
+				public_url: Url::parse(&format!("http://127.0.0.1:{guard_port}"))?,
 				api_peer_url: Url::parse(&format!("http://127.0.0.1:{api_peer_port}"))?,
-				guard_url: Url::parse(&format!("http://127.0.0.1:{guard_port}"))?,
+				valid_hosts: None,
 			});
 			ports.push((api_peer_port, guard_port));
 		}

packages/common/config/src/config/topology.rs

@@ -69,3 +69,16 @@ pub struct ActorDestroyed {
 pub struct ActorReadyTimeout {
 	pub actor_id: Id,
 }
+
+#[derive(RivetError, Serialize)]
+#[error(
+	"guard",
+	"must_use_regional_host",
+	"Request must use a regional URL for this datacenter.",
+	"Invalid host {host} for datacenter {datacenter}. Please use one of the following hosts: {valid_hosts}"
+)]
+pub struct MustUseRegionalHost {
+	pub host: String,
+	pub datacenter: String,
+	pub valid_hosts: String,
+}

packages/common/test-deps/src/lib.rs

@@ -77,14 +77,14 @@ pub async fn route_request(
 			targets: vec![RouteTarget {
 				actor_id: Some(actor_id),
 				host: peer_dc
-					.guard_url
+					.public_url
 					.host()
-					.context("peer dc guard_url has no host")?
+					.context("peer dc public_url has no host")?
 					.to_string(),
 				port: peer_dc
-					.guard_url
+					.public_url
 					.port()
-					.context("peer dc guard_url has no port")?,
+					.context("peer dc public_url has no port")?,
 				path: path.to_owned(),
 			}],
 			timeout: RoutingTimeout {

packages/core/guard/server/src/errors.rs

@@ -11,14 +11,38 @@ pub(crate) const WS_PROTOCOL_TOKEN: &str = "rivet_token.";
 pub async fn route_request(
 	ctx: &StandaloneCtx,
 	target: &str,
-	_host: &str,
+	host: &str,
 	_path: &str,
 	headers: &hyper::HeaderMap,
 ) -> Result<Option<RoutingOutput>> {
 	if target != "runner" {
 		return Ok(None);
 	}
 
+	// Validate that the host is valid for the current datacenter
+	let current_dc = ctx.config().topology().current_dc()?;
+	if !current_dc.is_valid_regional_host(host) {
+		tracing::warn!(?host, datacenter = ?current_dc.name, "invalid host for current datacenter");
+
+		// Determine valid hosts for error message
+		let valid_hosts = if let Some(hosts) = &current_dc.valid_hosts {
+			hosts.join(", ")
+		} else {
+			current_dc
+				.public_url
+				.host_str()
+				.map(|h| h.to_string())
+				.unwrap_or_else(|| "unknown".to_string())
+		};
+
+		return Err(crate::errors::MustUseRegionalHost {
+			host: host.to_string(),
+			datacenter: current_dc.name.clone(),
+			valid_hosts,
+		}
+		.build());
+	}
+
 	let is_websocket = headers
 		.get("upgrade")
 		.and_then(|v| v.to_str().ok())

packages/core/guard/server/src/routing/pegboard_gateway.rs

@@ -87,7 +87,7 @@ pub async fn check_config_changes(
 				replica_id: dc.datacenter_label as u64,
 				status: status.into(),
 				api_peer_url: dc.api_peer_url.to_string(),
-				guard_url: dc.guard_url.to_string(),
+				guard_url: dc.public_url.to_string(),
 			}
 		})
 		.collect::<Vec<types::ReplicaConfig>>();
@@ -277,7 +277,7 @@ fn should_abort_reconfigure(
 			return Ok(true);
 		}
 
-		if url::Url::parse(&replica.guard_url)? != current_dc.guard_url {
+		if url::Url::parse(&replica.guard_url)? != current_dc.public_url {
 			tracing::info!(
 				"config changed during reconfigure (guard_url changed), aborting reconfigure"
 			);

packages/core/guard/server/src/routing/runner.rs

@@ -0,0 +1,164 @@
+use std::time::Duration;
+
+use anyhow::Result;
+use futures_util::{StreamExt, TryFutureExt, stream::FuturesUnordered};
+use gas::prelude::*;
+use rivet_api_types::runners::list as runners_list;
+use rivet_api_util::{HeaderMap, Method, request_remote_datacenter};
+use serde::de::DeserializeOwned;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct Input {
+	pub namespace_id: Id,
+	pub runner_name: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct Output {
+	pub dc_label: Option<u16>,
+}
+
+/// Finds a datacenter that contains a given runner name in a namespace.
+///
+/// This is core to determining which datacenter actors should run in, since actors can only run in
+/// datacenters with supported runners.
+#[operation]
+pub async fn find_dc_with_runner(ctx: &OperationCtx, input: &Input) -> Result<Output> {
+	// TODO: We should figure out how to pre-emptively validate this cache so we don't have
+	// "stutters" where every 15s we have a high request duration
+	ctx.cache()
+		.clone()
+		.request()
+		.ttl(15_000)
+		.fetch_one_json(
+			"runner.find_dc_with_runner",
+			(input.namespace_id, input.runner_name.clone()),
+			{
+				move |mut cache, key| async move {
+					let dc_id = find_dc_with_runner_inner(ctx, input).await?;
+
+					if let Some(dc_id) = dc_id {
+						cache.resolve(&key, dc_id);
+					}
+
+					Ok(cache)
+				}
+			},
+		)
+		.await
+		.map(|dc_label| Output { dc_label })
+}
+
+async fn find_dc_with_runner_inner(ctx: &OperationCtx, input: &Input) -> Result<Option<u16>> {
+	// Check if this DC has any runners
+	let res = ctx
+		.op(super::list_for_ns::Input {
+			namespace_id: input.namespace_id,
+			name: Some(input.runner_name.clone()),
+			include_stopped: false,
+			created_before: None,
+			limit: 1,
+		})
+		.await?;
+	if !res.runners.is_empty() {
+		return Ok(Some(ctx.config().dc_label()));
+	}
+
+	// Get namespace
+	let namespace = ctx
+		.op(namespace::ops::get_global::Input {
+			namespace_ids: vec![input.namespace_id],
+		})
+		.await?
+		.into_iter()
+		.next()
+		.ok_or_else(|| namespace::errors::Namespace::NotFound.build())?;
+
+	// Fanout to all datacenters
+	let runners =
+		race_request_to_datacenters::<runners_list::ListQuery, runners_list::ListResponse, _>(
+			ctx,
+			Default::default(),
+			"/runners",
+			runners_list::ListQuery {
+				namespace: namespace.name,
+				name: Some(input.runner_name.clone()),
+				runner_ids: None,
+				include_stopped: Some(false),
+				limit: Some(1),
+				cursor: None,
+			},
+			|res| !res.runners.is_empty(),
+		)
+		.await?;
+
+	Ok(runners.map(|x| x.0))
+}
+
+const REQUEST_TIMEOUT: Duration = Duration::from_secs(5);
+
+/// Helper fn that will send a request to all datacenters and return the response of the first
+/// datacenter that matches `filter`.
+pub async fn race_request_to_datacenters<Q, R, F>(
+	ctx: &OperationCtx,
+	headers: HeaderMap,
+	endpoint: &str,
+	query: Q,
+	filter: F,
+) -> Result<Option<(u16, R)>>
+where
+	R: DeserializeOwned + Send + 'static,
+	Q: Serialize + Clone + Send + 'static,
+	F: Fn(&R) -> bool,
+{
+	// Create futures for all dcs except the current
+	let dcs = &ctx.config().topology().datacenters;
+	let mut responses = futures_util::stream::iter(
+		dcs.iter()
+			.filter(|dc| dc.datacenter_label != ctx.config().dc_label())
+			.map(|dc| {
+				let headers = headers.clone();
+				let query = query.clone();
+				async move {
+					tokio::time::timeout(
+						REQUEST_TIMEOUT,
+						// Remote datacenter
+						request_remote_datacenter::<R>(
+							ctx.config(),
+							dc.datacenter_label,
+							&endpoint,
+							Method::GET,
+							headers,
+							Some(&query),
+							Option::<&()>::None,
+						)
+						.map_ok(|x| (dc.datacenter_label, x)),
+					)
+					.await
+				}
+			}),
+	)
+	.collect::<FuturesUnordered<_>>()
+	.await;
+
+	// Collect responses until we reach quorum or all futures complete
+	while let Some(out) = responses.next().await {
+		match out {
+			std::result::Result::Ok(result) => match result {
+				std::result::Result::Ok((dc_label, response)) => {
+					if filter(&response) {
+						return Ok(Some((dc_label, response)));
+					}
+				}
+				std::result::Result::Err(err) => {
+					tracing::warn!(?err, "received error from replica");
+				}
+			},
+			std::result::Result::Err(err) => {
+				tracing::warn!(?err, "received timeout from replica");
+			}
+		}
+	}
+
+	Ok(None)
+}

packages/services/epoxy/src/workflows/coordinator/reconfigure.rs

@@ -1,3 +1,4 @@
+pub mod find_dc_with_runner;
 pub mod get;
 pub mod get_by_key;
 pub mod list_for_ns;