fix: move runner configs to epoxy

Author: MasterPtato

engine/artifacts/openapi.json

@@ -352,7 +352,6 @@ pub async fn get_epoxy_key_debug(
 		.await?;
 
 	let instances = local_value
-		.value
 		.map(|_| {
 			vec![EpoxyKeyInstance {
 				replica_id,
@@ -499,10 +498,8 @@ pub async fn get_epoxy_kv_local(
 		.await?;
 
 	Ok(GetEpoxyKvResponse {
-		exists: output.value.is_some(),
-		value: output
-			.value
-			.map(|v| base64::engine::general_purpose::STANDARD.encode(&v)),
+		exists: output.is_some(),
+		value: output.map(|v| base64::engine::general_purpose::STANDARD.encode(&v.value)),
 	})
 }
 

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

@@ -28,7 +28,7 @@ pub async fn list(ctx: ApiCtx, _path: ListPath, query: ListQuery) -> Result<List
 	]
 	.concat();
 
-	let runner_configs = if !runner_names.is_empty() {
+	let (runner_configs, cursor) = if !runner_names.is_empty() {
 		let runner_configs = ctx
 			.op(pegboard::ops::runner_config::get::Input {
 				runners: runner_names
@@ -42,7 +42,7 @@ pub async fn list(ctx: ApiCtx, _path: ListPath, query: ListQuery) -> Result<List
 		(
 			runner_configs
 				.into_iter()
-				.map(|c| (c.name, c.config))
+				.map(|c| (c.name, c.config, c.protocol_version))
 				.collect::<Vec<_>>(),
 			None,
 		)
@@ -76,48 +76,49 @@ pub async fn list(ctx: ApiCtx, _path: ListPath, query: ListQuery) -> Result<List
 
 		let cursor = runner_configs
 			.last()
-			.map(|(name, config)| format!("{}:{}", runner_config_variant(&config), name));
+			.map(|c| format!("{}:{}", runner_config_variant(&c.config), c.name));
 
-		(runner_configs, cursor)
+		(
+			runner_configs
+				.into_iter()
+				.map(|c| (c.name, c.config, c.protocol_version))
+				.collect::<Vec<_>>(),
+			cursor,
+		)
 	};
 
 	// Fetch pool errors
-	let runner_pool_errors: HashMap<String, _> = if !runner_configs.0.is_empty() {
+	let runner_pool_errors = if !runner_configs.is_empty() {
 		let runners = runner_configs
-			.0
 			.iter()
-			.map(|(name, _)| (namespace.namespace_id, name.clone()))
+			.map(|(name, _, _)| (namespace.namespace_id, name.clone()))
 			.collect();
 		ctx.op(pegboard::ops::runner_config::get_error::Input { runners })
 			.await?
 			.into_iter()
 			.map(|e| (e.runner_name, e.error))
-			.collect()
+			.collect::<HashMap<_, _>>()
 	} else {
 		HashMap::new()
 	};
 
-	// Build response with pool errors
-	let runner_configs_with_errors: HashMap<String, RunnerConfigResponse> = runner_configs
-		.0
-		.into_iter()
-		.map(|(name, config)| {
-			let runner_pool_error = runner_pool_errors.get(&name).cloned();
-			(
-				name,
-				RunnerConfigResponse {
-					config,
-					runner_pool_error,
-				},
-			)
-		})
-		.collect();
-
 	Ok(ListResponse {
-		runner_configs: runner_configs_with_errors,
-		pagination: Pagination {
-			cursor: runner_configs.1,
-		},
+		// Build response with pool errors
+		runner_configs: runner_configs
+			.into_iter()
+			.map(|(name, config, protocol_version)| {
+				let runner_pool_error = runner_pool_errors.get(&name).cloned();
+				(
+					name,
+					RunnerConfigResponse {
+						config,
+						runner_pool_error,
+						protocol_version,
+					},
+				)
+			})
+			.collect(),
+		pagination: Pagination { cursor },
 	})
 }
 

engine/packages/api-peer/src/runner_configs.rs

@@ -58,8 +58,6 @@ async fn upsert_inner(
 ) -> Result<UpsertResponse> {
 	ctx.auth().await?;
 
-	tracing::debug!(runner_name = ?path.runner_name, datacenters_count = body.datacenters.len(), "starting upsert");
-
 	// Resolve namespace
 	let namespace = ctx
 		.op(namespace::ops::resolve_for_name_global::Input {
@@ -68,24 +66,6 @@ async fn upsert_inner(
 		.await?
 		.ok_or_else(|| namespace::errors::Namespace::NotFound.build())?;
 
-	// Store serverless config before processing (since we'll remove from body.datacenters)
-	let serverless_config = body
-		.datacenters
-		.iter()
-		.filter_map(|(_dc_name, runner_config)| {
-			if let rivet_api_types::namespaces::runner_configs::RunnerConfigKind::Serverless {
-				url,
-				headers,
-				..
-			} = &runner_config.kind
-			{
-				Some((url.clone(), headers.clone().unwrap_or_default()))
-			} else {
-				None
-			}
-		})
-		.next();
-
 	let dcs = ctx
 		.config()
 		.topology()
@@ -162,35 +142,6 @@ async fn upsert_inner(
 		.into_iter()
 		.any(|endpoint_config_changed| endpoint_config_changed);
 
-	// Update runner metadata
-	//
-	// This allows us to populate the actor names immediately upon configuring a serverless runner
-	if let Some((url, metadata_headers)) = serverless_config {
-		if any_endpoint_config_changed {
-			tracing::debug!("endpoint config changed, refreshing metadata");
-			if let Err(err) = utils::refresh_runner_config_metadata(
-				ctx.clone(),
-				namespace.namespace_id,
-				path.runner_name.clone(),
-				url,
-				metadata_headers,
-			)
-			.await
-			{
-				tracing::warn!(?err, runner_name=?path.runner_name, "failed to refresh runner config metadata");
-			}
-		} else {
-			tracing::debug!("endpoint config unchanged, skipping metadata refresh");
-		}
-	}
-
-	pegboard::utils::purge_runner_config_caches(
-		ctx.cache(),
-		namespace.namespace_id,
-		&path.runner_name,
-	)
-	.await?;
-
 	Ok(UpsertResponse {
 		endpoint_config_changed: any_endpoint_config_changed,
 	})

engine/packages/api-public/src/runner_configs/upsert.rs

@@ -25,9 +25,13 @@ pub enum RunnerConfigKind {
 		max_concurrent_actors: Option<u64>,
 		/// Seconds.
 		drain_grace_period: Option<u32>,
+		/// Deprecated.
 		slots_per_runner: u32,
+		/// Deprecated.
 		min_runners: Option<u32>,
+		/// Deprecated.
 		max_runners: u32,
+		/// Deprecated.
 		runners_margin: Option<u32>,
 		/// Milliseconds between metadata polling. If not set, uses the global default.
 		metadata_poll_interval: Option<u64>,

engine/packages/api-types/src/namespaces/runner_configs.rs

@@ -7,6 +7,7 @@ pub mod list;
 pub struct RunnerConfigResponse {
 	#[serde(flatten)]
 	pub config: rivet_types::runner_configs::RunnerConfig,
+	pub protocol_version: Option<u16>,
 	#[serde(default, skip_serializing_if = "Option::is_none")]
 	#[schema(value_type = Option<Object>, additional_properties = true)]
 	pub runner_pool_error: Option<rivet_types::actor::RunnerPoolError>,

engine/packages/api-types/src/runner_configs/mod.rs

@@ -80,6 +80,7 @@ pub async fn epoxy_kv_get_optimistic(ctx: &OperationCtx, input: &Input) -> Resul
 		&quorum_members,
 		utils::QuorumType::Any,
 		|replica_id| {
+			let ctx = ctx.clone();
 			let config = config.clone();
 			let key = input.key.clone();
 			let from_replica_id = input.replica_id;
@@ -109,7 +110,7 @@ pub async fn epoxy_kv_get_optimistic(ctx: &OperationCtx, input: &Input) -> Resul
 	.await?;
 
 	// Should only have 1 response
-	if let Some(value) = responses.first().and_then(|r| r.response) {
+	if let Some(value) = responses.into_iter().flatten().next() {
 		let value = CommittedValue {
 			value: value.value,
 			version: value.version,