fix(guard): use intercom_address for guard service discovery

Author: NathanFlurry

docker/dev-full/rivet-edge-server/config.jsonc

@@ -8,7 +8,7 @@
 				"cluster_id": "11ca8960-acab-4963-909c-99d72af3e1cb",
 				"datacenter_id": "f288913c-735d-4188-bf9b-2fcf6eac7b9c",
 				"server_id": "174aca2a-98b7-462c-9ad9-3835094a9a10",
-				"intercom_endpoint": "http://rivet-server:8081"
+				"intercom_address": "http://rivet-server:8081"
 			},
 			"guard": {
 				// TLS not configured for local development

docker/dev-full/rivet-guard/config.jsonc

@@ -20,7 +20,7 @@
 				"cluster_id": "11ca8960-acab-4963-909c-99d72af3e1cb",
 				"datacenter_id": "f288913c-735d-4188-bf9b-2fcf6eac7b9c",
 				"server_id": "174aca2a-98b7-462c-9ad9-3835094a9a10",
-				"intercom_endpoint": "http://rivet-server:8081"
+				"intercom_address": "http://rivet-server:8081"
 			},
 			"guard": {
 				// TLS not configured for local development

packages/common/config/src/config/server/rivet/mod.rs

@@ -841,10 +841,20 @@ pub struct Edge {
 	pub cluster_id: Uuid,
 	pub datacenter_id: Uuid,
 	pub server_id: Uuid,
-	pub intercom_endpoint: Url,
+	/// Url of the core cluster.
+	pub intercom_address: Url,
 	/// This API address will be used if there are no worker servers listed in the cluster package
 	#[serde(default)]
 	pub api_lan_address: Option<(String, u16)>,
 	#[serde(default)]
 	pub redirect_logs_dir: Option<PathBuf>,
 }
+
+impl Edge {
+	/// u16 used in the `Id` type for actors
+	pub fn datacenter_label(&self) -> u16 {
+		// Read first 2 bytes
+		let bytes = self.datacenter_id.as_bytes();
+		u16::from_be_bytes([bytes[0], bytes[1]])
+	}
+}

packages/core/services/cluster/src/workflows/server/install/install_scripts/components/rivet/guard.rs

@@ -58,7 +58,7 @@ pub fn configure(config: &rivet_config::Config) -> GlobalResult<String> {
 					datacenter_id: Uuid::nil(),
 					server_id: Uuid::nil(),
 					api_lan_address: None,
-					intercom_endpoint: Url::parse(&format!("http://127.0.0.1:{TUNNEL_API_EDGE_PORT}"))?,
+					intercom_address: Url::parse(&format!("http://127.0.0.1:{TUNNEL_API_EDGE_PORT}"))?,
 					redirect_logs_dir: Some(PathBuf::from("/var/log/rivet-guard")),
 				}),
 				status: server_config.rivet.status.clone(),

packages/core/services/cluster/src/workflows/server/install/install_scripts/components/rivet/worker.rs

@@ -72,7 +72,7 @@ pub fn configure(config: &rivet_config::Config) -> GlobalResult<String> {
 					datacenter_id: Uuid::nil(),
 					server_id: Uuid::nil(),
 					api_lan_address: None,
-					intercom_endpoint: Url::parse(&format!("http://127.0.0.1:{TUNNEL_API_EDGE_PORT}"))?,
+					intercom_address: Url::parse(&format!("http://127.0.0.1:{TUNNEL_API_EDGE_PORT}"))?,
 					redirect_logs_dir: Some(PathBuf::from("/var/log/rivet-edge-server")),
 				}),
 				..Default::default()

packages/edge/infra/guard/server/src/routing/api.rs

@@ -10,9 +10,6 @@ use service_discovery::ServiceDiscovery;
 use url::Url;
 use uuid::Uuid;
 
-// TODO: Copied from cluster/src/workflows/server/install/install_scripts/components/rivet/mod.rs
-const TUNNEL_API_EDGE_PORT: u16 = 5010;
-
 /// Route requests to the API service
 #[tracing::instrument(skip_all)]
 pub async fn route_api_request(
@@ -46,7 +43,8 @@ pub async fn route_api_request(
 
 	// NOTE: We use service discovery instead of server::list or datacenter::server_discovery because cache is not
 	// shared between edge-edge or edge-core. SD requests the core which has a single cache.
-	let url = Url::parse(&format!("http://127.0.0.1:{TUNNEL_API_EDGE_PORT}/provision/datacenters/{dc_id}/servers?pools=worker"))?;
+	let edge = ctx.config().server()?.rivet.edge()?;
+	let url = Url::parse(&format!("{}provision/datacenters/{dc_id}/servers?pools=worker", edge.intercom_address))?;
 	let sd = ServiceDiscovery::new(url);
 	let servers = sd.fetch().await?;
 

packages/edge/services/pegboard/src/workflows/client/mod.rs

@@ -142,23 +142,41 @@ pub async fn pegboard_client(ctx: &mut WorkflowCtx, input: &Input) -> GlobalResu
 									state,
 								} = event.inner.deserialize()?
 								{
+									// Try actor2 first
 									let res = ctx
 										.signal(crate::workflows::actor::StateUpdate {
 											generation,
-											state,
+											state: state.clone(),
 										})
-										.to_workflow::<crate::workflows::actor::Workflow>()
+										.to_workflow::<crate::workflows::actor2::Workflow>()
 										.tag("actor_id", actor_id)
 										.send()
 										.await;
 
 									if let Some(WorkflowError::WorkflowNotFound) =
 										res.as_workflow_error()
 									{
-										tracing::warn!(
-											?actor_id,
-											"actor workflow not found, likely already stopped"
-										);
+										// Try old actors
+										let res = ctx
+											.signal(crate::workflows::actor::StateUpdate {
+												generation,
+												state,
+											})
+											.to_workflow::<crate::workflows::actor::Workflow>()
+											.tag("actor_id", actor_id)
+											.send()
+											.await;
+
+										if let Some(WorkflowError::WorkflowNotFound) =
+											res.as_workflow_error()
+										{
+											tracing::warn!(
+												?actor_id,
+												"actor workflow not found, likely already stopped"
+											);
+										} else {
+											res?;
+										}
 									} else {
 										res?;
 									}
@@ -510,7 +528,7 @@ async fn publish_registered(ctx: &ActivityCtx, input: &PublishRegisteredInput) -
 
 	let config = Configuration {
 		client: rivet_pools::reqwest::client().await?,
-		base_path: util::url::to_string_without_slash(&edge.intercom_endpoint),
+		base_path: util::url::to_string_without_slash(&edge.intercom_address),
 		bearer_access_token: Some(token),
 		..Default::default()
 	};
@@ -772,6 +790,9 @@ pub async fn handle_commands(
 					}
 				}
 			}
+			protocol::Command::SignalRunner { .. } => {
+				// No-op in this workflow
+			}
 		}
 	}