fix(rivetkit): reject conflicting local envoy connections

rivetkit-rust/packages/rivetkit-core/src/registry/mod.rs

@@ -233,6 +233,7 @@
 	pub engine_port: Option<u16>,
 	pub engine_spawn: EngineSpawnMode,
 	pub engine_auto_download: bool,
+	pub reject_existing_envoy: bool,
 	pub handle_inspector_http_in_runtime: bool,
 	pub serverless_base_path: Option<String>,
 	pub serverless_package_version: String,
@@ -338,7 +339,7 @@
 		.map(|ip| ip.is_loopback() || ip.is_unspecified())
 		.unwrap_or(false))
 }
 
 #[cfg(test)]
 mod engine_spawn_tests {
 	use super::{EngineSpawnMode, should_manage_engine};
@@ -608,6 +609,8 @@
 
 		#[cfg(feature = "native-runtime")]
 		runner_config::ensure_local_normal_runner_config(&config).await?;
+		#[cfg(feature = "native-runtime")]
+		runner_config::ensure_development_envoy_available(&config).await?;
 		let callbacks = Arc::new(RegistryCallbacks {
 			dispatcher: dispatcher.clone(),
 		});

rivetkit-rust/packages/rivetkit-core/src/registry/runner_config.rs

@@ -12,6 +12,8 @@ use super::ServeConfig;
 
 const RUNNER_CONFIG_MAX_ATTEMPTS: usize = 60;
 const RUNNER_CONFIG_RETRY_DELAY: Duration = Duration::from_millis(500);
+const DEVELOPMENT_ENVOY_CHECK_MAX_ATTEMPTS: usize = 5;
+const DEVELOPMENT_ENVOY_CHECK_RETRY_DELAY: Duration = Duration::from_millis(100);
 
 #[derive(Debug, Deserialize)]
 struct DatacentersResponse {
@@ -23,6 +25,11 @@ struct Datacenter {
 	name: String,
 }
 
+#[derive(Debug, Deserialize)]
+struct EnvoysResponse {
+	envoys: Vec<serde_json::Value>,
+}
+
 pub(super) async fn ensure_local_normal_runner_config(config: &ServeConfig) -> Result<()> {
 	if !is_local_engine_endpoint(&config.endpoint) {
 		return Ok(());
@@ -40,6 +47,71 @@ pub(super) async fn ensure_local_normal_runner_config(config: &ServeConfig) -> R
 	.await
 }
 
+pub(super) async fn ensure_development_envoy_available(config: &ServeConfig) -> Result<()> {
+	if !config.reject_existing_envoy {
+		return Ok(());
+	}
+
+	let client = Client::builder()
+		.build()
+		.context("build reqwest client for development Envoy check")?;
+	ensure_development_envoy_available_with_retry(
+		&client,
+		config,
+		DEVELOPMENT_ENVOY_CHECK_MAX_ATTEMPTS,
+		DEVELOPMENT_ENVOY_CHECK_RETRY_DELAY,
+	)
+	.await
+}
+
+async fn ensure_development_envoy_available_with_retry(
+	client: &Client,
+	config: &ServeConfig,
+	max_attempts: usize,
+	retry_delay: Duration,
+) -> Result<()> {
+	let max_attempts = max_attempts.max(1);
+	for attempt in 1..=max_attempts {
+		if !envoy_is_connected(client, config).await? {
+			return Ok(());
+		}
+		if attempt < max_attempts {
+			sleep(retry_delay).await;
+		}
+	}
+
+	anyhow::bail!(
+		"another Envoy is already connected to namespace `{}` and pool `{}` at {}. Stop the other project, set RIVET_NAMESPACE to a different namespace, or set RIVET_RUN_ENGINE_PORT to a different Engine port. See https://rivet.dev/docs/general/environment-variables/",
+		config.namespace,
+		config.pool_name,
+		config.endpoint,
+	)
+}
+
+async fn envoy_is_connected(client: &Client, config: &ServeConfig) -> Result<bool> {
+	let mut url = engine_api_url(&config.endpoint, &["envoys"], &config.namespace)?;
+	url.query_pairs_mut()
+		.append_pair("name", &config.pool_name)
+		.append_pair("limit", "1");
+	let response = apply_auth(client.get(url), config)
+		.send()
+		.await
+		.context("list local Envoys")?;
+	let status = response.status();
+	if !status.is_success() {
+		let body = response
+			.text()
+			.await
+			.context("read failed local Envoy list response")?;
+		anyhow::bail!("failed to list local Envoys: {status} {body}");
+	}
+	let response = response
+		.json::<EnvoysResponse>()
+		.await
+		.context("decode local Envoy list response")?;
+	Ok(!response.envoys.is_empty())
+}
+
 async fn ensure_local_normal_runner_config_with_retry(
 	client: &Client,
 	config: &ServeConfig,

rivetkit-rust/packages/rivetkit-core/tests/runner_config.rs

@@ -9,15 +9,20 @@
 use axum::response::Response;
 use reqwest::Client;
 use tokio::net::TcpListener;
 use tokio::sync::oneshot;
 
-use super::{ServeConfig, ensure_local_normal_runner_config_with_retry};
+use super::{
+	ServeConfig, ensure_development_envoy_available,
+	ensure_development_envoy_available_with_retry, ensure_local_normal_runner_config_with_retry,
+};
 
 #[derive(Debug)]
 struct FakeEngineState {
 	failed_upserts: usize,
+	envoy_conflicts_before_clear: usize,
 	datacenter_requests: AtomicUsize,
 	upsert_requests: AtomicUsize,
+	envoy_requests: AtomicUsize,
 }
 
 async fn fake_engine_handler(
@@ -48,6 +53,19 @@
 					.expect("build upsert response")
 			}
 		}
+		(&Method::GET, "/envoys") => {
+			let attempt = state.envoy_requests.fetch_add(1, Ordering::SeqCst);
+			let body = if attempt < state.envoy_conflicts_before_clear {
+				r#"{"envoys":[{"envoyKey":"other"}],"pagination":{"cursor":null}}"#
+			} else {
+				r#"{"envoys":[],"pagination":{"cursor":null}}"#
+			};
+			Response::builder()
+				.status(StatusCode::OK)
+				.header("content-type", "application/json")
+				.body(Body::from(body))
+				.expect("build Envoy list response")
+		}
 		_ => Response::builder()
 			.status(StatusCode::NOT_FOUND)
 			.body(Body::empty())
@@ -57,6 +75,7 @@
 
 async fn start_fake_engine(
 	failed_upserts: usize,
+	envoy_conflicts_before_clear: usize,
 ) -> (
 	String,
 	Arc<FakeEngineState>,
@@ -65,8 +84,10 @@
 ) {
 	let state = Arc::new(FakeEngineState {
 		failed_upserts,
+		envoy_conflicts_before_clear,
 		datacenter_requests: AtomicUsize::new(0),
 		upsert_requests: AtomicUsize::new(0),
+		envoy_requests: AtomicUsize::new(0),
 	});
 	let app = Router::new()
 		.fallback(fake_engine_handler)
@@ -98,9 +119,16 @@
 	}
 }
 
+#[tokio::test]
+async fn skips_envoy_conflict_check_when_rejection_is_disabled() {
+	ensure_development_envoy_available(&config("not a URL".to_owned()))
+		.await
+		.expect("disabled conflict check should not inspect the endpoint");
+}
+
 #[tokio::test]
 async fn retries_namespace_not_found_until_runner_config_is_ready() {
-	let (endpoint, state, shutdown, server) = start_fake_engine(1).await;
+	let (endpoint, state, shutdown, server) = start_fake_engine(1, 0).await;
 
 	ensure_local_normal_runner_config_with_retry(
 		&Client::new(),
@@ -119,7 +147,7 @@
 
 #[tokio::test]
 async fn stops_after_the_configured_retry_bound() {
-	let (endpoint, state, shutdown, server) = start_fake_engine(usize::MAX).await;
+	let (endpoint, state, shutdown, server) = start_fake_engine(usize::MAX, 0).await;
 
 	let error = ensure_local_normal_runner_config_with_retry(
 		&Client::new(),
@@ -140,3 +168,37 @@
 	let _ = shutdown.send(());
 	server.await.expect("join fake Engine");
 }
+
+#[tokio::test]
+async fn waits_for_hot_reload_envoy_to_disconnect() {
+	let (endpoint, state, shutdown, server) = start_fake_engine(0, 1).await;
+	let config = config(endpoint);
+
+	ensure_development_envoy_available_with_retry(&Client::new(), &config, 2, Duration::ZERO)
+		.await
+		.expect("old Envoy should clear during the handoff");
+
+	assert_eq!(state.envoy_requests.load(Ordering::SeqCst), 2);
+	let _ = shutdown.send(());
+	server.await.expect("join fake Engine");
+}
+
+#[tokio::test]
+async fn rejects_a_conflicting_development_envoy_with_actionable_options() {
+	let (endpoint, state, shutdown, server) = start_fake_engine(0, usize::MAX).await;
+	let config = config(endpoint);
+
+	let error =
+		ensure_development_envoy_available_with_retry(&Client::new(), &config, 2, Duration::ZERO)
+			.await
+			.expect_err("connected Envoy should conflict");
+	let message = error.to_string();
+
+	assert!(message.contains("another Envoy is already connected"));
+	assert!(message.contains("RIVET_NAMESPACE"));
+	assert!(message.contains("RIVET_RUN_ENGINE_PORT"));
+	assert!(message.contains("https://rivet.dev/docs/general/environment-variables/"));
+	assert_eq!(state.envoy_requests.load(Ordering::SeqCst), 2);
+	let _ = shutdown.send(());
+	server.await.expect("join fake Engine");
+}

rivetkit-rust/packages/rivetkit-core/tests/serverless.rs

@@ -350,6 +350,7 @@ mod moved_tests {
 			engine_port: None,
 			engine_spawn: EngineSpawnMode::Never,
 			engine_auto_download: false,
+			reject_existing_envoy: false,
 			handle_inspector_http_in_runtime: true,
 			serverless_base_path: Some("/api/rivet".to_owned()),
 			serverless_package_version: "test-version".to_owned(),

rivetkit-typescript/packages/rivetkit-napi/src/registry.rs

@@ -33,6 +33,7 @@ pub struct JsServeConfig {
 	pub engine_binary_path: Option<String>,
 	pub engine_host: Option<String>,
 	pub engine_port: Option<u16>,
+	pub reject_existing_envoy: bool,
 	pub handle_inspector_http_in_runtime: Option<bool>,
 	pub serverless_base_path: Option<String>,
 	pub serverless_package_version: String,
@@ -816,6 +817,7 @@ fn serve_config_from_js(
 		engine_port: config.engine_port,
 		engine_spawn: EngineSpawnMode::Auto,
 		engine_auto_download: false,
+		reject_existing_envoy: config.reject_existing_envoy,
 		handle_inspector_http_in_runtime: config
 			.handle_inspector_http_in_runtime
 			.unwrap_or(default_handle_inspector_http_in_runtime),

rivetkit-typescript/packages/rivetkit-wasm/src/lib.rs

@@ -119,6 +119,7 @@ pub struct WasmServeConfig {
 	pub namespace: String,
 	pub pool_name: String,
 	pub engine_binary_path: Option<String>,
+	pub reject_existing_envoy: bool,
 	pub handle_inspector_http_in_runtime: Option<bool>,
 	pub inspector_test_token: Option<String>,
 	pub serverless_base_path: Option<String>,
@@ -154,6 +155,7 @@ impl From<WasmServeConfig> for ServeConfig {
 			engine_port: None,
 			engine_spawn: EngineSpawnMode::Never,
 			engine_auto_download: false,
+			reject_existing_envoy: config.reject_existing_envoy,
 			handle_inspector_http_in_runtime: config
 				.handle_inspector_http_in_runtime
 				.unwrap_or(false),
@@ -3119,6 +3121,7 @@ mod tests {
 			namespace: "default".to_owned(),
 			pool_name: "default".to_owned(),
 			engine_binary_path,
+			reject_existing_envoy: false,
 			handle_inspector_http_in_runtime: None,
 			inspector_test_token: None,
 			serverless_base_path: None,

rivetkit-typescript/packages/rivetkit/src/registry/native.ts

@@ -65,6 +65,7 @@ import type {
 } from "@/registry/config";
 import { decodeCborCompat, encodeCborCompat } from "@/serde";
 import { getEnvUniversal, VERSION } from "@/utils";
+import { isDev } from "@/utils/env-vars";
 import {
 	getNodeFsSync,
 	getNodePath,
@@ -5028,6 +5029,7 @@ export async function buildServeConfig(
 		token: config.token,
 		namespace: config.namespace,
 		poolName: config.envoy.poolName,
+		rejectExistingEnvoy: isDev(),
 		handleInspectorHttpInRuntime: true,
 		serverlessBasePath: config.serverless.basePath,
 		serverlessPackageVersion: VERSION,

rivetkit-typescript/packages/rivetkit/src/registry/runtime.ts

@@ -1,5 +1,6 @@
 import type { SqliteNativeMetrics } from "@/common/database/config";
 import { stringifyError } from "@/common/utils";
+import { isDev } from "@/utils/env-vars";
 import type { RegistryConfig } from "./config";
 import { logger } from "./log";
 
@@ -331,6 +332,7 @@ export interface RuntimeServeConfig {
 	engineBinaryPath?: string;
 	engineHost?: string;
 	enginePort?: number;
+	rejectExistingEnvoy: boolean;
 	handleInspectorHttpInRuntime?: boolean;
 	inspectorTestToken?: string;
 	serverlessBasePath?: string;
@@ -768,6 +770,7 @@ export async function buildServeConfig(
 		token: config.token,
 		namespace: config.namespace,
 		poolName: config.envoy.poolName,
+		rejectExistingEnvoy: isDev(),
 		handleInspectorHttpInRuntime: true,
 		serverlessBasePath: config.serverless.basePath,
 		serverlessPackageVersion: version,