diff --git a/rivetkit-rust/packages/rivetkit-core/src/registry/mod.rs b/rivetkit-rust/packages/rivetkit-core/src/registry/mod.rs index 7bb68a3bfe..8c3384d36f 100644 --- a/rivetkit-rust/packages/rivetkit-core/src/registry/mod.rs +++ b/rivetkit-rust/packages/rivetkit-core/src/registry/mod.rs @@ -233,6 +233,7 @@ pub struct ServeConfig { pub engine_port: Option, 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, pub serverless_package_version: String, @@ -608,6 +609,8 @@ impl CoreRegistry { #[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(), }); diff --git a/rivetkit-rust/packages/rivetkit-core/src/registry/runner_config.rs b/rivetkit-rust/packages/rivetkit-core/src/registry/runner_config.rs index efb5b55057..8b7575fcdf 100644 --- a/rivetkit-rust/packages/rivetkit-core/src/registry/runner_config.rs +++ b/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, +} + 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 { + 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::() + .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, diff --git a/rivetkit-rust/packages/rivetkit-core/tests/runner_config.rs b/rivetkit-rust/packages/rivetkit-core/tests/runner_config.rs index b9a13b3587..8eb167d120 100644 --- a/rivetkit-rust/packages/rivetkit-core/tests/runner_config.rs +++ b/rivetkit-rust/packages/rivetkit-core/tests/runner_config.rs @@ -11,13 +11,18 @@ 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 @@ async fn fake_engine_handler( .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 fake_engine_handler( async fn start_fake_engine( failed_upserts: usize, + envoy_conflicts_before_clear: usize, ) -> ( String, Arc, @@ -65,8 +84,10 @@ async fn start_fake_engine( ) { 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 @@ fn config(endpoint: String) -> ServeConfig { } } +#[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 @@ async fn retries_namespace_not_found_until_runner_config_is_ready() { #[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 @@ async fn stops_after_the_configured_retry_bound() { 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"); +} diff --git a/rivetkit-rust/packages/rivetkit-core/tests/serverless.rs b/rivetkit-rust/packages/rivetkit-core/tests/serverless.rs index a6f0d48757..36c1c52ee4 100644 --- a/rivetkit-rust/packages/rivetkit-core/tests/serverless.rs +++ b/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(), diff --git a/rivetkit-typescript/packages/rivetkit-napi/src/registry.rs b/rivetkit-typescript/packages/rivetkit-napi/src/registry.rs index 6820684eb0..7b97aaee97 100644 --- a/rivetkit-typescript/packages/rivetkit-napi/src/registry.rs +++ b/rivetkit-typescript/packages/rivetkit-napi/src/registry.rs @@ -33,6 +33,7 @@ pub struct JsServeConfig { pub engine_binary_path: Option, pub engine_host: Option, pub engine_port: Option, + pub reject_existing_envoy: bool, pub handle_inspector_http_in_runtime: Option, pub serverless_base_path: Option, 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), diff --git a/rivetkit-typescript/packages/rivetkit-wasm/src/lib.rs b/rivetkit-typescript/packages/rivetkit-wasm/src/lib.rs index 29c9d1db47..f2cdac3a4a 100644 --- a/rivetkit-typescript/packages/rivetkit-wasm/src/lib.rs +++ b/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, + pub reject_existing_envoy: bool, pub handle_inspector_http_in_runtime: Option, pub inspector_test_token: Option, pub serverless_base_path: Option, @@ -154,6 +155,7 @@ impl From 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, diff --git a/rivetkit-typescript/packages/rivetkit/src/registry/native.ts b/rivetkit-typescript/packages/rivetkit/src/registry/native.ts index c9fe937f74..65d0c87a89 100644 --- a/rivetkit-typescript/packages/rivetkit/src/registry/native.ts +++ b/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, diff --git a/rivetkit-typescript/packages/rivetkit/src/registry/runtime.ts b/rivetkit-typescript/packages/rivetkit/src/registry/runtime.ts index 6827db8a3c..05bac2b83e 100644 --- a/rivetkit-typescript/packages/rivetkit/src/registry/runtime.ts +++ b/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,