fix: fix epoxy replica ids (#3223)

Author: MasterPtato

out/errors/guard.service_unavailable.json

@@ -97,9 +97,13 @@ pub async fn epoxy_replica_reconfigure(
 	_query: (),
 	_body: ReplicaReconfigureRequest,
 ) -> Result<ReplicaReconfigureResponse> {
-	ctx.signal(epoxy::workflows::coordinator::ReplicaReconfigure {})
-		.send()
-		.await?;
+	if ctx.config().is_leader() {
+		ctx.signal(epoxy::workflows::coordinator::ReplicaReconfigure {})
+			.to_workflow::<epoxy::workflows::coordinator::Workflow>()
+			.tag("replica", ctx.config().epoxy_replica_id())
+			.send()
+			.await?;
+	}
 
 	Ok(ReplicaReconfigureResponse {})
 }

packages/core/api-peer/src/internal.rs

@@ -69,6 +69,10 @@ pub struct ConnectionError {
 	pub remote_addr: String,
 }
 
+#[derive(RivetError, Serialize, Deserialize)]
+#[error("guard", "service_unavailable", "Service unavailable.")]
+pub struct ServiceUnavailable;
+
 #[derive(RivetError, Serialize, Deserialize)]
 #[error(
 	"guard",

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

@@ -6,8 +6,11 @@ use gas::prelude::*;
 use http_body_util::{BodyExt, Full};
 use hyper::{Request, Response, StatusCode, header::HeaderName};
 use rivet_guard_core::{
-	WebSocketHandle, custom_serve::CustomServeTrait, errors::WebSocketServiceUnavailable,
-	proxy_service::ResponseBody, request_context::RequestContext,
+	WebSocketHandle,
+	custom_serve::CustomServeTrait,
+	errors::{ServiceUnavailable, WebSocketServiceUnavailable},
+	proxy_service::ResponseBody,
+	request_context::RequestContext,
 };
 use rivet_runner_protocol as protocol;
 use rivet_util::serde::HashableMap;
@@ -122,20 +125,20 @@ impl CustomServeTrait for PegboardGateway {
 					},
 					TunnelMessageData::Timeout => {
 						tracing::warn!("tunnel message timeout");
-						return Err(WebSocketServiceUnavailable.build());
+						return Err(ServiceUnavailable.build());
 					}
 				}
 			}
 
 			tracing::warn!("received no message response");
-			Err(WebSocketServiceUnavailable.build())
+			Err(ServiceUnavailable.build())
 		};
 		let response_start = tokio::time::timeout(TUNNEL_ACK_TIMEOUT, fut)
 			.await
 			.map_err(|_| {
 				tracing::warn!("timed out waiting for tunnel ack");
 
-				WebSocketServiceUnavailable.build()
+				ServiceUnavailable.build()
 			})??;
 		tracing::debug!("response handler task ended");
 

packages/core/pegboard-gateway/src/lib.rs

@@ -16,7 +16,7 @@ use crate::types;
 
 #[tracing::instrument(skip_all)]
 pub async fn setup_replica(ctx: &mut WorkflowCtx, _input: &super::Input) -> Result<()> {
-	// Wait for cooridinator to send begin learning signal
+	// Wait for coordinator to send begin learning signal
 	let begin_learning = ctx.listen::<super::BeginLearning>().await?;
 
 	// TODO: Paralellize replicas

packages/services/epoxy/src/workflows/replica/setup.rs

@@ -5,7 +5,7 @@ use futures_util::{FutureExt, StreamExt, TryFutureExt, stream::FuturesUnordered}
 use gas::prelude::*;
 use rivet_api_types::{runner_configs::list as runner_configs_list, runners::list as runners_list};
 use rivet_api_util::{Method, request_remote_datacenter};
-use rivet_types::runner_configs::{RunnerConfig, RunnerConfigKind};
+use rivet_types::runner_configs::RunnerConfigKind;
 use serde::de::DeserializeOwned;
 
 #[derive(Debug, Clone, Serialize, Deserialize)]

packages/services/pegboard/src/ops/runner/find_dc_with_runner.rs

@@ -6,9 +6,8 @@ use gas::prelude::*;
 use rivet_metrics::KeyValue;
 use rivet_runner_protocol as protocol;
 use rivet_types::{
-	actors::CrashPolicy,
-	keys::namespace::runner_config::RunnerConfigVariant,
-	runner_configs::{RunnerConfig, RunnerConfigKind},
+	actors::CrashPolicy, keys::namespace::runner_config::RunnerConfigVariant,
+	runner_configs::RunnerConfigKind,
 };
 use std::time::Instant;
 use universaldb::options::{ConflictRangeType, MutationType, StreamingMode};